test: update tests to use ssh keys instead of passwords

With the switch to bootc we need to adjust the testing. We inject
a root ssh key now and just use that for login.

test/test_build.py

@@ -1,4 +1,3 @@
-import json
 import os
 import pathlib
 import platform
@@ -37,7 +36,7 @@ class ImageBuildResult(NamedTuple):
     img_path: str
     img_arch: str
     username: str
-    password: str
+    ssh_keyfile_private_path: str
     bib_output: str
     journal_output: str
     metadata: dict = {}
@@ -123,9 +122,6 @@ def build_images(shared_tmpdir, build_container, request, force_aws_upload):
     # split and check each one
     image_types = images.split("+")
 
-    username = "test"
-    password = "password"
-
     # 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
@@ -135,6 +131,10 @@ def build_images(shared_tmpdir, build_container, request, force_aws_upload):
 
     journal_log_path = output_path / "journal.log"
     bib_output_path = output_path / "bib-output.log"
+
+    ssh_keyfile_private_path = output_path / "ssh-keyfile"
+    ssh_keyfile_public_path = ssh_keyfile_private_path.with_suffix(".pub")
+
     artifact = {
         "qcow2": pathlib.Path(output_path) / "qcow2/disk.qcow2",
         "ami": pathlib.Path(output_path) / "image/disk.raw",
@@ -166,9 +166,22 @@ def build_images(shared_tmpdir, build_container, request, force_aws_upload):
             journal_output = journal_log_path.read_text(encoding="utf8")
             bib_output = bib_output_path.read_text(encoding="utf8")
             results.append(ImageBuildResult(
-                image_type, generated_img, target_arch, username, password,
+                image_type, generated_img, target_arch,
+                "root", ssh_keyfile_private_path,
                 bib_output, journal_output))
 
+    # generate new keyfile
+    if not ssh_keyfile_private_path.exists():
+        subprocess.run([
+            "ssh-keygen",
+            "-N", "",
+            # be very conservative with keys for paramiko
+            "-b", "2048",
+            "-t", "rsa",
+            "-f", os.fspath(ssh_keyfile_private_path),
+        ], check=True)
+    ssh_pubkey = ssh_keyfile_public_path.read_text(encoding="utf8")
+
     # Because we always build all image types, regardless of what was requested, we should either have 0 results or all
     # should be available, so if we found at least one result but not all of them, this is a problem with our setup
     assert not results or len(results) == len(image_types), \
@@ -181,23 +194,6 @@ def build_images(shared_tmpdir, build_container, request, force_aws_upload):
     print(f"Requested {len(image_types)} images but found {len(results)} cached images. Building...")
 
     # not all requested image types are available - build them
-    cfg = {
-        "blueprint": {
-            "customizations": {
-                "user": [
-                    {
-                        "name": username,
-                        "password": password,
-                        "groups": ["wheel"],
-                    },
-                ],
-            },
-        },
-    }
-
-    config_json_path = output_path / "config.json"
-    config_json_path.write_text(json.dumps(cfg), encoding="utf-8")
-
     cursor = testutil.journal_cursor()
 
     upload_args = []
@@ -245,7 +241,7 @@ def build_images(shared_tmpdir, build_container, request, force_aws_upload):
             *creds_args,
             build_container,
             container_ref,
-            "--config", "/output/config.json",
+            "--experimental-root-ssh-authorized-key", ssh_pubkey,
             *types_arg,
             *upload_args,
             *target_arch_args,
@@ -281,8 +277,10 @@ def del_ami():
 
     results = []
     for image_type in image_types:
-        results.append(ImageBuildResult(image_type, artifact[image_type], target_arch,
-                                        username, password, bib_output, journal_output, metadata))
+        results.append(ImageBuildResult(
+            image_type, artifact[image_type], target_arch,
+            "root", ssh_keyfile_private_path,
+            bib_output, journal_output, metadata))
     yield results
 
     # Try to cache as much as possible
@@ -318,9 +316,10 @@ def test_image_is_generated(image_type):
 @pytest.mark.parametrize("image_type", gen_testcases("qemu-boot"), indirect=["image_type"])
 def test_image_boots(image_type):
     with QEMU(image_type.img_path, arch=image_type.img_arch) as test_vm:
-        exit_status, _ = test_vm.run("true", user=image_type.username, password=image_type.password)
+        exit_status, _ = test_vm.run("true", user=image_type.username, keyfile=image_type.ssh_keyfile_private_path)
         assert exit_status == 0
-        exit_status, output = test_vm.run("echo hello", user=image_type.username, password=image_type.password)
+        exit_status, output = test_vm.run(
+            "echo hello", user=image_type.username, keyfile=image_type.ssh_keyfile_private_path)
         assert exit_status == 0
         assert "hello" in output
 
@@ -337,9 +336,10 @@ def test_ami_boots_in_aws(image_type, force_aws_upload):
     # 4.30 GiB / 10.00 GiB [------------>____________] 43.02% 58.04 MiB p/s
     assert "] 100.00%" in image_type.bib_output
     with AWS(image_type.metadata["ami_id"]) as test_vm:
-        exit_status, _ = test_vm.run("true", user=image_type.username, password=image_type.password)
+        exit_status, _ = test_vm.run("true", user=image_type.username, keyfile=image_type.ssh_keyfile_private_path)
         assert exit_status == 0
-        exit_status, output = test_vm.run("echo hello", user=image_type.username, password=image_type.password)
+        exit_status, output = test_vm.run(
+            "echo hello", user=image_type.username, keyfile=image_type.ssh_keyfile_private_path)
         assert exit_status == 0
         assert "hello" in output
 
@@ -401,7 +401,7 @@ def test_iso_installs(image_type):
     # boot test disk and do extremly simple check
     with QEMU(test_disk_path) as vm:
         vm.start(use_ovmf=True)
-        exit_status, _ = vm.run("true", user=image_type.username, password=image_type.password)
+        exit_status, _ = vm.run("true", user=image_type.username, keyfile=image_type.ssh_keyfile_private_path)
         assert exit_status == 0
 
 

test/vm.py

@@ -1,5 +1,6 @@
 import abc
 import os
+import paramiko
 import pathlib
 import platform
 import subprocess
@@ -43,16 +44,19 @@ def force_stop(self):
         Stop the VM and clean up any resources that were created when setting up and starting the machine.
         """
 
-    def run(self, cmd, user, password):
+    def run(self, cmd, user, keyfile):
         """
         Run a command on the VM via SSH using the provided credentials.
         """
         if not self.running():
             self.start()
         client = SSHClient()
         client.set_missing_host_key_policy(AutoAddPolicy)
+        # workaround, see https://github.com/paramiko/paramiko/issues/2048
+        pkey = paramiko.RSAKey.from_private_key_file(keyfile)
         client.connect(
-            self._address, self._ssh_port, user, password,
+            self._address, self._ssh_port,
+            user, pkey=pkey,
             allow_agent=False, look_for_keys=False)
         chan = client.get_transport().open_session()
         chan.get_pty()