test: add anaconda-iso build tests with signed containers

Add anaconda-iso iso build tests with signed containers.
The rest of the images can be also added to the test once
[1] and [2] are merged

[1] osbuild/images#990
[2] osbuild/osbuild#1906

Signed-off-by: Miguel Martín <[email protected]>

Author: mmartinv

.github/workflows/tests.yml

@@ -93,7 +93,7 @@ jobs:
         echo "deb $sources_url/ /" | sudo tee /etc/apt/sources.list.d/devel-kubic-libcontainers-unstable.list
         curl -fsSL $key_url | gpg --dearmor | sudo tee /etc/apt/trusted.gpg.d/devel_kubic_libcontainers_unstable.gpg > /dev/null
         sudo apt update
-        sudo apt install -y podman
+        sudo apt install -y podman skopeo
     - name: Install python test deps
       run: |
         # make sure test deps are available for root

test/test_build.py

@@ -30,6 +30,7 @@
 if platform.system() == "Linux" and platform.machine() == "x86_64" and not testutil.has_x86_64_v3_cpu():
     pytest.skip("need x86_64-v3 capable CPU", allow_module_level=True)
 
+registry_port = testutil.get_free_port()
 
 class ImageBuildResult(NamedTuple):
     img_type: str
@@ -113,11 +114,20 @@ def build_images(shared_tmpdir, build_container, request, force_aws_upload):
     password = "password"
     kargs = "systemd.journald.forward_to_console=1"
 
+    container_ref = tc.container_ref
+
+    # We cannot use localhost as we need to access the registry from both
+    # the host system and the bootc-image-builder container.
+    default_ip = testutil.get_ip_from_default_route()
+    local_registry = f"{default_ip}:{registry_port}"
+    if tc.sign:
+        container_ref = testutil.get_signed_container_ref(local_registry, tc.container_ref)
+
     # params can be long and the qmp socket (that has a limit of 100ish
     # AF_UNIX) is derived from the path
     # hash the container_ref+target_arch, but exclude the image_type so that the output path is shared between calls to
     # different image type combinations
-    output_path = shared_tmpdir / format(abs(hash(tc.container_ref + str(tc.target_arch))), "x")
+    output_path = shared_tmpdir / format(abs(hash(container_ref + str(tc.target_arch))), "x")
     output_path.mkdir(exist_ok=True)
 
     # make sure that the test store exists, because podman refuses to start if the source directory for a volume
@@ -164,7 +174,7 @@ def build_images(shared_tmpdir, build_container, request, force_aws_upload):
             bib_output = bib_output_path.read_text(encoding="utf8")
             results.append(ImageBuildResult(
                 image_type, generated_img, tc.target_arch, tc.osinfo_template,
-                tc.container_ref, tc.rootfs, username, password,
+                container_ref, tc.rootfs, username, password,
                 ssh_keyfile_private_path, kargs, bib_output, journal_output))
 
     # generate new keyfile
@@ -257,15 +267,35 @@ def build_images(shared_tmpdir, build_container, request, force_aws_upload):
         if tc.local:
             cmd.extend(["-v", "/var/lib/containers/storage:/var/lib/containers/storage"])
 
+        if tc.sign:
+            registry_conf = testutil.RegistryConf()
+            registry_conf.local_registry = local_registry
+            registry_conf.base_dir = output_path
+            gpg_conf = testutil.GPGConf()
+            gpg_conf.base_dir = output_path
+            testutil.sign_container_image(gpg_conf, registry_conf, tc.container_ref)
+            policy_file = registry_conf.policy_file
+            lookaside_conf_file = registry_conf.lookaside_conf_file
+            sigstore_dir = registry_conf.sigstore_dir
+            pub_key_file = gpg_conf.pub_key_file
+            signed_image_args = [
+                "-v", f"{policy_file}:/etc/containers/policy.json",
+                "-v", f"{lookaside_conf_file}:/etc/containers/registries.d/bib-tests.yaml",
+                "-v", f"{sigstore_dir}:{sigstore_dir}",
+                "-v", f"{pub_key_file}:{pub_key_file}",
+            ]
+            cmd.extend(signed_image_args)
+
         cmd.extend([
             *creds_args,
             build_container,
-            tc.container_ref,
+            container_ref,
             *types_arg,
             *upload_args,
             *target_arch_args,
             *tc.bib_rootfs_args(),
             "--local" if tc.local else "--local=false",
+            "--tls-verify=false" if tc.sign else "--tls-verify=true"
         ])
 
         # print the build command for easier tracing
@@ -299,7 +329,7 @@ def del_ami():
     for image_type in image_types:
         results.append(ImageBuildResult(
             image_type, artifact[image_type], tc.target_arch, tc.osinfo_template,
-            tc.container_ref, tc.rootfs, username, password,
+            container_ref, tc.rootfs, username, password,
             ssh_keyfile_private_path, kargs, bib_output, journal_output, metadata))
     yield results
 
@@ -316,7 +346,7 @@ def del_ami():
                 img.unlink()
         else:
             print("does not exist")
-    subprocess.run(["podman", "rmi", tc.container_ref], check=False)
+    subprocess.run(["podman", "rmi", container_ref], check=False)
     return
 
 

test/testcases.py

@@ -23,6 +23,8 @@ class TestCase:
     # rootfs to use (e.g. ext4), some containers like fedora do not
     # have a default rootfs. If unset the container default is used.
     rootfs: str = ""
+    # Sign the container_ref and use the new signed image instead of the original one
+    sign: bool = False
 
     def bib_rootfs_args(self):
         if self.rootfs:
@@ -31,7 +33,7 @@ def bib_rootfs_args(self):
 
     def __str__(self):
         return ",".join([
-            attr
+            f"{name}={attr}"
             for name, attr in inspect.getmembers(self)
             if not name.startswith("_") and not callable(attr) and attr
         ])
@@ -68,7 +70,11 @@ def gen_testcases(what):  # pylint: disable=too-many-return-statements
     if what == "ami-boot":
         return [TestCaseCentos(image="ami"), TestCaseFedora(image="ami")]
     if what == "anaconda-iso":
-        return [TestCaseCentos(image="anaconda-iso"), TestCaseFedora(image="anaconda-iso")]
+        return [
+            TestCaseFedora(image="anaconda-iso", sign=True),
+            TestCaseCentos(image="anaconda-iso"),
+            TestCaseFedora(image="anaconda-iso"),
+        ]
     if what == "qemu-boot":
         test_cases = [
             klass(image=img)

test/testutil.py

@@ -1,3 +1,5 @@
+import dataclasses
+import json
 import os
 import pathlib
 import platform
@@ -147,3 +149,225 @@ def create_filesystem_customizations(rootfs: str):
     "-v", "/var/lib/containers/storage:/var/lib/containers/storage",
     "--security-opt", "label=type:unconfined_t",
 ]
+
+
+def get_ip_from_default_route():
+    default_route = subprocess.run([
+        "ip",
+        "route",
+        "list",
+        "default"
+    ], check=True, capture_output=True, text=True).stdout
+    return default_route.split()[8]
+
+
+@dataclasses.dataclass
+class GPGConf:
+    _key_params_tmpl: str = """
+      %no-protection
+      Key-Type: RSA
+      Key-Length: {key_length}
+      Key-Usage: sign
+      Name-Real: Bootc Image Builder Tests
+      Name-Email: {email}
+      Expire-Date: 0
+    """
+    _base_dir: str = "/tmp/bib-tests"
+    _email: str = "[email protected]"
+    _key_length: str = "3072"
+    home_dir: str = f"{_base_dir}/.gnupg"
+    pub_key_file: str = f"{_base_dir}/GPG-KEY-bib-tests"
+    key_params: str = _key_params_tmpl.format(key_length=_key_length, email=_email)
+
+    @property
+    def base_dir(self) -> str:
+        return self._base_dir
+
+    @base_dir.setter
+    def base_dir(self, base_dir: str) -> None:
+        self._base_dir = base_dir
+        self.home_dir = f"{base_dir}/.gnupg"
+        self.pub_key_file = f"{base_dir}/GPG-KEY-bib-tests"
+
+    @property
+    def email(self) -> str:
+        return self._email
+
+    @email.setter
+    def email(self, email: str) -> None:
+        self._email = email
+        self.key_params = self._key_params_tmpl.format(
+            email=self._email,
+            key_length=self._key_length
+        )
+
+    @property
+    def key_length(self) -> str:
+        return self._key_length
+
+    @key_length.setter
+    def key_length(self, length: str) -> None:
+        self._key_length = length
+        self._key_params = self._key_params_tmpl.format(
+            email=self._email,
+            key_length=self._key_length
+        )
+
+def gpg_gen_key(gpg_conf: GPGConf):
+    if os.path.exists(gpg_conf.home_dir):
+        return
+
+    os.makedirs(gpg_conf.home_dir, mode=0o700, exist_ok=False)
+
+    subprocess.run(
+        ["gpg", "--gen-key", "--batch"],
+        check=True, env={"GNUPGHOME": gpg_conf.home_dir},
+        input=gpg_conf.key_params,
+        text=True)
+
+    subprocess.run(
+        ["gpg", "--output", gpg_conf.pub_key_file,
+         "--armor", "--export", gpg_conf.email],
+        check=True, env={"GNUPGHOME": gpg_conf.home_dir})
+
+
+@dataclasses.dataclass
+class RegistryConf():
+    _lookaside_conf_tmpl: str = """
+    docker:
+      {local_registry}:
+        lookaside: file:///{sigstore_dir}
+    """
+    _local_registry: str = "localhost:5000"
+    _base_dir: str = "/tmp/bib-tests"
+    _sigstore_dir: str = f"{_base_dir}/sigstore"
+    _registries_d_dir: str = f"{_base_dir}/registries.d"
+    policy_file: str = f"{_base_dir}/policy.json"
+    lookaside_conf_file: str = f"{_registries_d_dir}/lookaside.yml"
+    lookaside_conf: str = _lookaside_conf_tmpl.format(
+        local_registry=_local_registry,
+        sigstore_dir=_sigstore_dir
+    )
+
+    @property
+    def local_registry(self) -> str:
+        return self._local_registry
+
+    @local_registry.setter
+    def local_registry(self, registry: str) -> None:
+        self._local_registry = registry
+        self.lookaside_conf = self._lookaside_conf_tmpl.format(
+            local_registry=self._local_registry,
+            sigstore_dir=self._sigstore_dir
+        )
+
+    @property
+    def base_dir(self) -> str:
+        return self._base_dir
+
+    @base_dir.setter
+    def base_dir(self, base_dir: str) -> None:
+        self._base_dir = base_dir
+        self._sigstore_dir = f"{base_dir}/sigstore"
+        self.policy_file = f"{base_dir}/policy.json"
+        self._registries_d_dir = f"{base_dir}/registries.d"
+        self.lookaside_conf_file = f"{self._registries_d_dir}/lookaside.yaml"
+        self.lookaside_conf = self._lookaside_conf_tmpl.format(
+            local_registry=self._local_registry,
+            sigstore_dir=self._sigstore_dir
+        )
+
+    @property
+    def sigstore_dir(self) -> str:
+        return self._sigstore_dir
+
+    @sigstore_dir.setter
+    def sigstore_dir(self, sigstore_dir: str) -> None:
+        self._sigstore_dir = sigstore_dir
+        self.lookaside_conf = self._lookaside_conf_tmpl.format(
+            local_registry=self._local_registry,
+            sigstore_dir=self._sigstore_dir
+        )
+
+    @property
+    def registries_d_dir(self) -> str:
+        return self._registries_d_dir
+
+    @registries_d_dir.setter
+    def registries_d_dir(self, dir: str) -> None:
+        self._registries_d_dir= dir
+        self.lookaside_conf_file = f"{self._registries_d_dir}/lookaside.yaml"
+
+def ensure_registry(registry_conf: RegistryConf):
+    registry_container_name = subprocess.run([
+        "podman", "ps", "-a", "--filter", "name=registry", "--format", "{{.Names}}"
+    ], check=True, capture_output=True, text=True).stdout.strip()
+
+    registry_port = registry_conf.local_registry.split(":")[1]
+    if registry_container_name != "registry":
+        subprocess.run([
+            "podman", "run", "-d",
+            "-p", f"{registry_port}:5000",
+            "--restart", "always",
+            "--name", "registry",
+            "registry:2"
+        ], check=True)
+
+    registry_container_state = subprocess.run([
+        "podman", "ps", "-a", "--filter", "name=registry", "--format", "{{.State}}"
+    ], check=True, capture_output=True, text=True).stdout.strip()
+
+    if registry_container_state in ("paused", "exited"):
+        subprocess.run([
+            "podman", "start", "registry"
+        ], check=True)
+
+
+def get_signed_container_ref(local_registry: str, container_ref: str):
+    container_ref_path = container_ref[container_ref.index('/'):]
+    return f"{local_registry}{container_ref_path}"
+
+
+def sign_container_image(gpg_conf: GPGConf, registry_conf: RegistryConf, container_ref):
+    gpg_gen_key(gpg_conf)
+    ensure_registry(registry_conf)
+    local_registry = registry_conf.local_registry
+    policy_file = registry_conf.policy_file
+    registries_d_dir = registry_conf.registries_d_dir
+    lookaside_conf_file = registry_conf.lookaside_conf_file
+    lookaside_conf = registry_conf.lookaside_conf
+    pub_key_file = gpg_conf.pub_key_file
+    home_dir = gpg_conf.home_dir
+    email = gpg_conf.email
+    registry_policy = {
+        "default": [{"type": "insecureAcceptAnything"}],
+        "transports": {
+            "docker": {
+                f"{local_registry}": [
+                    {
+                        "type": "signedBy",
+                        "keyType": "GPGKeys",
+                        "keyPath": f"{pub_key_file}"
+                    }
+                ]
+            },
+            "docker-daemon": {
+                "": [{"type": "insecureAcceptAnything"}]
+            }
+        }
+    }
+    with open(policy_file, mode="w", encoding="utf-8") as f:
+        f.write(json.dumps(registry_policy))
+
+    os.makedirs(os.path.dirname(lookaside_conf_file), mode=0o700, exist_ok=False)
+    with open(lookaside_conf_file, mode="w", encoding="utf-8") as f:
+        f.write(lookaside_conf)
+
+    signed_container_ref = get_signed_container_ref(local_registry, container_ref)
+    subprocess.run([
+        "skopeo", "--registries.d", f"{registries_d_dir}",
+        "copy", "--dest-tls-verify=false", "--remove-signatures",
+        "--sign-by", email,
+        f"docker://{container_ref}",
+        f"docker://{signed_container_ref}",
+    ], check=True, env={"GNUPGHOME": home_dir})