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]>

test/test_build.py

@@ -113,11 +113,20 @@ def build_images(shared_tmpdir, build_container, request, force_aws_upload):
     password = "password"
     kargs = "systemd.journald.forward_to_console=1"
 
+    default_ip = testutil.get_ip_from_default_route()
+
+    gpg_config = testutil.GPGConfig()
+    registry_config = testutil.RegistryConfig(local_registry=f"{default_ip}:5000")
+    container_ref = tc.container_ref
+
+    if tc.sign:
+        container_ref = testutil.sign_container_image(gpg_config, registry_config, 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 +173,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 +266,28 @@ 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:
+            lookaside_config = registry_config.lookaside_config
+            gpg_pub_key = gpg_config.pub_key
+            sigstore_dir = registry_config.sigstore_dir
+            signed_image_args = [
+                "-v", "/etc/containers/policy.json:/etc/containers/policy.json",
+                "-v", f"{gpg_pub_key}:{gpg_pub_key}",
+                "-v", f"{lookaside_config}:{lookaside_config}",
+                "-v", f"{sigstore_dir}:{sigstore_dir}",
+            ]
+            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 +321,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 +338,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,12 @@ 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 = ""
+    # osinfo_template is a string template describing the OS detected by
+    # 'osinfo-detect'. It can contain '{arch}' that will be replaced with the
+    # actual container image arch
+    osinfo_template: 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 +37,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 +74,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 [
+            klass(image="anaconda-iso", sign=sign, local=False)
+            for klass in (TestCaseCentos, TestCaseFedora)
+            for sign in [False, True]
+        ]
     if what == "qemu-boot":
         test_cases = [
             klass(image=img)

test/testutil.py

@@ -1,9 +1,11 @@
+import dataclasses
 import os
 import pathlib
 import platform
 import shutil
 import socket
 import subprocess
+import tempfile
 import time
 
 import boto3
@@ -147,3 +149,97 @@ 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).stdout
+    return default_route.split()[8].decode("utf-8")
+
+
+@dataclasses.dataclass
+class GPGConfig():
+    email: str = "[email protected]"
+    pub_key: str = "/etc/pki/rpm-gpg/RPM-GPG-KEY-booc-image-builder"
+    passphrase: str = "redhat"
+
+
+@dataclasses.dataclass
+class RegistryConfig():
+    local_registry: str = "localhost:5000"
+    sigstore_dir: str = "/var/lib/containers/sigstore"
+    lookaside_config: str = "/etc/containers/registries.d/bib.yaml"
+
+
+def sign_container_image(gpg_config: GPGConfig, registry_config: RegistryConfig, container_ref):
+    if not os.path.exists(gpg_config.pub_key):
+        subprocess.run([
+            "gpg",
+            "--quick-gen-key",
+            "--batch",
+            "--passphrase", gpg_config.passphrase,
+            gpg_config.email
+        ], check=True)
+        subprocess.run([
+            "gpg",
+            "--output", gpg_config.pub_key,
+            "--armor",
+            "--export",
+            gpg_config.email
+        ], check=True)
+        subprocess.run([
+            "podman", "image", "trust", "set",
+            "--pubkeysfile", gpg_config.pub_key,
+            "--type", "signedBy",
+            registry_config.local_registry
+        ], check=True)
+
+    registry_lookaside_config = f"""docker:
+      {registry_config.local_registry}:
+        lookaside: file:///{registry_config.sigstore_dir}
+    """
+    with open(registry_config.lookaside_config, mode="w", encoding="utf-8") as f:
+        f.write(registry_lookaside_config)
+
+    registry_container_name = subprocess.run([
+        "podman", "ps", "-a", "--filter", "name=registry", "--format", "{{.Names}}"
+    ], check=True, capture_output=True).stdout.decode("utf-8").strip()
+
+    if registry_container_name != "registry":
+        subprocess.run([
+            "podman", "run", "-d",
+            "-p", "5000: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).stdout.decode("utf-8").strip()
+
+    if registry_container_state in ("paused", "exited"):
+        subprocess.run([
+            "podman", "start", "registry"
+        ], check=True, )
+
+    container_ref_path = container_ref[container_ref.index('/'):]
+    signed_container_ref = f"{registry_config.local_registry}{container_ref_path}"
+    with tempfile.NamedTemporaryFile(mode="w") as f:
+        f.write(gpg_config.passphrase)
+        f.flush()
+        subprocess.run([
+            "skopeo", "copy",
+            "--dest-tls-verify=false",
+            "--remove-signatures",
+            "--sign-by", gpg_config.email,
+            "--sign-passphrase-file", f.name,
+            f"docker://{container_ref}",
+            f"docker://{signed_container_ref}",
+        ], check=True)
+
+    return signed_container_ref