From a9192672db9fa36480c6bd148c377e851074a0d5 Mon Sep 17 00:00:00 2001 From: Allison Karlitskaya Date: Fri, 12 Sep 2025 11:09:38 +0200 Subject: [PATCH 1/6] image-create: rewrite using test.thing Drop the --no-save, --upload, and --store arguments. Nobody uses those: image-refresh calls image-upload for itself. --- image-create | 401 +++++++------- lib/testthing.py | 1299 ++++++++++++++++++++++++++++++++++++++++++++++ pyproject.toml | 1 + test/run | 2 +- 4 files changed, 1494 insertions(+), 209 deletions(-) create mode 100755 lib/testthing.py diff --git a/image-create b/image-create index b83c38ea5b..32cd4803b2 100755 --- a/image-create +++ b/image-create @@ -1,215 +1,200 @@ #!/usr/bin/env python3 -# This file is part of Cockpit. -# -# Copyright (C) 2015 Red Hat, Inc. -# -# Cockpit is free software; you can redistribute it and/or modify it -# under the terms of the GNU Lesser General Public License as published by -# the Free Software Foundation; either version 2.1 of the License, or -# (at your option) any later version. -# -# Cockpit is distributed in the hope that it will be useful, but -# WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU -# Lesser General Public License for more details. -# -# You should have received a copy of the GNU Lesser General Public License -# along with Cockpit; If not, see . - -# image-create -- Make a root image suitable for use with vm-run. -# -# Installs the OS indicated by TEST_OS into the image -# for test machine and tweaks it to be useable with -# vm-run and testlib.py. +# SPDX-License-Identifier: GPL-3.0-or-later + +"""image-create - Make a root image suitable for use with vm-run. + +Installs the OS indicated by TEST_OS into the image for test machine and tweaks +it to be useable with vm-run and testlib.py. +""" import argparse +import asyncio +import hashlib import os -import shutil -import subprocess -import sys import tempfile +from pathlib import Path + +from lib import testthing +from lib.aio.jsonutil import JsonObject +from lib.constants import IMAGES_DIR, MACHINE_DIR, SCRIPTS_DIR +from lib.directories import get_images_data_dir + + +def cloud_init_config(image: str) -> JsonObject: + vm_pass = "$5$CockpitCloudInit$Iw89f.aPgqHPXAHC2Zs9h9335n3E1FQDFvR6MLqwPK9" + ssh_authorized_keys = [Path(MACHINE_DIR, "identity.pub").read_text()] + host_key = Path(MACHINE_DIR, "host_key").read_text() + host_key_pub = Path(MACHINE_DIR, "host_key.pub").read_text() + + bootcmd = [] + runcmd = ["printf X_SYSTEMD_UNIT_ACTIVE=los-gehts > /dev/virtio-ports/tt-notify"] + + if image in ["arch", "centos-9-bootc"]: + bootcmd.append("echo 100::55:4e4b:4e4f:574e UNKNOWN >> /etc/hosts") + + if image in ["arch"]: + bootcmd.append("systemctl disable --now systemd-time-wait-sync.service || true") + + if image in ["opensuse-tumbleweed"]: + bootcmd.append("echo 'Include /usr/etc/ssh/sshd_config' > /etc/ssh/sshd_config") + + return { + "users": [ + "default", + { + "name": "root", + "lock_passwd": False, + "hashed_passwd": vm_pass, + "groups": "users,wheel", + "ssh_authorized_keys": ssh_authorized_keys, + }, + { + "name": "admin", + "lock_passwd": False, + "hashed_passwd": vm_pass, + "gecos": "Administrator", + "groups": "users,wheel", + "ssh_authorized_keys": ssh_authorized_keys, + }, + ], + "ssh_pwauth": True, + "ssh_keys": { + "rsa_private": host_key, + "rsa_public": host_key_pub, + }, + "bootcmd": [("sh", "-c", cmd) for cmd in bootcmd], + "runcmd": [("sh", "-c", cmd) for cmd in runcmd], + "cloud_final_modules": [ + "scripts-per-once", + "scripts-per-boot", + "scripts-per-instance", + ["scripts-user", "always"], + "final-message", + ], + } + + +class ImageCreate: + def __init__(self, *, image: str, no_build: bool, sit: bool, verbose: bool) -> None: + self.ui = testthing.UI(status_messages=True, verbose=verbose) + + if image in ["fedora-coreos", "services"] and not no_build: + self.ui.print_verbose("Creating machine without build capabilities based on the image type") + no_build = True + + self.image = image + self.no_build = no_build + self.sit = sit + + def heading(self, text: str) -> None: + self.ui.print(f"\n\033[1m## {text}\033[0m\n") + + async def bootstrap(self, tmpfile: Path) -> None: + self.heading("Run bootstrap script on host") + bootstrap_script = Path(SCRIPTS_DIR) / f"{self.image}.bootstrap" + await self.ui.run(bootstrap_script, tmpfile) + + async def setup(self, tmpfile: Path) -> None: + self.heading("Start guest") + cockpit_test_identity = Path(MACHINE_DIR) / "identity" + cockpit_test_identity.chmod(0o600) -from lib.constants import BOTS_DIR -from machine import testvm - -parser = argparse.ArgumentParser(description='Create a virtual machine image') -parser.add_argument('-v', '--verbose', action='store_true', help='Display verbose progress details') -parser.add_argument('-s', '--sit', action='store_true', help='Sit and wait if setup script fails') -parser.add_argument('-n', '--no-save', action='store_true', help='Don\'t save the new image') -parser.add_argument('-u', '--upload', action='store_true', help='Upload the image after creation') -parser.add_argument('--no-build', action='store_true', dest='no_build', - help="Don't build packages and create the vm without build capabilities") -parser.add_argument("--store", default=None, help="Where to send images") -parser.add_argument('image', help='The image to create') -args = parser.parse_args() - -# default to --no-build for some images -if args.image in ["fedora-coreos", "services"]: - if not args.no_build: - if args.verbose: - print("Creating machine without build capabilities based on the image type") - args.no_build = True - - -class MachineBuilder: - def __init__(self, machine: testvm.VirtMachine): - self.machine = machine - self.iso = self.machine.image.endswith("boot") - self.payload = self.machine.image.endswith("payload") - if self.iso: - self.suffix = ".iso" - elif self.payload: - self.suffix = ".tar.gz" - else: - self.suffix = ".qcow2" - - # Use /var/tmp/ as this is going to be a huge file; /tmp/ is commonly tmpfs - self.target_file = self.machine.image_file - fp, self.machine.image_file = tempfile.mkstemp(dir="/var/tmp/", prefix=self.machine.image, suffix=self.suffix) - os.close(fp) - - def bootstrap_system(self) -> None: - assert not self.machine._domain - - os.makedirs(self.machine.run_dir, 0o750, exist_ok=True) - - bootstrap_script = os.path.join(testvm.SCRIPTS_DIR, f"{self.machine.image}.bootstrap") - - if os.path.isfile(bootstrap_script): - subprocess.check_call([bootstrap_script, self.machine.image_file]) - else: - raise testvm.Failure(f"Unsupported OS {self.machine.image}: {bootstrap_script}") - - def run_setup_script(self, script: str) -> None: - """Prepare a test image further by running some commands in it.""" - self.machine.start() - - try: - self.machine.wait_boot(timeout_sec=300) - self.machine.upload([os.path.join(testvm.SCRIPTS_DIR, "lib/")], "/var/lib/testvm") - self.machine.upload([script], "/var/tmp/SETUP") - - env = { - "TEST_OS": self.machine.image, - "DO_BUILD": "0" if args.no_build else "1", - "SERVER_REPO_URL": os.environ.get("SERVER_REPO_URL", ""), - "EXTRAS_REPO_URL": os.environ.get("EXTRAS_REPO_URL", ""), - "BASEOS_REPO_URL": os.environ.get("BASEOS_REPO_URL", ""), - "APPSTREAM_REPO_URL": os.environ.get("APPSTREAM_REPO_URL", ""), - } - self.machine.message("run setup script on guest") - - try: - self.machine.execute(f"/var/tmp/SETUP {self.machine.image}", - environment=env, stdout=None, quiet=not self.machine.verbose, timeout=7200) - self.machine.execute("rm -f /var/tmp/SETUP") - except subprocess.CalledProcessError as exc: - if args.sit: - sys.stderr.write(self.machine.diagnose()) - input("Press RET to continue... ") - raise testvm.Failure(f"setup failed with code {exc.returncode}\n") from exc - - finally: - self.machine.stop(timeout_sec=500) - - def boot_system(self) -> None: - """Start the system to make sure it can boot, then shutdown cleanly""" - - self.machine.start() - # avoid too long boot times -- they cause painfully long tests, and are a bug - try: - timeout = 30 - if self.machine.image == "fedora-rawhide": - # except with fedora-rawhide, which is slow by design: - # https://lists.fedoraproject.org/archives/list/devel@lists.fedoraproject.org/thread/HRJT4UF35MRJ7C4ZXVHBFXB4BTW64YSO/ - timeout = 120 - self.machine.wait_boot(timeout_sec=timeout) - finally: - self.machine.stop(timeout_sec=30) - - def build(self) -> None: - self.bootstrap_system() - - # gather the scripts, separated by reboots - script = os.path.join(testvm.SCRIPTS_DIR, f"{self.machine.image}.setup") - - if not os.path.exists(script): - return - - self.machine.message("Running setup script %s" % (script)) - self.run_setup_script(script) - - # make sure we can boot the system - self.boot_system() - - def save(self) -> None: - data_dir = testvm.get_images_data_dir() - - os.makedirs(data_dir, 0o750, exist_ok=True) - - if not os.path.exists(self.machine.image_file): - raise testvm.Failure("Nothing to save.") - - if not self.iso and not self.payload: - rebuild = os.path.join(data_dir, self.machine.image + ".rebuild") - - # Copy image via convert, to make it sparse again - subprocess.check_call(["qemu-img", "convert", "-c", "-O", "qcow2", self.machine.image_file, rebuild]) - - # Hash the image here - sha, _, _rest = subprocess.check_output( - ["sha256sum", self.machine.image_file if (self.iso or self.payload) else rebuild], text=True - ).partition(" ") - if not sha: - raise testvm.Failure("sha256sum returned invalid output") - - name = self.machine.image + "-" + sha + self.suffix - data_file = os.path.join(data_dir, name) - if os.path.exists(data_file): - # shutil.move has trouble when the destination exists but - # has too restrictive permissions to be overwritten. - os.unlink(data_file) - shutil.move(self.machine.image_file if (self.iso or self.payload) else rebuild, data_file) - - if not self.iso and not self.payload: - # Remove temp image file - os.unlink(self.machine.image_file) - - # Update the images symlink - if os.path.islink(self.target_file): - os.unlink(self.target_file) - os.symlink(name, self.target_file) - - # Handle alternate images data directory - image_file = os.path.join(testvm.IMAGES_DIR, name) - if not os.path.exists(image_file): - os.symlink(os.path.abspath(data_file), image_file) - - -try: - if args.image == 'services': # deploying candlepin needs oodles of memory - memory_mb = 3072 - else: - memory_mb = 2048 - - machine = testvm.VirtMachine(verbose=args.verbose, - image=args.image, - memory_mb=memory_mb, - maintain=True) - builder = MachineBuilder(machine) - builder.build() - if not args.no_save: - print("Saving...") - builder.save() - if args.upload: - print("Uploading...") - cmd = [os.path.join(BOTS_DIR, "image-upload"), '--prune-s3'] - if args.store: - cmd += ["--store", args.store] - cmd += [args.image] - subprocess.check_call(cmd) - -except testvm.Failure as ex: - sys.stderr.write("image-create: %s\n" % ex) - sys.exit(1) + memory = "3072M" if self.image in ["services"] else "2048M" + + with testthing.IpcDirectory() as ipc: + async with testthing.VirtualMachine( + boot="mbr", + cloud_init_user_data=cloud_init_config(self.image), + identity=(cockpit_test_identity, None), + image=tmpfile, + ipc=ipc, + memory=memory, + networks=[testthing.Network("user")], + snapshot=False, + sit=self.sit, + target="los-gehts", + timeout=300, + ui=self.ui, + ) as vm: + scripts = Path(SCRIPTS_DIR) + await vm.wait_boot() + + self.heading("Run setup script on guest") + + await vm.scp("-r", scripts / "lib", "vm:/var/lib/testvm") + await vm.scp("-r", scripts / f"{self.image}.setup", "vm:/var/tmp/SETUP") + await vm.execute( + "/var/tmp/SETUP", + self.image, + environment={ + "TEST_OS": self.image, + "DO_BUILD": "0" if self.no_build else "1", + "SERVER_REPO_URL": os.environ.get("SERVER_REPO_URL", ""), + "EXTRAS_REPO_URL": os.environ.get("EXTRAS_REPO_URL", ""), + "BASEOS_REPO_URL": os.environ.get("BASEOS_REPO_URL", ""), + "APPSTREAM_REPO_URL": os.environ.get("APPSTREAM_REPO_URL", ""), + }, + stdout=None, + ) + + self.heading("Shutting down guest") + + async def save(self, tmpfile: Path) -> None: + self.heading("Saving image") + data_dir = Path(get_images_data_dir()) + data_dir.mkdir(exist_ok=True, mode=0o750, parents=True) + + with tempfile.NamedTemporaryFile( + dir=data_dir, prefix=f"{self.image}", suffix=".sparse.qcow2", delete_on_close=False + ) as tmp: + # Copy image via convert, to make it sparse again + sparse = Path(tmp.name) + await self.ui.run( + ("qemu-img", "convert"), + ("-f", "qcow2", tmpfile), + ("-O", "qcow2", "-c", sparse), + ) + + # Hash the image here + with sparse.open("rb") as fp: + sha = hashlib.file_digest(fp, "sha256").hexdigest() + name = f"{self.image}-{sha}.qcow2" + + # a simple rename always works because it's the same directory + data_file = data_dir / name + self.ui.print_verbose(f"~ mv \\\n {sparse} \\\n {name}\n") + sparse.rename(data_file) + tmp.close() + + images_dir = Path(IMAGES_DIR) + + self.ui.print_verbose(f"~ ln -sf \\\n {data_file} \\\n {images_dir}/\n") + (images_dir / name).unlink(missing_ok=True) + (images_dir / name).symlink_to(data_file) + + self.ui.print_verbose(f"~ ln -sf \\\n {name} \\\n {images_dir}/{self.image}\n") + (images_dir / self.image).unlink(missing_ok=True) + (images_dir / self.image).symlink_to(name) + + async def create(self) -> None: + with testthing.cli_helper(), tempfile.NamedTemporaryFile() as tmp: + tmpfile = Path(tmp.name) + await self.bootstrap(tmpfile) + await self.setup(tmpfile) + await self.save(tmpfile) + + +def main() -> None: + parser = argparse.ArgumentParser(description="Create a virtual machine image") + parser.add_argument("-v", "--verbose", action="store_true", help="Display verbose progress details") + parser.add_argument("-s", "--sit", action="store_true", help="Sit and wait if setup script fails") + parser.add_argument("--no-build", action="store_true", help="Create the VM without build capabilities") + parser.add_argument("image", help="The image to create") + args = parser.parse_args() + + asyncio.run(ImageCreate(**vars(args)).create()) + + +if __name__ == "__main__": + main() diff --git a/lib/testthing.py b/lib/testthing.py new file mode 100755 index 0000000000..6990027cfd --- /dev/null +++ b/lib/testthing.py @@ -0,0 +1,1299 @@ +#!/usr/bin/python3 +# SPDX-License-Identifier: GPL-3.0-or-later + +"""test.thing - A simple modern VM runner. + +https://codeberg.org/lis/test.thing + +A simple VM runner script exposing an API useful for use as a pytest fixture. +Can also be used to run a VM and login via the console. + +Each VM is allocated an identifier: 'tt.0', 'tt.1', etc. + +For each VM, an ephemeral ssh key is created and used to connect to the VM via +vsock with systemd-ssh-proxy, which works even if the guest doesn't have +networking configured. The ephemeral key means that access is limited to the +current user (since vsock connections are otherwise available to all users on +the host system). The guest needs to have systemd 256 for this to work. + +An ssh control socket is created for sending commands and can also be used +externally, avoiding the need to authenticate. A suggested ssh config: + +``` +Host tt.* + ControlPath ${XDG_RUNTIME_DIR}/test.thing/%h/ssh +``` + +And then you can say `ssh tt.0` or `scp file tt.0:/tmp`. +""" + +# When copying test.thing into your own project, try to use a tagged version. +# If you need to use a version between tags or have made your own +# modifications, please make note if it by modifying the version number. +__version__ = "0.3.4+hacks!" + +import argparse +import asyncio +import contextlib +import ctypes +import dataclasses +import functools +import itertools +import json +import logging +import os +import pathlib +import re +import shlex +import shutil +import signal +import sys +import tempfile +import traceback +import weakref +from collections.abc import AsyncGenerator, Callable, Iterable, Mapping, Sequence, Iterator +from pathlib import Path +from types import TracebackType +from typing import Any, Literal, Never, Self + +logger = logging.getLogger(__name__) + +COCKPIT_TEST_IDENTITY = """ +-----BEGIN RSA PRIVATE KEY----- +MIIEpQIBAAKCAQEA1DrTSXQRF8isQQfPfK3U+eFC4zBrjur+Iy15kbHUYUeSHf5S +jXPYbHYqD1lHj4GJajC9okle9rykKFYZMmJKXLI6987wZ8vfucXo9/kwS6BDAJto +ZpZSj5sWCQ1PI0Ce8CbkazlTp5NIkjRfhXGP8mkNKMEhdNjaYceO49ilnNCIxhpb +eH5dH5hybmQQNmnzf+CGCCLBFmc4g3sFbWhI1ldyJzES5ZX3ahjJZYRUfnndoUM/ +TzdkHGqZhL1EeFAsv5iV65HuYbchch4vBAn8jDMmHh8G1ixUCL3uAlosfarZLLyo +3HrZ8U/llq7rXa93PXHyI/3NL/2YP3OMxE8baQIDAQABAoIBAQCxuOUwkKqzsQ9W +kdTWArfj3RhnKigYEX9qM+2m7TT9lbKtvUiiPc2R3k4QdmIvsXlCXLigyzJkCsqp +IJiPEbJV98bbuAan1Rlv92TFK36fBgC15G5D4kQXD/ce828/BSFT2C3WALamEPdn +v8Xx+Ixjokcrxrdeoy4VTcjB0q21J4C2wKP1wEPeMJnuTcySiWQBdAECCbeZ4Vsj +cmRdcvL6z8fedRPtDW7oec+IPkYoyXPktVt8WsQPYkwEVN4hZVBneJPCcuhikYkp +T3WGmPV0MxhUvCZ6hSG8D2mscZXRq3itXVlKJsUWfIHaAIgGomWrPuqC23rOYCdT +5oSZmTvFAoGBAPs1FbbxDDd1fx1hisfXHFasV/sycT6ggP/eUXpBYCqVdxPQvqcA +ktplm5j04dnaQJdHZ8TPlwtL+xlWhmhFhlCFPtVpU1HzIBkp6DkSmmu0gvA/i07Z +pzo5Z+HRZFzruTQx6NjDtvWwiXVLwmZn2oiLeM9xSqPu55OpITifEWNjAoGBANhH +XwV6IvnbUWojs7uiSGsXuJOdB1YCJ+UF6xu8CqdbimaVakemVO02+cgbE6jzpUpo +krbDKOle4fIbUYHPeyB0NMidpDxTAPCGmiJz7BCS1fCxkzRgC+TICjmk5zpaD2md +HCrtzIeHNVpTE26BAjOIbo4QqOHBXk/WPen1iC3DAoGBALsD3DSj46puCMJA2ebI +2EoWaDGUbgZny2GxiwrvHL7XIx1XbHg7zxhUSLBorrNW7nsxJ6m3ugUo/bjxV4LN +L59Gc27ByMvbqmvRbRcAKIJCkrB1Pirnkr2f+xx8nLEotGqNNYIawlzKnqr6SbGf +Y2wAGWKmPyEoPLMLWLYkhfdtAoGANsFa/Tf+wuMTqZuAVXCwhOxsfnKy+MNy9jiZ +XVwuFlDGqVIKpjkmJyhT9KVmRM/qePwgqMSgBvVOnszrxcGRmpXRBzlh6yPYiQyK +2U4f5dJG97j9W7U1TaaXcCCfqdZDMKnmB7hMn8NLbqK5uLBQrltMIgt1tjIOfofv +BNx0raECgYEApAvjwDJ75otKz/mvL3rUf/SNpieODBOLHFQqJmF+4hrSOniHC5jf +f5GS5IuYtBQ1gudBYlSs9fX6T39d2avPsZjfvvSbULXi3OlzWD8sbTtvQPuCaZGI +Df9PUWMYZ3HRwwdsYovSOkT53fG6guy+vElUEDkrpZYczROZ6GUcx70= +-----END RSA PRIVATE KEY----- +""" +"""A copy of `bots/machine/identity` from the cockpit project. Many existing +VM images have the public half of this key inside of them, so it's useful for +gaining access to those if they lack support for ephemeral ssh keys.""" + +COCKPIT_TEST_IDENTITY_PUB = "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQDUOtNJdBEXyKxBB898rdT54ULjMGuO6v4jLXmRsdRhR5Id/lKNc9hsdioPWUePgYlqML2iSV72vKQoVhkyYkpcsjr3zvBny9+5xej3+TBLoEMAm2hmllKPmxYJDU8jQJ7wJuRrOVOnk0iSNF+FcY/yaQ0owSF02Nphx47j2KWc0IjGGlt4fl0fmHJuZBA2afN/4IYIIsEWZziDewVtaEjWV3InMRLllfdqGMllhFR+ed2hQz9PN2QcapmEvUR4UCy/mJXrke5htyFyHi8ECfyMMyYeHwbWLFQIve4CWix9qtksvKjcetnxT+WWrutdr3c9cfIj/c0v/Zg/c4zETxtp cockpit-test" # noqa: E501 +"""A copy of `bots/machine/identity.pub` from the cockpit project. The +corresponding public key for COCKPIT_TEST_IDENTITY.""" + + +# This is basically tempfile.TemporaryDirectory but sequentially-allocated. +# We do that so we can easily interact with the VMs from outside (with ssh). +class IpcDirectory: + """A context manager for the VM IPC directory. + + This is very similar to tempfile.TemporaryDirectory() except that the + allocation is predictable (sequential): the created directory will be + `/run/user/$uid/test.thing/tt.n` for the smallest `n` that we find. + + It works the same way: + + with IpcDirectory() as path: + ...use path... + + The directory gets tagged with a `pid` file containing the pid and pidfd + inode of the current process. This could be helpful for pruning dead + directories, but is currently unused. + """ + + finalizer: Callable[[], None] | None = None + + @staticmethod + def _find_dir() -> Path: + try: + xdg_rundir = Path(os.environ["XDG_RUNTIME_DIR"]) + except KeyError: + # No XDG_RUNTIME_DIR? Somewhere in /tmp will have to do + return Path(tempfile.mkdtemp()) + + for n in range(10000): + tmpdir = xdg_rundir / "test.thing" / f"tt.{n}" + + try: + tmpdir.mkdir(exist_ok=False, parents=True, mode=0o700) + except FileExistsError: + continue + + return tmpdir + + raise FileExistsError + + def __enter__(self) -> Path: + """Create a unique directory. + + This will sequentially allocate the first available 'tt.0', 'tt.1', + etc. directory and return it as a `Path`. + """ + pid = os.getpid() + pidfd = os.pidfd_open(pid) + try: + buf = os.fstat(pidfd) + unique_id = f"{pid} {buf.st_ino}\n" + finally: + os.close(pidfd) + + tmpdir = self._find_dir() + self.finalizer = weakref.finalize(self, shutil.rmtree, tmpdir) + (tmpdir / "pid").write_text(unique_id) + return tmpdir + + def __exit__(self, *args: object) -> None: + """Delete the IPC directory and its contents.""" + del args + if self.finalizer: + self.finalizer() + + +def _normalize_args( + *args: str | pathlib.PurePath | tuple[str | pathlib.PurePath, ...], +) -> Iterable[Iterable[str]]: + for chunk in args: + if not isinstance(chunk, tuple): + yield (str(chunk),) + elif len(chunk) != 0: + yield map(str, chunk) + + +def _pretty_print_args( + *args: str | pathlib.PurePath | tuple[str | pathlib.PurePath, ...], +) -> str: + """Pretty-print a nested argument list. + + This takes the argument list format used by test.thing and turns it into a + format that looks like a nicer version of `set -x` from POSIX shell. + """ + if not any(isinstance(arg, tuple) for arg in args): + # No tuples: use the boring format + return shlex.join(map(str, args)) + + # There are tuples: use the fancy format + return " \\\n ".join(map(shlex.join, _normalize_args(*args))) + + +def _find_qemu() -> Path: + for candidate in ("qemu-kvm", "kvm"): + if cmd := shutil.which(candidate): + return Path(cmd) + + raise FileNotFoundError("Unable to find qemu-kvm") + + +def _find_ovmf() -> Path: + candidates = [ + # path for Fedora/RHEL (our tasks container) + "/usr/share/OVMF/OVMF_CODE.fd", + # path for Ubuntu (GitHub Actions runners) + "/usr/share/ovmf/OVMF.fd", + # path for Arch + "/usr/share/edk2/x64/OVMF.4m.fd", + ] + + for path in map(Path, candidates): + if path.exists(): + return path + + raise FileNotFoundError("Unable to find OVMF UEFI BIOS") + + +async def _qmp_command(ipc: Path, command: str) -> object: + reader, writer = await asyncio.open_unix_connection(ipc / "qmp") + + async def execute(command: str) -> object: + writer.write((json.dumps({"execute": command}) + "\n").encode()) + await writer.drain() + while True: + response = json.loads(await reader.readline()) + if "event" in response: + continue + if "return" in response: + return response["return"] + raise RuntimeError(f"Got error response from qmp: {response!r}") + + # Trivial handshake (ignore them, send nothing) + _ = json.loads(await reader.readline()) + await execute("qmp_capabilities") + + response = await execute(command) + + writer.close() + await writer.wait_closed() + + return response + + +def _ssh_direct_args(identities: Sequence[Path], port: Path) -> tuple[tuple[str, str], ...]: + options = { + # Fake that we know the host key + "KnownHostsCommand": "/bin/echo %H %t %K", + # Use systemd-ssh-proxy to connect via vsock + "ProxyCommand": f"/usr/lib/systemd/systemd-ssh-proxy vsock-mux/{port} 22", + "ProxyUseFdpass": "yes", + # Try to prevent interactive prompting and/or updating known_hosts + # files or otherwise interacting with the environment + "BatchMode": "yes", + "IdentitiesOnly": "yes", + "PKCS11Provider": "none", + "PasswordAuthentication": "no", + "StrictHostKeyChecking": "yes", + "User": "root", + "UserKnownHostsFile": "/dev/null", + } + + return ( + ("-F", "none"), # don't use the user's config + *(("-o", f"{k} {v}") for k, v in options.items()), + *(("-o", f"IdentityFile {path}") for path in identities), + ) + + +@functools.cache +def _stderr_is_tty() -> bool: + return os.isatty(2) + + +class UI: + def __init__(self, *, status_messages: bool, verbose: bool) -> None: + self._status_messages = status_messages + self._verbose = verbose + + def clear_status_message(self) -> None: + if _stderr_is_tty(): + sys.stderr.write("\r\033[2K") + + def print_status(self, line: str) -> None: + if self._status_messages: + if os.isatty(2): + sys.stderr.write("\r\033[2K " + line + "\r") + else: + sys.stderr.write(line + "\n") + + def print_verbose(self, line: str) -> None: + if self._verbose: + self.clear_status_message() + sys.stderr.write(line + "\n") + + def print(self, line: str) -> None: + self.clear_status_message() + sys.stderr.write(line + "\n") + + async def _wait_stdin(self, msg: str) -> None: + r"""Wait until stdin sees \n or EOF. + + This prints the given message to stdout without adding an extra newline. + + The input is consumed (and discarded) up to the \n and maybe more... + """ + done = asyncio.Event() + + def stdin_ready() -> None: + data = os.read(0, 4096) + if not data: + sys.stdout.write("\n") + if not data or b"\n" in data: + done.set() + + loop = asyncio.get_running_loop() + loop.add_reader(0, stdin_ready) + sys.stdout.write(msg) + sys.stdout.flush() + try: + await done.wait() + finally: + loop.remove_reader(0) + + async def sit(self, msg: str | None = None, exc: BaseException | None = None) -> None: + """Wait for the user to press Enter.""" + # lis: the ages old design question: does the button show + # the current state or the future one when you press it :) + # pitti: that's exactly my question. by taking the one with + # both then i don't have to choose! :D + # although to be honest, i find your argument convincing. + # i'll put the ⏸️ back + # lis: it was more of a joke -- I actually agree that a + # play/pause button is nicer + # too late lol + + self.clear_status_message() + + if exc is not None: + sys.stderr.write(f"\n🤦 {''.join(traceback.format_exception(exc))}\n") + if msg is not None: + sys.stderr.write(f"\n{msg}\n") + + await self._wait_stdin("\nEnter or EOF to exit ⏸️ ") + + async def spawn( + self, + *args: str | Path | tuple[str | Path, ...], + stdin: int | None = None, + stdout: int | None = None, + stderr: int | None = None, + ) -> asyncio.subprocess.Process: + """Spawn a process. + + This has a couple of extra niceties: the args list is flattened, Path is + converted to str, the spawned process is logged to stderr for debugging, + and we call PR_SET_PDEATHSIG with SIGTERM after forking to make sure the + process exits with us. + + The flattening allows grouping logically-connected arguments together, + producing nicer verbose output, allowing for adding groups of arguments + from helper functions or comprehensions, and works nicely with code + formatters: + + For example: + + private = Path(...) + options = { ... } + + ssh = await spawn( + "ssh", + ("-i", private), + *(("-o", f"{k} {v}") for k, v in options.items()), + ("-l", "root", "x"), + ... + ) + + The type of the groups is `tuple`. It could be `Sequence` but this would + also allow using bare strings, which would be split into their individual + characters. Using `tuple` prevents this from happening. + """ + # This might be complicated: do it before the fork + prctl = ctypes.CDLL(None, use_errno=True).prctl + + def pr_set_pdeathsig() -> None: + PR_SET_PDEATHSIG = 1 # noqa: N806 + if prctl(PR_SET_PDEATHSIG, signal.SIGTERM): + os._exit(1) # should never happen + + # If we are expecting output, clear the status message + if stdout is None or stderr is None: + self.clear_status_message() + + self.print_verbose(f"+ {_pretty_print_args(*args)}\n") + + return await asyncio.subprocess.create_subprocess_exec( + *itertools.chain(*_normalize_args(*args)), + stdin=stdin, + stdout=stdout, + stderr=stderr, + preexec_fn=pr_set_pdeathsig, + ) + + async def run( + self, + *args: str | Path | tuple[str | Path, ...], + check: bool = True, + ) -> int: + """Run a process, waiting for it to exit. + + This takes the same arguments as _spawn, plus a "check" argument (True by + default) which works in the usual way. + """ + process = await self.spawn( + *args, + stdin=asyncio.subprocess.DEVNULL, + stdout=None if self._verbose else asyncio.subprocess.DEVNULL, + stderr=None if self._verbose else asyncio.subprocess.DEVNULL, + ) + returncode = await process.wait() + if check and returncode != 0: + raise SubprocessError(args, returncode=returncode) + return returncode + + +class GuestPath(pathlib.PurePosixPath): + """A path on the virtual machine guest. + + This aims to support similar operations to pathlib.Path (with similar + APIs), but most operations are async and many have slightly different + feature sets. + """ + + __slots__ = ("_vm",) + + def __init__(self, *args: str | os.PathLike[str], vm: "VirtualMachine") -> None: + """Create a GuestPath for a path on a guest.""" + super().__init__(*args) + self._vm = vm + + def with_segments(self, *pathsegments: str | os.PathLike[str]) -> Self: + """Create a new path by combining the given pathsegments.""" + return type(self)(*pathsegments, vm=self._vm) + + async def mkdir(self, *, mode: int | None = None, parents: bool = False) -> None: + """Create a directory.""" + await self._vm.execute( + "mkdir", + "-p" if parents else (), + ("-m", f"{mode:0o}") if mode is not None else (), + self, + ) + + async def chmod(self, mode: int | str, *, follow_symlinks: bool = True) -> None: + """Change a file mode.""" + await self._vm.execute( + "chmod", + "-h" if not follow_symlinks else (), + f"{mode:0o}" if isinstance(mode, int) else mode, + self, + ) + + async def chown( + self, + owner: str | tuple[str | int | None, str | int | None], + *, + follow_symlinks: bool = True, + ) -> None: + """Change the owner of a file. + + The owner can be a string like 'user:group' or a pair of (user, group) + where each can be a string, int, or None (to make no change). + """ + if isinstance(owner, tuple): + user, group = owner + owner = f"{user or ''}:{group or ''}" + + await self._vm.execute("chown", "-h" if not follow_symlinks else (), owner, self) + + async def write_bytes(self, data: bytes, *, append: bool = False) -> None: + """Write or append to to a binary file.""" + await self._vm.execute( + "dd", + "status=none", + ("conv=notrunc", "oflag=append") if append else (), + f"of={self}", + input=data, + ) + + async def read_text(self) -> str: + """Read a text file.""" + return await self._vm.execute("cat", self) + + async def write_text(self, data: str, *, append: bool = False) -> None: + """Write or append to a text file.""" + await self.write_bytes(data.encode(), append=append) + + async def unlink(self, *, missing_ok: bool = False, recursive: bool = False) -> None: + """Unlink the given file.""" + await self._vm.execute("rm", "-f" if missing_ok else (), "-r" if recursive else (), self) + + async def rmdir(self) -> None: + """Remove a directory.""" + await self._vm.execute("rmdir", self) + + +@dataclasses.dataclass +class Network: + """A virtual machine network.""" + + id: int | Literal["user"] + """The network identifier. If this is an integer then it specifies a + multicast network on which other virtual machines can communicate. If it's + the literal value "user" then this sets up usermode networking.""" + + @classmethod + def user(cls) -> Self: + """Create a user-mode network.""" + return cls(id="user") + + @classmethod + def multicast(cls, netnr: int) -> Self: + """Create a multicast network for talking to other machines.""" + return cls(id=netnr) + + def to_qemu(self) -> str: + """Describe the network in a way that qemu understands.""" + if self.id == "user": + return "user" + + # Same as Cockpit + return f"socket,mcast=230.0.0.1:{self.id},localaddr=127.0.0.1" + + +class VirtualMachine: + """A handle to a running virtual machine. + + This is meant to be used as an async context manager like so: + + with IpcDirectory() as ipc: + image = Path("...") + async with VirtualMachine(image, ipc=ipc) as vm: + await vm.execute("cat", "/usr/lib/os-release") + + The user of the context manager runs in context of an asyncio.TaskGroup and + will be cancelled if anything unexpected happens (ssh connection lost, VM + exiting, etc). + + When the context manager is exited the machine is taken down. + + When the machine is running it is also possible to access it from outside. + See the documentation for the module. + """ + + """ssh command-line arguments for executing commands""" + ssh_args: tuple[str | Path | tuple[str | Path, ...], ...] + + """ssh command-line arguments to be used to connect directly to the vsock""" + ssh_direct_args: tuple[str | Path | tuple[str | Path, ...], ...] | None + + # our three background tasks + _notify_server_task: asyncio.Task[None] | None = None + _ssh_control_task: asyncio.Task[None] | None = None + _qemu_task: asyncio.Task[None] | None = None + + def __init__( + self, + image: Path | str, + *, + ipc: Path, + attach_console: bool = False, + boot: Literal["efi", "mbr"] = "efi", + cloud_init_user_data: Mapping[str, object] | None = None, + credentials: Mapping[str, str] = {}, + cpus: int = 4, + identity: tuple[Path, str | None] | None = None, + identities: Sequence[str | Path] = (), + memory: int | str = "4G", + networks: Sequence[Network] = (), + sit: bool = False, + snapshot: bool = True, + status_messages: bool = False, + target: str = 'sockets.target', + timeout: float = 30.0, + ui: UI | None = None, + verbose: bool = False, + ) -> None: + """Construct a VM. + + The kwargs allow customizing the behaviour: + - attach_console: if qemu should connect the console to stdio + - boot: if we should boot with EFI or via the MBR + - cloud_init_user_data: JSON user-data for cloud-init + - credentials: extra system credentials + - cpus: the number of CPUs + - identity: a path to an ssh private key and the public key as a string. + If the public key is specified as None then it won't be configured on + the guest. The default (None) is to generate an ephemeral keypair. + - identities: extra private keys (either as paths on the disk or + directly as strings) to pass to ssh. Useful if you're not sure + which key the image will accept and want to try multiple. + - memory: how much memory the guest gets in MiB, or a string like "4G" + - networks: a list of Network objects (or empty to disable networking) + - sit: if we should "sit" when an exception occurs: print the exception + and wait for input (to allow inspecting the running VM) + - snapshot: if the 'snapshot' option is used on the disk (changes are + transient) + - status_messages: if we should do output of status messages (stderr) + - target: the name of the systemd target to wait for + - timeout: how long to wait for the VM to start, or 'inf' + - ui: a custom instance of UI (otherwise a new one is constructed per status_messages= and verbose=) + - verbose: if we should do output of verbose messages (stderr) + """ + self.image = image + self._ipc = ipc + self._attach_console = attach_console + self._boot = boot + self._cloud_init_user_data = cloud_init_user_data + self._cpus = cpus + self._credentials = credentials + self._identity = identity + self._identities = identities + self._memory = memory + self._networks = networks + self._sit = sit + self._snapshot = snapshot + self._target = target + self._timeout = timeout + self._ui = ui or UI(status_messages=status_messages, verbose=verbose) + + self._tasks = asyncio.TaskGroup() + self._ssh_control_ready = asyncio.Event() + self._qemu_exited = asyncio.Event() + self._shutdown_ok = False + + self.root = GuestPath("/", vm=self) + self.home = GuestPath(".", vm=self) + + async def _run_helper( + self, + *args: str | Path | tuple[str | Path, ...], + ready_when: Path, + ) -> asyncio.Task[None]: + """Run a helper process in the background. + + The function returns when the given named file (usually a socket) + appears. The process is awaited from our task group. It should never + exit. On cancellation, the process is terminated. + """ + process = await self._ui.spawn(*args) + + # block until we see the expected socket or an unexpected exit + while not ready_when.exists(): + with contextlib.suppress(TimeoutError): + returncode = await asyncio.wait_for(process.wait(), 0.01) + raise SubprocessError(args, returncode=returncode) + + async def wait_exit() -> None: + try: + returncode = await process.wait() # this should never return + raise SubprocessError(args, returncode=returncode) + except asyncio.CancelledError: + process.kill() + await process.wait() + + return self._tasks.create_task(wait_exit()) + + async def _ssh_keygen(self) -> tuple[Path, str]: + """Create a ssh key in the given directory. + + Returns the path to the private key and the public key as a string. + """ + private_key = self._ipc / "id" + + await self._ui.run( + "ssh-keygen", + "-q", # quiet + ("-t", "ed25519"), + ("-N", ""), # no passphrase + ("-C", ""), # no comment + ("-f", f"{private_key}"), + ) + + return private_key, (self._ipc / "id.pub").read_text().strip() + + def _sd_notify(self, line: str) -> None: + logger.debug("sd_notify:%s", line) + + # Only print target updates when ssh is offline + if self._ssh_control_task is not None: + return + + key, _, value = line.partition("=") + if key == "X_SYSTEMD_UNIT_ACTIVE": + self._ui.print_status(f"Reached target: {value}") + if value == self._target: + self._ssh_control_task = self._tasks.create_task(self._ssh_control()) + + elif key == "X_SYSTEMD_UNIT_INACTIVE": + self._ui.print_status(f"Unit inactive: {value}") + elif key == "X_SYSTEMD_SHUTDOWN": + self._ui.print_status(f"Shutdown: {value}") + + async def _sd_notify_connection(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + logger.debug("sd_notify connection") + try: + # Actually we should read until EOF but see + # https://github.com/rust-vmm/vhost-device/issues/874 + message = await reader.read(65536) + self._sd_notify(message.decode()) + finally: + writer.close() + await writer.wait_closed() + + async def _qemu( + self, + *, + port: int, + public: str | None, + ) -> None: + creds = { + **self._credentials, + "vmm.notify_socket": f"vsock-stream:2:{port}", + } + + if public is not None: + creds["ssh.ephemeral-authorized_keys-all"] = public + + snap = "on" if self._snapshot else "off" + drives = [f"file={self.image},format=qcow2,discard=unmap,cache=unsafe,aio=io_uring,snapshot={snap}"] + + if self._cloud_init_user_data: + cloud_init = self._ipc / "cloud-init" + cloud_init.mkdir() + (cloud_init / "meta-data").touch() + (cloud_init / "user-data").write_text("#cloud-config\n" + json.dumps(self._cloud_init_user_data) + "\n") + drives.append(f"driver=vvfat,dir={cloud_init},readonly=on,label=CIDATA") + + args = ( + _find_qemu(), + "-nodefaults", + ("-object", f"memory-backend-memfd,share=on,id=mem0,size={self._memory}"), + ("-bios", _find_ovmf()) if self._boot == "efi" else (), + ("-machine", "q35,accel=kvm,memory-backend=mem0"), + ("-cpu", "host"), + ("-smp", f"{self._cpus}"), + ("-m", f"{self._memory}"), + ("-display", "none"), + ("-qmp", f"unix:{self._ipc}/qmp,server,wait=off"), + ("-chardev", f"socket,id=vsock,reconnect=0,path={self._ipc}/vsock-device"), + ("-device", "vhost-user-vsock-pci,chardev=vsock"), + # Console stuff... + ("-device", "virtio-serial-pci"), + ("-chardev", f"socket,path={self._ipc}/vsock_1111,id=tt-notify,reconnect-ms=1"), + ("-device", "virtserialport,chardev=tt-notify,name=tt-notify"), + ("-serial", "chardev:console"), + *( + ( + ("-chardev", "stdio,mux=on,signal=off,id=console"), + ("-mon", "chardev=console,mode=readline"), + ) + if self._attach_console + else ( + # In the cases that the console isn't directed to stdio + # then we write it to a log file instead. Unfortunately, + # we also get a getty in our log file: + # https://github.com/systemd/systemd/issues/37928 + ("-chardev", f"file,path={self._ipc}/console,id=console"), + ) + ), + *(("-drive", f"{drive},if=virtio,media=disk") for drive in drives), + *(("-nic", net.to_qemu() + ",model=virtio-net-pci") for net in self._networks), + # Credentials + *(("-smbios", f"type=11,value=io.systemd.credential:{k}={v}") for k, v in creds.items()), + ) + + qemu = None + try: + self._ui.print_status("Waiting for guest") + qemu = await self._ui.spawn(*args) + returncode = await qemu.wait() + if not self._shutdown_ok: + raise SubprocessError(args, returncode) + except asyncio.CancelledError: + logger.debug("qemu task cancelled") + if qemu is not None: + logger.debug("Terminating qemu") + qemu.terminate() + try: + logger.debug("Waiting for qemu to quit") + await asyncio.shield(asyncio.wait_for(qemu.wait(), 5)) + except TimeoutError: + logger.debug("Timed out -- killing qemu") + qemu.kill() + await asyncio.shield(qemu.wait()) + finally: + logger.debug("qemu exited") + self._qemu_exited.set() + + async def _ssh_control(self) -> None: + ssh = None + try: + assert self.ssh_direct_args is not None + + self._ui.print_status("ssh control socket: Connecting") + + control_socket = self._ipc / "ssh" + + args = ( + "ssh", + *self.ssh_direct_args, + ("-N", "-n"), # no command, stdin disconnected + ("-M", "-S", control_socket), # listen on the control socket + self.get_id(), # unused, but shows up in messages + ) + ssh = await self._ui.spawn( + *args, + stdin=asyncio.subprocess.DEVNULL, + stdout=asyncio.subprocess.PIPE, + ) + + # ssh sends EOF after the connection succeeds + assert ssh.stdout + await ssh.stdout.read() + + # ..but that might have been because of an error, so check if the + # control socket actually exists + if not control_socket.exists(): + raise SubprocessError(args, await ssh.wait()) + + # we're online! + self._ssh_control_ready.set() + self._ui.print_status("ssh control socket: Connected") + self._ui.clear_status_message() + + returncode = await ssh.wait() + if not self._shutdown_ok: + raise SubprocessError(args, returncode) + except asyncio.CancelledError: + if ssh is not None: + ssh.terminate() + await asyncio.shield(ssh.wait()) + finally: + # We try to reset our state best as possible here to deal with + # reboots in the shutdown_ok case: we want the control socket + # reestablished when the machine comes back. + self._ssh_control_ready.clear() + self._ssh_control_task = None + + async def __aenter__(self) -> Self: + """Start the virtual machine.""" + await self._tasks.__aenter__() + + self.ssh_args = ( + ("-F", "none"), # don't use the user's config + ("-o", f"ControlPath={self._ipc}/ssh"), # connect via the control socket + ) + + try: + if self._identity is None: + self._identity = await self._ssh_keygen() + + private, public = self._identity + identities = [private] + + for nr, identity in enumerate(self._identities): + if isinstance(identity, str): + path = self._ipc / f"id.{nr}" + with path.open("x") as extra: + extra.write(identity) + path.chmod(0o400) + identities.append(path) + else: + identities.append(identity) + + self.ssh_direct_args = _ssh_direct_args(identities, self._ipc / "vsock") + + # It goes like this: + # - we start listening on the sd-notify socket + # - we start qemu + # - at some point the guest will notify that ssh is ready + # - this causes us to spawn the ssh control task + # - once connected, _ssh_control_ready gets set + # - we wait for that, so when it's done, we're done + + self._vsock_helper = await self._run_helper( + "vhost-device-vsock", + ("--socket", self._ipc / "vsock-device"), + ("--uds-path", self._ipc / "vsock"), + ready_when=self._ipc / "vsock-device", + ) + + server = await asyncio.start_unix_server(self._sd_notify_connection, self._ipc / "vsock_1111") + self._notify_server_task = self._tasks.create_task(server.serve_forever()) + + self._qemu_task = self._tasks.create_task(self._qemu(port=1111, public=public)) + + # the notify socket server will create the ssh control socket task + # which, in turn, sets this ready once it's online. + try: + await asyncio.wait_for(self._ssh_control_ready.wait(), self._timeout) + except TimeoutError as exc: + self._ui.clear_status_message() + + try: + log = (self._ipc / "console").read_text(errors="replace") + except FileNotFoundError: + log = "" + + # Remove ANSI escapes and weird/excessive line endings + log = re.sub(r"\033\[[0-9;]*[A-Za-z]", "", log) + log_lines = re.split(r"[\r\n]+", log) + + lines = [ + "Timed out waiting for the VM to start (after {self._timeout}s).", + "Console log:" if log_lines else "Console log unavailable.", + *log_lines, + ] + + raise TimeoutError("\n".join(lines) + "\n") from exc + + except: + exc_info = sys.exc_info() + logger.debug("Startup failed: %r", exc_info) + await self._tasks.__aexit__(*exc_info) + raise + else: + logger.debug("Startup was successful") + return self + finally: + logger.debug("Startup is done") + self._ui.clear_status_message() + + async def __aexit__( + self, + et: type[BaseException] | None, + exc: BaseException | None, + tb: TracebackType | None, + ) -> None: + """Stop the virtual machine.""" + self._shutdown_ok = True + + try: + if self._sit and exc is not None: + current = asyncio.current_task() + cancelled = current and current.cancelled() + + # This will report exceptions thrown from inside of the context managed + # block but won't work properly in case we were cancelled due to an + # error elsewhere (ie: qemu unexpectedly exiting) because at that point + # the main task will have been cancelled and we won't be able to block + # while waiting to read from stdin. That's ok: the goal if this is to + # catch test failures, not infra failures. + if not cancelled: + await self._ui.sit(exc=exc) + + if self._ssh_control_task is not None: + task = self._ssh_control_task + task.cancel() + await task # Cancellation is async, so wait + + # If we're in snapshot mode then we don't have to do a clean shutdown + with contextlib.suppress(FileNotFoundError): + await self.qmp("quit" if self._snapshot else "system_powerdown") + await self._qemu_exited.wait() + + assert self._vsock_helper is not None + self._vsock_helper.cancel() + + assert self._notify_server_task is not None + self._notify_server_task.cancel() + + finally: + # If the TaskGroup fails (possibly by way of 'exc') and shutdown + # also failed then the error from the shutdown will be suppressed. + await self._tasks.__aexit__(et, exc, tb) + if exc is not None: + raise exc + + def get_id(self) -> str: + """Get the machine identifier like `tt.0`, `tt.1`, etc.""" + return self._ipc.name + + async def wait_exit(self) -> None: + """Wait for the VM to exit.""" + self._shutdown_ok = True + await self._qemu_exited.wait() + + async def _ssh_cmd(self, *args: tuple[str | Path, ...]) -> None: + await self._ui.run("ssh", *self.ssh_args, *args, self.get_id()) + + async def forward_port(self, *args: tuple[str, ...]) -> None: + """Set up a port forward. + + The `spec` is the format used by `ssh -L`, and looks something like + `2222:127.0.0.1:22`. + """ + return await self._ssh_cmd(("-O", "forward"), *args) + + async def cancel_port(self, *args: tuple[str, ...]) -> None: + """Cancel a previous forward.""" + return await self._ssh_cmd(("-O", "cancel"), *args) + + async def scp(self, *args: str | pathlib.Path, direct: bool = False) -> None: + """Do a file transfer with scp. + + The hostname is ignored, so for paths on the guest use something like + `vm:/tmp/path`. + """ + assert self.ssh_direct_args is not None + await self._ui.run( + "scp", + *(self.ssh_direct_args if direct else self.ssh_args), + args, + ) + + async def wait_boot(self) -> None: + """Wait for the machine to be fully-booted.""" + await self.execute("systemctl", "is-system-running", "--wait") + + async def execute( + self, + cmd: str, + *args: str | GuestPath | tuple[str | GuestPath, ...], + check: bool = True, + direct: bool = False, + input: bytes | str | None = b"", # noqa:A002 # shadows `input()` but so does subprocess module + environment: Mapping[str, str] = {}, + stdout: int | None = asyncio.subprocess.PIPE, + ) -> str: + """Execute a command on the guest. + + If a single argument is given, it is expected to be a valid shell + script. If multiple arguments are given, they will interpreted as an + argument vector and will be properly quoted before being sent to the guest. + """ + if args: + cmd = shlex.join(itertools.chain(*_normalize_args(cmd, *args))) + + assert self.ssh_direct_args is not None + full_command = ( + "ssh", + *(self.ssh_direct_args if direct else self.ssh_args), + self.get_id(), # unused, but shows up in messages + ("--", "set -eu;"), + *(f"export {k}={v and shlex.quote(v)};" for k, v in environment.items()), + cmd, + ) + + ssh = await self._ui.spawn( + *full_command, + stdin=asyncio.subprocess.PIPE, + stdout=stdout, + ) + input_bytes = input.encode() if isinstance(input, str) else input + output, _ = await ssh.communicate(input_bytes) + returncode = await ssh.wait() + if check and returncode != 0: + raise SubprocessError(full_command, returncode, output) + return output.decode() if output is not None else "" + + async def write( + self, + dest: str | GuestPath, + content: str | bytes, + *, + mkdir: bool = True, + owner: str | tuple[str | int | None, str | int | None] | None = None, + perm: str | int | None = None, + ) -> None: + """Write a file into the test machine. + + Arguments: + dest: The file name in the machine to write to + content: Raw data to write to file + append: If True, append to existing file instead of replacing it + mkdir: if the parent directory should be created + owner: If set, call chown on the file with the given owner string + perm: Optional file permission as chmod shell string or integer + + """ + dest = GuestPath(dest, vm=self) + + if mkdir: + await dest.parent.mkdir(parents=True) + + if isinstance(content, str): + await dest.write_text(content) + else: + await dest.write_bytes(content) + + if owner is not None: + await dest.chown(owner) + + if perm: + await dest.chmod(perm) + + @contextlib.asynccontextmanager + async def disconnected(self) -> AsyncGenerator[None]: + """Temporarily disconnect the control socket. + + On enter, disconnect the ssh control socket from the guest system. + Inside of the block it's possible to perform commands that would + otherwise result in the control socket being destroyed (which would be + a hard error). + + On exit from the block, the connection is reestablished. + + The most obvious use-case for this is rebooting. + """ + assert self._ssh_control_task is not None + assert self._ssh_control_ready.is_set() + self._ssh_control_ready.clear() + self._ssh_control_task.cancel() + self._ssh_control_task = None + + try: + yield + finally: + await self._ssh_control_ready.wait() + + async def reboot(self) -> None: + """Reboot the guest, waiting until it's back online.""" + async with self.disconnected(): + await self.qmp("system_reset") + + async def qmp(self, command: str) -> object: + """Send a QMP command to the hypervisor. + + This can be used for things like modifying the hardware configuration. + Don't power it off this way: the correct way to stop the VM is to exit + the context manager. + """ + return await _qmp_command(self._ipc, command) + + +class SubprocessError(Exception): + """An exception thrown when a subprocess failed unexpectedly.""" + + def __init__( + self, + args: tuple[str | Path | tuple[str | Path, ...], ...], + returncode: int, + output: bytes | None = None, + ) -> None: + """Create a SubprocessError instance. + + - args: the arguments to the command that failed + - returncode: the non-zero return code + """ + self.args = args + self.returncode = returncode + self.output = output + + if returncode < 0: + msg = f"Subprocess terminated by {signal.Signals(-returncode).name}" + else: + msg = f"Subprocess exited unexpectedly with return code {returncode}:\n" + + super().__init__(f"{msg}\n{_pretty_print_args(*args)}\n\n") + + +def cleanup_on_signal() -> None: + """Register SIGHUP and SIGTERM signal handlers to cleanly exit. + + This raises an exception, cleaning up running subprocesses and the IPC + directory, in contrast to the default interpreter behaviour of a direct + exit. + """ + + def _term(*args: object) -> Never: + del args + # This raises SystemExit which will bubble out of the handler + sys.exit("I don't blame you.") + + signal.signal(signal.SIGHUP, _term) + signal.signal(signal.SIGTERM, _term) + + +@contextlib.contextmanager +def cli_helper() -> Iterator[None]: + """A context manager for using test.thing from CLI tools. + + This installs a signal handler for clean exit on SIGHUP and SIGTERM and + catches SubprocessError and TimeoutError, printing a message to stderr and + calling sys.exit().""" + cleanup_on_signal() + try: + yield + except* (SubprocessError, TimeoutError) as eg: + for exc in eg.exceptions: + sys.stderr.write(f"🤦 {exc}\n") + sys.exit("I'm sorry it didn't work out.") + + +def _main() -> None: # noqa: C901 + class AppendTuple(argparse.Action): + def __call__( + self, + parser: argparse.ArgumentParser, + namespace: argparse.Namespace, + values: str | Sequence[Any] | None, + option_string: str | None = None, + ) -> None: + del parser + fwds = getattr(namespace, self.dest) or () + fwds = (*fwds, (option_string, values)) + setattr(namespace, self.dest, fwds) + + parser = argparse.ArgumentParser(description="test.thing - a simple modern VM runner") + parser.add_argument("--maintain", "-m", action="store_true", help="Changes are permanent") + parser.add_argument("--attach", "-a", action="store_true", help="Attach to the VM console") + parser.add_argument("--sit", action="store_true", help="Wait for enter key on exceptions") + parser.add_argument("--debug", "-d", action="store_true", help="Enable debug output") + parser.add_argument("--cloud-init", action="store_true", help="Add Cockpit ssh key via cloud-init") + parser.add_argument( + "--boot", + choices=("efi", "mbr"), + default="efi", + help="How to boot the image (default: efi)", + ) + parser.add_argument("--verbose", "-v", action="store_true", help="Print verbose output") + parser.add_argument( + "--ssh-key", + "-i", + type=Path, + action="append", + help="Path to SSH private key (default: generate)", + ) + parser.add_argument("--timeout", type=float, help="For startup, in seconds, or 'inf' (default: 30)") + parser.add_argument("--no-network", action="store_true", help="Isolate the VM from the Internet") + parser.add_argument( + "-L", + "-R", + "-D", + default=[], + dest="fwd_spec", + action=AppendTuple, + help="Setup an SSH-style port forward", + ) + parser.add_argument("--no-wait", action="store_true", help="Don't wait for ENTER before exiting") + parser.add_argument( + "--script", + "-c", + metavar="COMMAND", + action="append", + help="Execute this (shell-interpreted) command", + ) + parser.add_argument( + "--start-unit", + "-s", + metavar="UNIT", + action="append", + dest="script", + type=(lambda s: f"systemctl enable --now {s!s}"), + help="Start this systemd unit", + ) + + parser.add_argument("image", type=Path, help="The path to a qcow2 VM image to run") + args = parser.parse_args() + + if args.cloud_init: + user_data = {"users": [{"name": "root", "ssh_authorized_keys": [COCKPIT_TEST_IDENTITY_PUB]}]} + else: + user_data = None + + async def _async_main() -> None: + with cli_helper(), IpcDirectory() as ipc: + async with VirtualMachine( + args.image, + ipc=ipc, + attach_console=args.attach, + boot=args.boot, + cloud_init_user_data=user_data, + identities=args.ssh_key or (COCKPIT_TEST_IDENTITY,), + networks=(() if args.no_network else (Network.user(),)), + sit=args.sit, + snapshot=not args.maintain, + status_messages=not args.attach, + timeout=args.timeout, + verbose=args.verbose, + ) as vm: + for spec in args.fwd_spec: + await vm.forward_port(spec) + + for cmd in args.script or (): + await vm.execute(cmd, stdout=None) + + if args.attach: + await vm.wait_exit() + elif not args.no_wait: + await vm._ui.sit(f"VM running: {vm.get_id()}") + + if args.debug: + logging.basicConfig(level=logging.DEBUG) + + asyncio.run(_async_main(), debug=args.debug) + + +if __name__ == "__main__": + _main() diff --git a/pyproject.toml b/pyproject.toml index de28854bcf..9e7e509824 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -38,6 +38,7 @@ module = [ [tool.ruff] exclude = [ + "lib/testthing.py", ".git/", ] line-length = 118 diff --git a/test/run b/test/run index 97fc396462..edf87f0a32 100755 --- a/test/run +++ b/test/run @@ -17,6 +17,6 @@ find_python_files() { find_scripts 'python3' '*.py' } -find_python_files | xargs -0 ruff check --quiet +find_python_files | xargs -0 ruff check --quiet --force-exclude find_python_files | xargs -0 mypy --no-error-summary pytest -vv From 4b34f663473b511d9ba3574798d05bb93d3b1c4a Mon Sep 17 00:00:00 2001 From: Allison Karlitskaya Date: Fri, 12 Sep 2025 14:25:11 +0200 Subject: [PATCH 2/6] image-diff: rewrite against test.thing This benefits from the fact that test.thing is async: instead of the very non-DRY parallel pattern for booting machines, waiting on them, etc. we can just do everything in a single simple async function and use asyncio.gather() to run both instances at once. --- image-diff | 113 +++++++++++++++++++++++++---------------------------- 1 file changed, 53 insertions(+), 60 deletions(-) diff --git a/image-diff b/image-diff index 5a36076030..570ffbb047 100755 --- a/image-diff +++ b/image-diff @@ -1,77 +1,70 @@ -#!/usr/bin/python3 - -# This file is part of Cockpit. -# -# Copyright (C) 2020 Red Hat, Inc. -# -# Cockpit is free software; you can redistribute it and/or modify it -# under the terms of the GNU Lesser General Public License as published by -# the Free Software Foundation; either version 2.1 of the License, or -# (at your option) any later version. -# -# Cockpit is distributed in the hope that it will be useful, but -# WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU -# Lesser General Public License for more details. -# -# You should have received a copy of the GNU Lesser General Public License -# along with Cockpit; If not, see . +#!/usr/bin/env python3 +# SPDX-License-Identifier: GPL-3.0-or-later + +"""image-diff - Compare package versions on VM images. + +Gives an overview of which packages have been added/removed/changed between two +versions of a VM image. +""" import argparse +import asyncio from collections.abc import Mapping +from pathlib import Path + +from lib import testthing +from lib.constants import IMAGES_DIR -from machine import testvm +async def get_packages(image: str) -> Mapping[str, str]: + file = Path(image) if image.startswith("/") else Path(IMAGES_DIR, image) -def get_packages(machine: testvm.VirtMachine) -> Mapping[str, str]: - # List all packages, irrespective of package manager (rpm or dpkg or pacman). - # If both are missing, then the command will fail. - # - # We'd ideally like to get source packages everywhere, but it's a - # bit more difficult on RPM. (TODO) - pkgcmd = """if type dpkg-query > /dev/null 2>&1; then - dpkg-query -W 2>/dev/null; - elif type rpm > /dev/null 2>&1; then - rpm -qa --qf '%{NAME}\t%{EVR}\n' 2>/dev/null; - else pacman -Q | sed 's/ /\t/' 2>/dev/null; fi""" + with testthing.IpcDirectory() as ipc: + async with testthing.VirtualMachine(file, boot="mbr", ipc=ipc) as vm: + # List all packages, irrespective of package manager (rpm or dpkg + # or pacman). If both are missing, then the command will fail. + # + # We'd ideally like to get source packages everywhere, but it's a + # bit more difficult on RPM. (TODO) + pkgcmd = """if type dpkg-query > /dev/null 2>&1; then + dpkg-query -W 2>/dev/null; + elif type rpm > /dev/null 2>&1; then + rpm -qa --qf '%{NAME}\t%{EVR}\n' 2>/dev/null; + else pacman -Q | sed 's/ /\t/' 2>/dev/null; fi""" - output = machine.execute(pkgcmd).strip() - return dict(line.split('\t') for line in output.splitlines()) + output = (await vm.execute(pkgcmd)).strip() + return dict(line.split("\t") for line in output.splitlines()) -parser = argparse.ArgumentParser(description='Compare package versions on VM images') -parser.add_argument('old', help='the "old" image to compare') -parser.add_argument('new', help='the "new" image to compare') -args = parser.parse_args() +async def diff(old: str, new: str) -> None: + old_pkgs, new_pkgs = await asyncio.gather(get_packages(old), get_packages(new)) -# boot the machines in parallel -old_vm = testvm.VirtMachine(image=args.old) -new_vm = testvm.VirtMachine(image=args.new) + print("Removed:") + for name in sorted(set(old_pkgs) - set(new_pkgs)): + print(f" {name} ({old_pkgs[name]})") + print() -old_vm.start() -new_vm.start() + print("Added:") + for name in sorted(set(new_pkgs) - set(old_pkgs)): + print(f" {name} ({new_pkgs[name]})") + print() -old_vm.wait_boot() -new_vm.wait_boot() + print("Changed:") + for name in sorted(set.intersection(set(new_pkgs), set(old_pkgs))): + if new_pkgs[name] != old_pkgs[name]: + print(f" {name} ({old_pkgs[name]} -> {new_pkgs[name]})") + print() -old_pkgs = get_packages(old_vm) -new_pkgs = get_packages(new_vm) -old_vm.kill() -new_vm.kill() +def main() -> None: + parser = argparse.ArgumentParser(description="Compare package versions on VM images") + parser.add_argument("old", help='the "old" image to compare') + parser.add_argument("new", help='the "new" image to compare') + args = parser.parse_args() -print('Removed:') -for name in sorted(set(old_pkgs) - set(new_pkgs)): - print(f' {name} ({old_pkgs[name]})') -print() + with testthing.cli_helper(): + asyncio.run(diff(args.old, args.new)) -print('Added:') -for name in sorted(set(new_pkgs) - set(old_pkgs)): - print(f' {name} ({new_pkgs[name]})') -print() -print('Changed:') -for name in sorted(set.intersection(set(new_pkgs), set(old_pkgs))): - if new_pkgs[name] != old_pkgs[name]: - print(f' {name} ({old_pkgs[name]} -> {new_pkgs[name]})') -print() +if __name__ == "__main__": + main() From 6c00eb9039cbc8f2881a23ec24967498b1965c84 Mon Sep 17 00:00:00 2001 From: Allison Karlitskaya Date: Fri, 12 Sep 2025 14:27:23 +0200 Subject: [PATCH 3/6] image-customize: port to test.thing This is mostly boring. There is really only one nice cleanup enabled by everything being async: instead of repeating `timeout=1800` everywhere, we can just apply it at the toplevel with a single asyncio.wait_for() invocation. Even though it's not a rewrite, apply a bit of `ruff format` here. --- image-customize | 222 ++++++++++++++++++++++++++---------------------- 1 file changed, 122 insertions(+), 100 deletions(-) diff --git a/image-customize b/image-customize index 6f3be93924..c15fa42ee2 100755 --- a/image-customize +++ b/image-customize @@ -17,17 +17,19 @@ # along with Cockpit; If not, see . import argparse +import asyncio import os import subprocess -import sys from collections.abc import Sequence +from pathlib import Path -from lib.constants import BOTS_DIR, TEST_DIR -from machine import testvm +from lib import testthing +from lib.constants import BOTS_DIR, IMAGES_DIR, MACHINE_DIR, TEST_DIR +from lib.testmap import get_test_image opt_quick: bool = False opt_verbose: bool = False -opt_build_options: str = '' +opt_build_options: str = "" stdout_disposition: int | None = None @@ -35,7 +37,7 @@ def prepare_install_image(base_image: str, install_image: str, resize: str | Non """Create the necessary layered image for the build/install""" if "/" not in base_image: - base_image = os.path.join(testvm.IMAGES_DIR, base_image) + base_image = os.path.join(IMAGES_DIR, base_image) if "/" not in install_image: install_image = os.path.join(os.path.join(TEST_DIR, "images"), os.path.basename(install_image)) @@ -54,8 +56,16 @@ def prepare_install_image(base_image: str, install_image: str, resize: str | Non install_image_dir = os.path.dirname(install_image) os.makedirs(install_image_dir, exist_ok=True) base_image = os.path.realpath(base_image) - subprocess.check_call(["qemu-img", "create", "-q", "-f", "qcow2", - "-o", f"backing_file={base_image},backing_fmt=qcow2", qcow2_image]) + subprocess.check_call([ + "qemu-img", + "create", + "-q", + "-f", + "qcow2", + "-o", + f"backing_file={base_image},backing_fmt=qcow2", + qcow2_image, + ]) if os.path.lexists(install_image): os.unlink(install_image) os.symlink(os.path.basename(qcow2_image), install_image) @@ -70,7 +80,7 @@ class ActionBase(argparse.Action): """Keep an ordered list of actions""" @staticmethod - def execute(machine_instance: testvm.Machine, argument: str) -> None: + async def execute(machine_instance: testthing.VirtualMachine, argument: str, /) -> None: raise NotImplementedError def __call__( @@ -78,68 +88,76 @@ class ActionBase(argparse.Action): parser: argparse.ArgumentParser, namespace: argparse.Namespace, value: str | Sequence[str] | None, - option_string: object = None + option_string: object = None, ) -> None: getattr(namespace, self.dest).append((self.execute, value)) class InstallAction(ActionBase): """Install local rpm or distro package""" + @staticmethod - def execute(machine_instance: testvm.Machine, package: str) -> None: + async def execute(machine_instance: testthing.VirtualMachine, package: str) -> None: # If we have a '/' in the package name, or if a file with that name # exists in the current directory, then assume that this is a package # we're uploading from the host. - if '/' in package or os.path.isfile(package): + if "/" in package or os.path.isfile(package): dest = "/var/tmp/" + os.path.basename(package) - machine_instance.upload([os.path.abspath(package)], dest) + await machine_instance.scp(os.path.abspath(package), f"vm:{dest}") package = dest # requesting install of Python wheel? - if package.endswith('.whl'): - machine_instance.execute(f"python3 -m pip install --no-index --prefix=/usr/local {package}", timeout=120) + if package.endswith(".whl"): + await asyncio.wait_for( + machine_instance.execute(f"python3 -m pip install --no-index --prefix=/usr/local {package}"), + timeout=120, + ) return # this will fail if neither is available -- exception is clear enough, this is a developer tool - out = machine_instance.execute("which dnf || which yum || which apt-get") - if 'dnf' in out: + out = await machine_instance.execute("which dnf || which yum || which apt-get || which pacman") + if "dnf" in out: install_command = "dnf install -y" - elif 'yum' in out: + elif "yum" in out: install_command = "yum --setopt=skip_missing_names_on_install=False -y install" - else: + elif "apt-get" in out: install_command = "apt-get install -y" + elif "pacman" in out: + install_command = "pacman -S --noconfirm" + else: + raise NotImplementedError(f"unknown build platform: {out}") - machine_instance.execute(f"{install_command} {package}", timeout=1800) + await machine_instance.execute(f"{install_command} {package}") class BuildAction(ActionBase): """Build and install distribution package(s) from dist tarball or source RPM""" @staticmethod - def execute(machine_instance: testvm.Machine, source: str) -> None: + async def execute(machine_instance: testthing.VirtualMachine, source: str) -> None: # upload the tarball or srpm sourcename = os.path.basename(source) vm_source = os.path.join("/var/tmp", sourcename) - machine_instance.upload([source], vm_source, relative_dir=".") + await machine_instance.scp(source, f"vm:{vm_source}") # this will fail if neither is available -- exception is clear enough, this is a developer tool - out = machine_instance.execute("(which pbuilder || which mock || which pacman) 2>/dev/null") - if 'pbuilder' in out: - BuildAction.build_deb(machine_instance, vm_source) - elif 'mock' in out: - BuildAction.build_rpm(machine_instance, vm_source) - elif 'pacman' in out: - BuildAction.build_arch(machine_instance, vm_source) + out = await machine_instance.execute("(which pbuilder || which mock || which pacman) 2>/dev/null") + if "pbuilder" in out: + await BuildAction.build_deb(machine_instance, vm_source) + elif "mock" in out: + await BuildAction.build_rpm(machine_instance, vm_source) + elif "pacman" in out: + await BuildAction.build_arch(machine_instance, vm_source) else: raise NotImplementedError(f"unknown build platform: {out}") @staticmethod - def build_deb(machine: testvm.Machine, vm_source: str) -> None: - build_opts = 'nocheck' if opt_quick else '' + async def build_deb(machine: testthing.VirtualMachine, vm_source: str) -> None: + build_opts = "nocheck" if opt_quick else "" # build source packge - machine.execute(f""" + await machine.execute(f""" set -eu rm -rf /var/tmp/build mkdir -p /var/tmp/build @@ -156,27 +174,30 @@ class BuildAction(ActionBase): dpkg-buildpackage -S -us -uc -nc""") # build binary packages - machine.execute(f"cd /var/tmp/build; DEB_BUILD_OPTIONS='{build_opts}' pbuilder build --buildresult . " - f"{opt_build_options} *.dsc", timeout=1800, stdout=stdout_disposition) + await machine.execute( + f"cd /var/tmp/build; DEB_BUILD_OPTIONS='{build_opts}' pbuilder build --buildresult . " + f"{opt_build_options} *.dsc", + stdout=stdout_disposition, + ) # install packages - machine.execute("dpkg -i /var/tmp/build/*.deb") + await machine.execute("dpkg -i /var/tmp/build/*.deb") @staticmethod - def build_rpm(machine: testvm.Machine, vm_source: str) -> None: - mock_opts = '' + async def build_rpm(machine: testthing.VirtualMachine, vm_source: str) -> None: + mock_opts = "" if opt_verbose: - mock_opts += ' --verbose' + mock_opts += " --verbose" if opt_quick: - mock_opts += ' --nocheck' + mock_opts += " --nocheck" if opt_build_options: - mock_opts += ' ' + opt_build_options + mock_opts += " " + opt_build_options # build source package, unless this is running against an srpm already if vm_source.endswith(".src.rpm"): srpm = vm_source else: - machine.execute(f""" + await machine.execute(f""" set -eu rm -rf /var/tmp/build su builder -c 'rpmbuild --define "_topdir /var/tmp/build" -ts "{vm_source}"' @@ -184,7 +205,7 @@ class BuildAction(ActionBase): srpm = "/var/tmp/build/SRPMS/*.src.rpm" # HACK: mock in openSUSE must be called through sudo (but still originally as builder) - if "opensuse" in machine.execute("cat /etc/os-release"): + if "opensuse" in await machine.execute("cat /etc/os-release"): mock_sudo = "sudo" else: mock_sudo = "" @@ -192,19 +213,25 @@ class BuildAction(ActionBase): # build binary RPMs from srpm; disable all repositorys as mock insists on # calling `dnf builddep`, which insists on a cache; our test VMs don't have a cache, # as the mock is offline and pre-installed - machine.execute(f"su builder -c '{mock_sudo} mock --no-clean --no-cleanup-after --disablerepo=* " - f"--offline --resultdir /var/tmp/build {mock_opts} --rebuild {srpm}'", - timeout=1800, stdout=stdout_disposition) + await machine.execute( + f"su builder -c '{mock_sudo} mock --no-clean --no-cleanup-after --disablerepo=* " + f"--offline --resultdir /var/tmp/build {mock_opts} --rebuild {srpm}'", + stdout=stdout_disposition, + ) # install RPMs - machine.execute('packages=$(find /var/tmp/build -name "*.rpm" -not -name "*.src.rpm"); ' - f'rpm -U --force --verbose {"--nodigest --nosignature" if opt_quick else ""} $packages') + await machine.execute( + 'packages=$(find /var/tmp/build -name "*.rpm" -not -name "*.src.rpm"); ' + f"rpm -U --force --verbose {'--nodigest --nosignature' if opt_quick else ''} $packages" + ) @staticmethod - def build_arch(machine: testvm.Machine, vm_source: str) -> None: + async def build_arch(machine: testthing.VirtualMachine, vm_source: str) -> None: # unpack source tree's arch packaging directory (PKGBUILD refers to some files) # and set PKGBUILD variables - machine.write("/var/tmp/mkbuild.sh", f"""#!/bin/sh + await machine.write( + "/var/tmp/mkbuild.sh", + f"""#!/bin/sh set -eu rm -rf /var/tmp/build mkdir -p /var/tmp/build @@ -215,52 +242,49 @@ class BuildAction(ActionBase): cp "$archdir"/* . # tarball must be in same directory as PKGBUILD cp '{vm_source}' /var/tmp/build/ - """, perm="755") - machine.execute(f"su builder {opt_build_options} /var/tmp/mkbuild.sh") + """, + perm="755", + ) + await machine.execute(f"su builder {opt_build_options} /var/tmp/mkbuild.sh") # build binaries - machine.execute("cd /var/tmp/build; makechrootpkg -r /var/lib/archbuild/cockpit -U builder", - timeout=1800, stdout=stdout_disposition) + await machine.execute( + "cd /var/tmp/build; makechrootpkg -r /var/lib/archbuild/cockpit -U builder", + stdout=stdout_disposition, + ) # install packages - machine.execute("pacman -U --noconfirm /var/tmp/build/*.pkg.tar.zst") + await machine.execute("pacman -U --noconfirm /var/tmp/build/*.pkg.tar.zst") class RunCommandAction(ActionBase): @staticmethod - def execute(machine_instance: testvm.Machine, command: str) -> None: - try: - machine_instance.execute(command, timeout=1800) - except subprocess.CalledProcessError as e: - sys.stderr.write("%s\n" % e) - sys.exit(e.returncode) + async def execute(machine_instance: testthing.VirtualMachine, command: str) -> None: + await machine_instance.execute(command) class ScriptAction(ActionBase): @staticmethod - def execute(machine_instance: testvm.Machine, script: str) -> None: + async def execute(machine_instance: testthing.VirtualMachine, script: str) -> None: uploadpath = "/var/tmp/" + os.path.basename(script) - machine_instance.upload([os.path.abspath(script)], uploadpath) - machine_instance.execute("chmod a+x %s" % uploadpath) - try: - machine_instance.execute(uploadpath, timeout=1800) - except subprocess.CalledProcessError as e: - sys.stderr.write("%s\n" % e) - sys.exit(e.returncode) + await machine_instance.scp(os.path.abspath(script), f"vm:{uploadpath}") + await machine_instance.execute("chmod a+x %s" % uploadpath) + await machine_instance.execute(uploadpath) class UploadAction(ActionBase): @staticmethod - def execute(machine_instance: testvm.Machine, srcdest: str) -> None: + async def execute(machine_instance: testthing.VirtualMachine, srcdest: str) -> None: src, dest = srcdest.split(":") abssrc = os.path.abspath(src) # preserve trailing / for rsync compatibility - if src.endswith('/'): - abssrc += '/' - machine_instance.upload([abssrc], dest) + if src.endswith("/"): + abssrc += "/" + await machine_instance.scp(abssrc, f"vm:{dest}") -def main() -> None: +async def main() -> None: + # fmt: off parser = argparse.ArgumentParser( description=('Run command inside or install packages into a Cockpit virtual machine. ' 'All actions can be specified multiple times and run in the given order.')) @@ -284,7 +308,7 @@ def main() -> None: help="Additional options for mock/pbuilder/arch builder") parser.add_argument('--resize', help="Resize the image. Size in bytes with using K, M, or G suffix.") parser.add_argument('-n', '--no-network', action='store_true', help='Do not connect the machine to the Internet') - parser.add_argument('--cpus', type=int, default=None, + parser.add_argument('--cpus', type=int, default=2, help="Number of CPUs for the virtual machine") parser.add_argument('--memory-mb', type=int, default=2048, help="RAM size for the virtual machine") @@ -294,6 +318,7 @@ def main() -> None: help='Disable tests during package build with --build') parser.add_argument('image', help='The image to use (destination name when using --base-image)') parser.add_argument('--sit', action='store_true', help='Sit and wait if any VM action fails') + # fmt: on args = parser.parse_args() if not args.actions and not args.resize: @@ -302,7 +327,7 @@ def main() -> None: if not args.base_image: args.base_image = os.path.basename(args.image) - args.base_image = testvm.get_test_image(args.base_image) + args.base_image = get_test_image(args.base_image) global opt_quick, opt_verbose, opt_build_options, stdout_disposition opt_quick = args.quick @@ -311,30 +336,27 @@ def main() -> None: if not args.verbose: stdout_disposition = subprocess.DEVNULL - if '/' not in args.base_image: + if "/" not in args.base_image: subprocess.check_call([os.path.join(BOTS_DIR, "image-download"), args.base_image]) - network = testvm.VirtNetwork(0, image=args.base_image) - machine = testvm.VirtMachine(maintain=True, - verbose=args.verbose, - networking=network.host(restrict=args.no_network), - image=prepare_install_image(args.base_image, args.image, args.resize, args.fresh), - cpus=args.cpus, - memory_mb=args.memory_mb) - machine.start() - machine.wait_boot() - try: - for (handler, arg) in args.actions: - handler(machine, arg) - except Exception as e: - if args.sit: - print(e, file=sys.stderr) - print(machine.diagnose(), file=sys.stderr) - print("Press RET to continue...") - sys.stdin.readline() - raise e - finally: - machine.stop() - - -if __name__ == '__main__': - main() + + cockpit_test_identity = Path(MACHINE_DIR) / "identity" + cockpit_test_identity.chmod(0o600) + + with testthing.cli_helper(), testthing.IpcDirectory() as ipc: + async with testthing.VirtualMachine( + prepare_install_image(args.base_image, args.image, args.resize, args.fresh), + cpus=args.cpus, + identity=(cockpit_test_identity, None), + ipc=ipc, + memory=f"{args.memory_mb}M", + networks=[] if args.no_network else [testthing.Network("user")], + sit=args.sit, + snapshot=False, + verbose=args.verbose, + ) as vm: + for handler, arg in args.actions: + await asyncio.wait_for(handler(vm, arg), timeout=1800) + + +if __name__ == "__main__": + asyncio.run(main()) From ac782b19edd5854918e2068bc9ce4025bfdbd0f8 Mon Sep 17 00:00:00 2001 From: Allison Karlitskaya Date: Sat, 13 Sep 2025 09:35:54 +0200 Subject: [PATCH 4/6] good --- image-customize | 1 + image-diff | 2 +- lib/testthing.py | 14 ++++---------- 3 files changed, 6 insertions(+), 11 deletions(-) diff --git a/image-customize b/image-customize index c15fa42ee2..d23ea6c98d 100755 --- a/image-customize +++ b/image-customize @@ -345,6 +345,7 @@ async def main() -> None: with testthing.cli_helper(), testthing.IpcDirectory() as ipc: async with testthing.VirtualMachine( prepare_install_image(args.base_image, args.image, args.resize, args.fresh), + boot="mbr", cpus=args.cpus, identity=(cockpit_test_identity, None), ipc=ipc, diff --git a/image-diff b/image-diff index 570ffbb047..1b5a9e57b2 100755 --- a/image-diff +++ b/image-diff @@ -20,7 +20,7 @@ async def get_packages(image: str) -> Mapping[str, str]: file = Path(image) if image.startswith("/") else Path(IMAGES_DIR, image) with testthing.IpcDirectory() as ipc: - async with testthing.VirtualMachine(file, boot="mbr", ipc=ipc) as vm: + async with testthing.VirtualMachine(file, boot="mbr", ipc=ipc, memory="1G") as vm: # List all packages, irrespective of package manager (rpm or dpkg # or pacman). If both are missing, then the command will fail. # diff --git a/lib/testthing.py b/lib/testthing.py index 6990027cfd..eb7808ffca 100755 --- a/lib/testthing.py +++ b/lib/testthing.py @@ -344,7 +344,7 @@ async def sit(self, msg: str | None = None, exc: BaseException | None = None) -> async def spawn( self, *args: str | Path | tuple[str | Path, ...], - stdin: int | None = None, + stdin: int | None = asyncio.subprocess.DEVNULL, stdout: int | None = None, stderr: int | None = None, ) -> asyncio.subprocess.Process: @@ -395,7 +395,6 @@ def pr_set_pdeathsig() -> None: *itertools.chain(*_normalize_args(*args)), stdin=stdin, stdout=stdout, - stderr=stderr, preexec_fn=pr_set_pdeathsig, ) @@ -406,15 +405,10 @@ async def run( ) -> int: """Run a process, waiting for it to exit. - This takes the same arguments as _spawn, plus a "check" argument (True by + This takes the same arguments as spawn, plus a "check" argument (True by default) which works in the usual way. """ - process = await self.spawn( - *args, - stdin=asyncio.subprocess.DEVNULL, - stdout=None if self._verbose else asyncio.subprocess.DEVNULL, - stderr=None if self._verbose else asyncio.subprocess.DEVNULL, - ) + process = await self.spawn(*args, stdin=asyncio.subprocess.DEVNULL) returncode = await process.wait() if check and returncode != 0: raise SubprocessError(args, returncode=returncode) @@ -725,7 +719,7 @@ async def _qemu( creds["ssh.ephemeral-authorized_keys-all"] = public snap = "on" if self._snapshot else "off" - drives = [f"file={self.image},format=qcow2,discard=unmap,cache=unsafe,aio=io_uring,snapshot={snap}"] + drives = [f"file={self.image},format=qcow2,discard=unmap,snapshot={snap}"] if self._cloud_init_user_data: cloud_init = self._ipc / "cloud-init" From 4c629ad33495e3bbef009ecfa89d0f87b056ea20 Mon Sep 17 00:00:00 2001 From: Allison Karlitskaya Date: Sun, 14 Sep 2025 19:31:15 +0200 Subject: [PATCH 5/6] scp -r --- image-customize | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/image-customize b/image-customize index d23ea6c98d..3f5f87e131 100755 --- a/image-customize +++ b/image-customize @@ -280,7 +280,7 @@ class UploadAction(ActionBase): # preserve trailing / for rsync compatibility if src.endswith("/"): abssrc += "/" - await machine_instance.scp(abssrc, f"vm:{dest}") + await machine_instance.scp("-r", abssrc, f"vm:{dest}") async def main() -> None: From 2bb4a149b706f79204fb4371a49b8c85ae26e834 Mon Sep 17 00:00:00 2001 From: Allison Karlitskaya Date: Fri, 23 Jan 2026 09:38:03 +0100 Subject: [PATCH 6/6] all images are working --- image-create | 187 ++++- images/scripts/arch.setup | 5 - images/scripts/bootc.setup | 14 - .../lib/test.thing-workarounds/sshd_vsock.te | 17 - .../tt-sd_notify.service | 19 - .../tt-sshd-vsock.socket | 10 - .../tt-sshd-vsock@.service | 13 - images/scripts/opensuse-tumbleweed.setup | 4 - images/scripts/rhel.setup | 10 - lib/testthing.py | 740 ++++++++++++------ 10 files changed, 647 insertions(+), 372 deletions(-) delete mode 100644 images/scripts/lib/test.thing-workarounds/sshd_vsock.te delete mode 100644 images/scripts/lib/test.thing-workarounds/tt-sd_notify.service delete mode 100644 images/scripts/lib/test.thing-workarounds/tt-sshd-vsock.socket delete mode 100644 images/scripts/lib/test.thing-workarounds/tt-sshd-vsock@.service diff --git a/image-create b/image-create index 32cd4803b2..7051b364f3 100755 --- a/image-create +++ b/image-create @@ -9,9 +9,11 @@ it to be useable with vm-run and testlib.py. import argparse import asyncio +import contextlib import hashlib import os import tempfile +from collections.abc import Mapping from pathlib import Path from lib import testthing @@ -19,24 +21,99 @@ from lib.aio.jsonutil import JsonObject from lib.constants import IMAGES_DIR, MACHINE_DIR, SCRIPTS_DIR from lib.directories import get_images_data_dir +WORKAROUNDS = Path(SCRIPTS_DIR) / 'test.thing-workarounds' + + +def format_journal_line(entry: Mapping[str, str]) -> str: + def one_of(*keys: str, default: str) -> str: + return next(filter(None, (entry.get(k) for k in keys)), default) + + stamp = one_of( + "_SOURCE_BOOTTIME_TIMESTAMP", + "_SOURCE_MONOTONIC_TIMESTAMP", + "__MONOTONIC_TIMESTAMP", + default="0", + ) + if entry["_TRANSPORT"].lower() == "kernel": + ident = "kernel" + else: + name = one_of("SYSLOG_IDENTIIFIER", "_COMM", "_EXE", default="?") + pid = one_of("SYSLOG_PID", "_PID", default="?") + ident = f"{name}[{pid}]" + + return f"[{float(stamp) / 1_000_000:+.6f}] {ident}: {entry['MESSAGE']}" + + +def print_journal_line(entry: Mapping[str, str]) -> None: + if entry.get('PRIORITY', '9') in '012345678' or 'ssh' in entry.get('SYSLOG_IDENTIFIER', ""): # 4 = "Warning" + with contextlib.suppress(OSError): + print(format_journal_line(entry)) + + +def cloud_init_config(image: str) -> JsonObject | None: + def extra_file(dest: str) -> JsonObject: + source = WORKAROUNDS / Path(dest).name + return { + "path": dest, + "content": source.read_text(), + "permissions": "0755" if os.access(source, os.X_OK) else "0644", + } + + # don't cloud-init images that we already used ignition for + if image in ['centos-9-bootc', 'fedora-coreos', 'services']: + return None -def cloud_init_config(image: str) -> JsonObject: - vm_pass = "$5$CockpitCloudInit$Iw89f.aPgqHPXAHC2Zs9h9335n3E1FQDFvR6MLqwPK9" ssh_authorized_keys = [Path(MACHINE_DIR, "identity.pub").read_text()] host_key = Path(MACHINE_DIR, "host_key").read_text() host_key_pub = Path(MACHINE_DIR, "host_key.pub").read_text() bootcmd = [] - runcmd = ["printf X_SYSTEMD_UNIT_ACTIVE=los-gehts > /dev/virtio-ports/tt-notify"] - - if image in ["arch", "centos-9-bootc"]: - bootcmd.append("echo 100::55:4e4b:4e4f:574e UNKNOWN >> /etc/hosts") - - if image in ["arch"]: - bootcmd.append("systemctl disable --now systemd-time-wait-sync.service || true") - - if image in ["opensuse-tumbleweed"]: - bootcmd.append("echo 'Include /usr/etc/ssh/sshd_config' > /etc/ssh/sshd_config") + write_files = {"/usr/bin/vsock-fling"} # weeeee + runcmd = [] + + # TODO: filter this? + runcmd.append("echo 100::55:4e4b:4e4f:574e UNKNOWN >> /etc/hosts") + + if image == 'arch': + bootcmd.append("systemctl disable --now systemd-time-wait-sync.service") + + openrc = image.startswith('alpine') + if openrc: + # ssh-vsock support for OpenRC + write_files.add("/usr/local/sbin/tt-vsock-sshd") + write_files.add("/etc/init.d/tt-vsock-ssh-service") + runcmd.append("rc-update add tt-vsock-ssh-service default") + runcmd.append("rc-service tt-vsock-ssh-service start") + + # do this one last: it notifies that we're done + runcmd.append("vsock-fling 2 1111 X_SYSTEMD_UNIT_ACTIVE=multi-user.target") + + else: + # systemd + + if image.startswith(("centos-9-", "rhel-8-", "rhel-9-", "ubuntu-22", "ubuntu-24")): + # no `vmm.notify_socket`, `systemd-ssh-generator` `ssh-vsock.socket` support + + if image.startswith(("centos-", "rhel-")): + # SELinux vsock fixes: https://issues.redhat.com/browse/RHEL-113647 + # We need to apply these *before* we can start the socket unit... + print("Updating SELinux policy can take a long time... please be patient.") + write_files.add("/tmp/tt-ssh-vsock-selinux.sh") + runcmd.append("/tmp/tt-ssh-vsock-selinux.sh") + + write_files.update({ + "/etc/systemd/system/tt-sd_notify.service", + "/etc/systemd/system/tt-sshd-vsock.socket", + "/etc/systemd/system/tt-sshd-vsock@.service", + }) + runcmd.extend([ + r"systemctl daemon-reload", + r"systemctl enable --now tt-sshd-vsock.socket", + r"systemctl enable tt-sd_notify.service", + ]) + + # do this one last: it notifies that we're done + runcmd.append("vsock-fling 2 1111 X_SYSTEMD_UNIT_ACTIVE=multi-user.target") return { "users": [ @@ -44,15 +121,16 @@ def cloud_init_config(image: str) -> JsonObject: { "name": "root", "lock_passwd": False, - "hashed_passwd": vm_pass, + "plain_text_passwd": "foobar", "groups": "users,wheel", "ssh_authorized_keys": ssh_authorized_keys, }, { "name": "admin", "lock_passwd": False, - "hashed_passwd": vm_pass, + "plain_text_passwd": "foobar", "gecos": "Administrator", + "primary_group": "admin", "groups": "users,wheel", "ssh_authorized_keys": ssh_authorized_keys, }, @@ -63,6 +141,7 @@ def cloud_init_config(image: str) -> JsonObject: "rsa_public": host_key_pub, }, "bootcmd": [("sh", "-c", cmd) for cmd in bootcmd], + "write_files": [extra_file(file) for file in write_files], "runcmd": [("sh", "-c", cmd) for cmd in runcmd], "cloud_final_modules": [ "scripts-per-once", @@ -86,6 +165,13 @@ class ImageCreate: self.no_build = no_build self.sit = sit + if image.endswith('boot'): + self.suffix = 'iso' + elif image.endswith('payload'): + self.suffix = 'tar.gz' + else: + self.suffix = 'qcow2' + def heading(self, text: str) -> None: self.ui.print(f"\n\033[1m## {text}\033[0m\n") @@ -95,35 +181,48 @@ class ImageCreate: await self.ui.run(bootstrap_script, tmpfile) async def setup(self, tmpfile: Path) -> None: + scripts = Path(SCRIPTS_DIR) + setup_script = scripts / f"{self.image}.setup" + if not setup_script.exists(): + self.heading("No additional setup required") + return + self.heading("Start guest") cockpit_test_identity = Path(MACHINE_DIR) / "identity" cockpit_test_identity.chmod(0o600) + cockpit_test_identity_pub = Path(MACHINE_DIR) / "identity.pub" + # deploying candlepin needs oodles of memory memory = "3072M" if self.image in ["services"] else "2048M" + cloud_init_user_data = cloud_init_config(self.image) + with testthing.IpcDirectory() as ipc: async with testthing.VirtualMachine( - boot="mbr", - cloud_init_user_data=cloud_init_config(self.image), + # attach_console=True, + cloud_init_user_data=cloud_init_user_data, + boot="mbr" if self.image in ['alpine'] else "efi", + credentials={ + # "ssh.authorized_keys.root": cockpit_test_identity_pub.read_text().strip(), + "passwd.plaintext-password.root": "foobar", + }, identity=(cockpit_test_identity, None), image=tmpfile, ipc=ipc, + journal=print_journal_line, memory=memory, networks=[testthing.Network("user")], snapshot=False, sit=self.sit, - target="los-gehts", - timeout=300, + target="multi-user.target", + timeout=120, ui=self.ui, ) as vm: - scripts = Path(SCRIPTS_DIR) - await vm.wait_boot() - self.heading("Run setup script on guest") await vm.scp("-r", scripts / "lib", "vm:/var/lib/testvm") - await vm.scp("-r", scripts / f"{self.image}.setup", "vm:/var/tmp/SETUP") + await vm.scp("-r", setup_script, "vm:/var/tmp/SETUP") await vm.execute( "/var/tmp/SETUP", self.image, @@ -146,39 +245,50 @@ class ImageCreate: data_dir.mkdir(exist_ok=True, mode=0o750, parents=True) with tempfile.NamedTemporaryFile( - dir=data_dir, prefix=f"{self.image}", suffix=".sparse.qcow2", delete_on_close=False + dir=data_dir, + prefix=f"{self.image}-", + suffix=".tmp.{self.suffix}", + delete_on_close=False, ) as tmp: # Copy image via convert, to make it sparse again - sparse = Path(tmp.name) - await self.ui.run( - ("qemu-img", "convert"), - ("-f", "qcow2", tmpfile), - ("-O", "qcow2", "-c", sparse), - ) + tmp_data_file = Path(tmp.name) + + if self.suffix == 'qcow': + # re-sparseify + await self.ui.run( + ("qemu-img", "convert"), + ("-f", "qcow2", tmpfile), + ("-O", "qcow2", "-c", tmp_data_file), + ) + else: + # copy from the tmpdir to the datadir + await self.ui.run('cp', tmpfile, tmp_data_file) # Hash the image here - with sparse.open("rb") as fp: + with tmp_data_file.open("rb") as fp: sha = hashlib.file_digest(fp, "sha256").hexdigest() - name = f"{self.image}-{sha}.qcow2" + name = f"{self.image}-{sha}.{self.suffix}" # a simple rename always works because it's the same directory - data_file = data_dir / name - self.ui.print_verbose(f"~ mv \\\n {sparse} \\\n {name}\n") - sparse.rename(data_file) + final_data_file = data_dir / name + self.ui.print_verbose(f"~ mv \\\n {final_data_file} \\\n {name}\n") + tmp_data_file.rename(final_data_file) tmp.close() images_dir = Path(IMAGES_DIR) - self.ui.print_verbose(f"~ ln -sf \\\n {data_file} \\\n {images_dir}/\n") + self.ui.print_verbose(f"~ ln -sf \\\n {final_data_file} \\\n {images_dir}/\n") (images_dir / name).unlink(missing_ok=True) - (images_dir / name).symlink_to(data_file) + (images_dir / name).symlink_to(final_data_file) self.ui.print_verbose(f"~ ln -sf \\\n {name} \\\n {images_dir}/{self.image}\n") (images_dir / self.image).unlink(missing_ok=True) (images_dir / self.image).symlink_to(name) async def create(self) -> None: - with testthing.cli_helper(), tempfile.NamedTemporaryFile() as tmp: + with tempfile.NamedTemporaryFile( + dir="/var/tmp", prefix=f"cockpit-image-create-{self.image}-", suffix=f".tmp.{self.suffix}" + ) as tmp: tmpfile = Path(tmp.name) await self.bootstrap(tmpfile) await self.setup(tmpfile) @@ -193,7 +303,8 @@ def main() -> None: parser.add_argument("image", help="The image to create") args = parser.parse_args() - asyncio.run(ImageCreate(**vars(args)).create()) + with testthing.cli_helper(): + asyncio.run(ImageCreate(**vars(args)).create()) if __name__ == "__main__": diff --git a/images/scripts/arch.setup b/images/scripts/arch.setup index 8b8813f911..987477d8ce 100755 --- a/images/scripts/arch.setup +++ b/images/scripts/arch.setup @@ -184,11 +184,6 @@ pacman -R --noconfirm cloud-init # https://github.com/cockpit-project/bots/issues/3901#issuecomment-1260579703 systemctl disable systemd-time-wait-sync.service -# https://gitlab.archlinux.org/archlinux/packaging/packages/openssh/-/issues/16 -# https://github.com/openssh/openssh-portable/pull/388 -# https://github.com/openssh/openssh-portable/pull/593 -echo 100::55:4e4b:4e4f:574e UNKNOWN | tee -a /etc/hosts - # Reduce image size, clear package cache (/var/cache/pacman/pkg) rm -f /var/cache/pacman/pkg/* diff --git a/images/scripts/bootc.setup b/images/scripts/bootc.setup index a551d8b016..b5f50f5149 100755 --- a/images/scripts/bootc.setup +++ b/images/scripts/bootc.setup @@ -27,20 +27,6 @@ podman rmi localhost:5000/bootc:latest localhost/bootc:latest podman rm -f -t0 ostree-registry rm /var/cache/bootc.oci.tar -# Older RHEL releases need vsock support for test.thing -if [ "${IMAGE#rhel-9}" != "$IMAGE" ] || [ "${IMAGE#centos-9}" != "$IMAGE" ]; then - cp -va /var/lib/testvm/test.thing-workarounds/*.{socket,service} /etc/systemd/system - systemctl daemon-reload - systemctl enable tt-sshd-vsock.socket tt-sd_notify.service - checkmodule -M -m -o /tmp/sshd_vsock.mod /var/lib/testvm/test.thing-workarounds/sshd_vsock.te - semodule_package -m /tmp/sshd_vsock.mod -o /tmp/sshd_vsock.pp - semodule -i /tmp/sshd_vsock.pp -fi - -# https://github.com/openssh/openssh-portable/pull/388 -# https://github.com/openssh/openssh-portable/pull/593 -echo 100::55:4e4b:4e4f:574e UNKNOWN | tee -a /etc/hosts - # disable various maintenance tasks which interfere with tests and don't make sense for our tests systemctl disable bootc-fetch-apply-updates.timer fstrim.timer logrotate.timer raid-check.timer diff --git a/images/scripts/lib/test.thing-workarounds/sshd_vsock.te b/images/scripts/lib/test.thing-workarounds/sshd_vsock.te deleted file mode 100644 index d05cceab2b..0000000000 --- a/images/scripts/lib/test.thing-workarounds/sshd_vsock.te +++ /dev/null @@ -1,17 +0,0 @@ -# test.thing workaround for SELinux issues with vsock on older RHEL versions -# https://issues.redhat.com/browse/RHEL-113647 - -module sshd_vsock 1.0; - -require { - type init_t; - type sshd_t; - type sshd_net_t; - class vsock_socket { - create bind listen accept getattr read write getopt setopt ioctl name_bind - }; -} - -allow init_t sshd_t:vsock_socket { create bind listen accept getattr setopt name_bind }; -allow sshd_t self:vsock_socket { read write getattr getopt setopt ioctl accept }; -allow sshd_net_t sshd_t:vsock_socket { read write getattr }; diff --git a/images/scripts/lib/test.thing-workarounds/tt-sd_notify.service b/images/scripts/lib/test.thing-workarounds/tt-sd_notify.service deleted file mode 100644 index cc7a4d9c84..0000000000 --- a/images/scripts/lib/test.thing-workarounds/tt-sd_notify.service +++ /dev/null @@ -1,19 +0,0 @@ -# test.thing polyfill for systems that lack support for vmm.notify_socket -# -# This works also on systemd versions that don't know about credentials from -# SMBIOS OEM strings by using dmidecode directly. socat is also required. - -[Unit] -After=sockets.target -Wants=sockets.target - -[Service] -Type=oneshot -ExecStart=/bin/bash -ec '\ - if addr="$$(dmidecode -qt11 | grep -om1 "vsock-stream:2:[0-9]*")"; then \ - printf X_SYSTEMD_UNIT_ACTIVE=sockets.target | socat - "$${addr/stream/connect}"; \ - fi \ -' - -[Install] -WantedBy=sockets.target diff --git a/images/scripts/lib/test.thing-workarounds/tt-sshd-vsock.socket b/images/scripts/lib/test.thing-workarounds/tt-sshd-vsock.socket deleted file mode 100644 index 2923be62dd..0000000000 --- a/images/scripts/lib/test.thing-workarounds/tt-sshd-vsock.socket +++ /dev/null @@ -1,10 +0,0 @@ -# test.thing polyfill for systems that lack support for sshd-vsock.socket -# -# If you use this, you also need to use tt-sshd-vsock@.service - -[Socket] -ListenStream=vsock::22 -Accept=yes - -[Install] -WantedBy=sockets.target diff --git a/images/scripts/lib/test.thing-workarounds/tt-sshd-vsock@.service b/images/scripts/lib/test.thing-workarounds/tt-sshd-vsock@.service deleted file mode 100644 index ba1a80c883..0000000000 --- a/images/scripts/lib/test.thing-workarounds/tt-sshd-vsock@.service +++ /dev/null @@ -1,13 +0,0 @@ -# test.thing polyfill for systems that lack support for sshd-vsock.socket -# -# If you use this, you also need to use tt-sshd-vsock.socket - -[Unit] -Wants=sshd-keygen.target -After=sshd-keygen.target - -[Service] -EnvironmentFile=-/etc/sysconfig/sshd -ExecStart=-/usr/sbin/sshd -i $OPTIONS -o "AuthorizedKeysFile ${CREDENTIALS_DIRECTORY}/ssh.ephemeral-authorized_keys-all .ssh/authorized_keys" -StandardInput=socket -LoadCredential=ssh.ephemeral-authorized_keys-all diff --git a/images/scripts/opensuse-tumbleweed.setup b/images/scripts/opensuse-tumbleweed.setup index 077f31fd56..0b75841cfd 100755 --- a/images/scripts/opensuse-tumbleweed.setup +++ b/images/scripts/opensuse-tumbleweed.setup @@ -171,10 +171,6 @@ echo 'GRUB_CMDLINE_LINUX_DEFAULT=""' >> /etc/default/grub echo 'GRUB_CMDLINE_LINUX="no_timer_check net.ifnames=0 console=tty1 console=ttyS0,115200n8"' >> /etc/default/grub grub2-mkconfig -o /boot/grub2/grub.cfg -# https://github.com/openssh/openssh-portable/pull/388 -# https://github.com/openssh/openssh-portable/pull/593 -echo 100::55:4e4b:4e4f:574e UNKNOWN | tee -a /etc/hosts - # reduce image size zypper clean diff --git a/images/scripts/rhel.setup b/images/scripts/rhel.setup index d4430d0361..8b34855343 100755 --- a/images/scripts/rhel.setup +++ b/images/scripts/rhel.setup @@ -291,16 +291,6 @@ if [ "${IMAGE#rhel-[89]}" != "$IMAGE" ] || [ "${IMAGE#centos-[89]}" != "$IMAGE" TEST_PACKAGES="$TEST_PACKAGES python3-tracer" fi -# Older RHEL releases need vsock support for test.thing -if [ "${IMAGE#rhel-[89]}" != "$IMAGE" ] || [ "${IMAGE#centos-9}" != "$IMAGE" ]; then - cp -va /var/lib/testvm/test.thing-workarounds/*.{socket,service} /etc/systemd/system - systemctl daemon-reload - systemctl enable tt-sshd-vsock.socket tt-sd_notify.service - checkmodule -M -m -o /tmp/sshd_vsock.mod /var/lib/testvm/test.thing-workarounds/sshd_vsock.te - semodule_package -m /tmp/sshd_vsock.mod -o /tmp/sshd_vsock.pp - semodule -i /tmp/sshd_vsock.pp -fi - # These packages are downloaded to the image so that the tests can # install them on-demand. diff --git a/lib/testthing.py b/lib/testthing.py index eb7808ffca..02a2ba6812 100755 --- a/lib/testthing.py +++ b/lib/testthing.py @@ -30,11 +30,12 @@ # When copying test.thing into your own project, try to use a tagged version. # If you need to use a version between tags or have made your own # modifications, please make note if it by modifying the version number. -__version__ = "0.3.4+hacks!" +__version__ = "0.4.0" import argparse import asyncio import contextlib +import contextvars import ctypes import dataclasses import functools @@ -51,10 +52,19 @@ import tempfile import traceback import weakref -from collections.abc import AsyncGenerator, Callable, Iterable, Mapping, Sequence, Iterator +from collections.abc import ( + AsyncGenerator, + AsyncIterator, + Callable, + Coroutine, + Iterable, + Iterator, + Mapping, + Sequence, +) from pathlib import Path -from types import TracebackType from typing import Any, Literal, Never, Self +from types import TracebackType logger = logging.getLogger(__name__) @@ -91,10 +101,6 @@ VM images have the public half of this key inside of them, so it's useful for gaining access to those if they lack support for ephemeral ssh keys.""" -COCKPIT_TEST_IDENTITY_PUB = "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQDUOtNJdBEXyKxBB898rdT54ULjMGuO6v4jLXmRsdRhR5Id/lKNc9hsdioPWUePgYlqML2iSV72vKQoVhkyYkpcsjr3zvBny9+5xej3+TBLoEMAm2hmllKPmxYJDU8jQJ7wJuRrOVOnk0iSNF+FcY/yaQ0owSF02Nphx47j2KWc0IjGGlt4fl0fmHJuZBA2afN/4IYIIsEWZziDewVtaEjWV3InMRLllfdqGMllhFR+ed2hQz9PN2QcapmEvUR4UCy/mJXrke5htyFyHi8ECfyMMyYeHwbWLFQIve4CWix9qtksvKjcetnxT+WWrutdr3c9cfIj/c0v/Zg/c4zETxtp cockpit-test" # noqa: E501 -"""A copy of `bots/machine/identity.pub` from the cockpit project. The -corresponding public key for COCKPIT_TEST_IDENTITY.""" - # This is basically tempfile.TemporaryDirectory but sequentially-allocated. # We do that so we can easily interact with the VMs from outside (with ssh). @@ -240,12 +246,14 @@ async def execute(command: str) -> object: return response -def _ssh_direct_args(identities: Sequence[Path], port: Path) -> tuple[tuple[str, str], ...]: +def _ssh_direct_args( + identities: Sequence[Path], vsock: Path +) -> tuple[tuple[str, str], ...]: options = { # Fake that we know the host key "KnownHostsCommand": "/bin/echo %H %t %K", # Use systemd-ssh-proxy to connect via vsock - "ProxyCommand": f"/usr/lib/systemd/systemd-ssh-proxy vsock-mux/{port} 22", + "ProxyCommand": f"/usr/lib/systemd/systemd-ssh-proxy vsock-mux/{vsock} 22", "ProxyUseFdpass": "yes", # Try to prevent interactive prompting and/or updating known_hosts # files or otherwise interacting with the environment @@ -260,8 +268,8 @@ def _ssh_direct_args(identities: Sequence[Path], port: Path) -> tuple[tuple[str, return ( ("-F", "none"), # don't use the user's config - *(("-o", f"{k} {v}") for k, v in options.items()), - *(("-o", f"IdentityFile {path}") for path in identities), + *(("-o", f"{k}={v}") for k, v in options.items()), + *(("-i", f"{path}") for path in identities), ) @@ -271,15 +279,35 @@ def _stderr_is_tty() -> bool: class UI: + """A helper for printing messages and launching subprocesses.""" + def __init__(self, *, status_messages: bool, verbose: bool) -> None: + """Create a UI helper. + + This controls the stderr output of test.thing. The test.thing library + never writes to stdout. + + - status_messages: if intermediary messages about the state of the + machine should be printed or not + - verbose: extra output is printed (like all executed commands) + + If both are false then nothing will be printed. + """ self._status_messages = status_messages self._verbose = verbose def clear_status_message(self) -> None: + """Clear any displayed status message.""" if _stderr_is_tty(): sys.stderr.write("\r\033[2K") def print_status(self, line: str) -> None: + """Print a status line message. + + This is only printed if status_messages=True. A status message is a + transient message that will be erased at the next output (unless stderr + is not a TTY). + """ if self._status_messages: if os.isatty(2): sys.stderr.write("\r\033[2K " + line + "\r") @@ -287,11 +315,13 @@ def print_status(self, line: str) -> None: sys.stderr.write(line + "\n") def print_verbose(self, line: str) -> None: + """Print a verbose message, if verbose=True.""" if self._verbose: self.clear_status_message() sys.stderr.write(line + "\n") def print(self, line: str) -> None: + """Print a message, unconditionally.""" self.clear_status_message() sys.stderr.write(line + "\n") @@ -320,7 +350,12 @@ def stdin_ready() -> None: finally: loop.remove_reader(0) - async def sit(self, msg: str | None = None, exc: BaseException | None = None) -> None: + async def sit( + self, + msg: str | None = None, + vm_id: str | None = None, + exc: BaseException | None = None, + ) -> None: """Wait for the user to press Enter.""" # lis: the ages old design question: does the button show # the current state or the future one when you press it :) @@ -335,9 +370,13 @@ async def sit(self, msg: str | None = None, exc: BaseException | None = None) -> self.clear_status_message() if exc is not None: - sys.stderr.write(f"\n🤦 {''.join(traceback.format_exception(exc))}\n") + self.print(f"\n🤦 {''.join(traceback.format_exception(exc))}") if msg is not None: - sys.stderr.write(f"\n{msg}\n") + self.print(f"\n{msg}") + if vm_id is not None: + self.print( + f"Guest is still running. Connect with: \033[1mssh {vm_id}\033[0m" + ) await self._wait_stdin("\nEnter or EOF to exit ⏸️ ") @@ -385,22 +424,21 @@ def pr_set_pdeathsig() -> None: if prctl(PR_SET_PDEATHSIG, signal.SIGTERM): os._exit(1) # should never happen - # If we are expecting output, clear the status message - if stdout is None or stderr is None: - self.clear_status_message() - self.print_verbose(f"+ {_pretty_print_args(*args)}\n") return await asyncio.subprocess.create_subprocess_exec( *itertools.chain(*_normalize_args(*args)), stdin=stdin, stdout=stdout, + stderr=stderr, preexec_fn=pr_set_pdeathsig, + start_new_session=True, ) async def run( self, *args: str | Path | tuple[str | Path, ...], + stdin: int | None = asyncio.subprocess.DEVNULL, check: bool = True, ) -> int: """Run a process, waiting for it to exit. @@ -408,7 +446,7 @@ async def run( This takes the same arguments as spawn, plus a "check" argument (True by default) which works in the usual way. """ - process = await self.spawn(*args, stdin=asyncio.subprocess.DEVNULL) + process = await self.spawn(*args, stdin=stdin) returncode = await process.wait() if check and returncode != 0: raise SubprocessError(args, returncode=returncode) @@ -467,7 +505,9 @@ async def chown( user, group = owner owner = f"{user or ''}:{group or ''}" - await self._vm.execute("chown", "-h" if not follow_symlinks else (), owner, self) + await self._vm.execute( + "chown", "-h" if not follow_symlinks else (), owner, self + ) async def write_bytes(self, data: bytes, *, append: bool = False) -> None: """Write or append to to a binary file.""" @@ -487,9 +527,13 @@ async def write_text(self, data: str, *, append: bool = False) -> None: """Write or append to a text file.""" await self.write_bytes(data.encode(), append=append) - async def unlink(self, *, missing_ok: bool = False, recursive: bool = False) -> None: + async def unlink( + self, *, missing_ok: bool = False, recursive: bool = False + ) -> None: """Unlink the given file.""" - await self._vm.execute("rm", "-f" if missing_ok else (), "-r" if recursive else (), self) + await self._vm.execute( + "rm", "-f" if missing_ok else (), "-r" if recursive else (), self + ) async def rmdir(self) -> None: """Remove a directory.""" @@ -524,7 +568,54 @@ def to_qemu(self) -> str: return f"socket,mcast=230.0.0.1:{self.id},localaddr=127.0.0.1" -class VirtualMachine: +class _ServiceGroup(asyncio.TaskGroup): + """A special kind of TaskGroup to support 'background services'. + + Tasks normally run 'forever', and end only via cancellation. There's also a + "ready" notification mechanism via contextvars, which allows combining + "setup" and "running" phases into a single task, allowing parallel startup. + """ + + ready_var = contextvars.ContextVar[asyncio.Event]("Service task ready event") + + def __init__(self) -> None: + super().__init__() + self.__ready: list[asyncio.Event] = [] + self.__tasks: list[asyncio.Task[None]] = [] + + def add_service(self, coro: Coroutine[None, None, None]) -> None: + event = asyncio.Event() + context = contextvars.copy_context() + context.run(self.ready_var.set, event) + self.__tasks.append(self.create_task(coro, context=context)) + self.__ready.append(event) + + @classmethod + def notify_ready(cls) -> None: + cls.ready_var.get().set() + + async def wait_all_ready(self) -> None: + for event in self.__ready: + await event.wait() + + async def cancel_all(self) -> None: + for task in self.__tasks: + task.cancel() # already checks task.done() + + async def __aexit__( + self, + et: type[BaseException] | None, + exc: BaseException | None, + tb: TracebackType | None, + ) -> None: + try: + return await super().__aexit__(et, exc, tb) + except BaseExceptionGroup as eg: + if len(eg.exceptions) == 1: + raise eg.exceptions[0] + raise + +class VirtualMachine(contextlib.AsyncExitStack): """A handle to a running virtual machine. This is meant to be used as an async context manager like so: @@ -544,16 +635,24 @@ class VirtualMachine: See the documentation for the module. """ - """ssh command-line arguments for executing commands""" ssh_args: tuple[str | Path | tuple[str | Path, ...], ...] + """ssh command-line arguments for executing commands""" - """ssh command-line arguments to be used to connect directly to the vsock""" ssh_direct_args: tuple[str | Path | tuple[str | Path, ...], ...] | None + """ssh command-line arguments to be used to connect directly to the vsock""" + + journal: list[dict[str, str]] + """A list of journal entries from the guest. + + Each entry is a dictionary. Multiple values per key are not supported, but + binary data is: the strings are encoded with errors='surrogateescape' so + it's possible to get the original binary back if that's what you're + expecting. + + See the journal= kwarg to VirtualMachine(). + """ - # our three background tasks - _notify_server_task: asyncio.Task[None] | None = None _ssh_control_task: asyncio.Task[None] | None = None - _qemu_task: asyncio.Task[None] | None = None def __init__( self, @@ -567,12 +666,14 @@ def __init__( cpus: int = 4, identity: tuple[Path, str | None] | None = None, identities: Sequence[str | Path] = (), + journal: bool | Callable[[dict[str, str]], bool | None] = False, memory: int | str = "4G", networks: Sequence[Network] = (), + provision_ssh_key: bool = False, sit: bool = False, snapshot: bool = True, status_messages: bool = False, - target: str = 'sockets.target', + target: str = "sockets.target", timeout: float = 30.0, ui: UI | None = None, verbose: bool = False, @@ -591,8 +692,12 @@ def __init__( - identities: extra private keys (either as paths on the disk or directly as strings) to pass to ssh. Useful if you're not sure which key the image will accept and want to try multiple. + - journal: False (default) to disable journal handling, True to + record all entries, and a callable to decide on a per-entry basis. - memory: how much memory the guest gets in MiB, or a string like "4G" - networks: a list of Network objects (or empty to disable networking) + - provision_ssh_key: if we should attempt to install the public side + of the ssh key as ~root/.ssh/authorized_keys in the guest - sit: if we should "sit" when an exception occurs: print the exception and wait for input (to allow inspecting the running VM) - snapshot: if the 'snapshot' option is used on the disk (changes are @@ -600,9 +705,12 @@ def __init__( - status_messages: if we should do output of status messages (stderr) - target: the name of the systemd target to wait for - timeout: how long to wait for the VM to start, or 'inf' - - ui: a custom instance of UI (otherwise a new one is constructed per status_messages= and verbose=) + - ui: a custom instance of UI (otherwise a new one is constructed per + status_messages= and verbose=) - verbose: if we should do output of verbose messages (stderr) """ + super().__init__() + self.image = image self._ipc = ipc self._attach_console = attach_console @@ -612,32 +720,33 @@ def __init__( self._credentials = credentials self._identity = identity self._identities = identities + self._journal = journal self._memory = memory self._networks = networks + self._provision_ssh_key = provision_ssh_key self._sit = sit self._snapshot = snapshot self._target = target self._timeout = timeout self._ui = ui or UI(status_messages=status_messages, verbose=verbose) - self._tasks = asyncio.TaskGroup() + self._tasks = _ServiceGroup() self._ssh_control_ready = asyncio.Event() self._qemu_exited = asyncio.Event() self._shutdown_ok = False self.root = GuestPath("/", vm=self) self.home = GuestPath(".", vm=self) + self.journal = [] async def _run_helper( - self, - *args: str | Path | tuple[str | Path, ...], - ready_when: Path, - ) -> asyncio.Task[None]: + self, *args: str | Path | tuple[str | Path, ...], ready_when: Path + ) -> None: """Run a helper process in the background. - The function returns when the given named file (usually a socket) - appears. The process is awaited from our task group. It should never - exit. On cancellation, the process is terminated. + This is designed to be used from _ServiceGroup and will notify when the + process is properly running. On cancellation, the process is + terminated. """ process = await self._ui.spawn(*args) @@ -647,21 +756,23 @@ async def _run_helper( returncode = await asyncio.wait_for(process.wait(), 0.01) raise SubprocessError(args, returncode=returncode) - async def wait_exit() -> None: - try: - returncode = await process.wait() # this should never return - raise SubprocessError(args, returncode=returncode) - except asyncio.CancelledError: - process.kill() - await process.wait() + _ServiceGroup.notify_ready() - return self._tasks.create_task(wait_exit()) + try: + returncode = await process.wait() # this should never return + raise SubprocessError(args, returncode=returncode) + except asyncio.CancelledError: + with contextlib.suppress(ProcessLookupError): + process.kill() + await process.wait() - async def _ssh_keygen(self) -> tuple[Path, str]: - """Create a ssh key in the given directory. + async def _ssh_keygen_service(self) -> None: + """Create the ephemeral ssh key. - Returns the path to the private key and the public key as a string. + This is designed to be used from _ServiceGroup and will notify when the + key has been generated and stored on self._identity. """ + assert self._identity is None private_key = self._ipc / "id" await self._ui.run( @@ -673,7 +784,8 @@ async def _ssh_keygen(self) -> tuple[Path, str]: ("-f", f"{private_key}"), ) - return private_key, (self._ipc / "id.pub").read_text().strip() + self._identity = private_key, (self._ipc / "id.pub").read_text().strip() + _ServiceGroup.notify_ready() def _sd_notify(self, line: str) -> None: logger.debug("sd_notify:%s", line) @@ -693,31 +805,73 @@ def _sd_notify(self, line: str) -> None: elif key == "X_SYSTEMD_SHUTDOWN": self._ui.print_status(f"Shutdown: {value}") - async def _sd_notify_connection(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: - logger.debug("sd_notify connection") - try: - # Actually we should read until EOF but see - # https://github.com/rust-vmm/vhost-device/issues/874 - message = await reader.read(65536) - self._sd_notify(message.decode()) - finally: - writer.close() - await writer.wait_closed() - - async def _qemu( - self, - *, - port: int, - public: str | None, - ) -> None: - creds = { - **self._credentials, - "vmm.notify_socket": f"vsock-stream:2:{port}", - } + async def _sd_notify_service(self, path: Path) -> None: + async def connection( + reader: asyncio.StreamReader, writer: asyncio.StreamWriter + ) -> None: + logger.debug("sd_notify connection") + try: + # Actually we should read until EOF but see + # https://github.com/rust-vmm/vhost-device/issues/874 + message = await reader.read(65536) + self._sd_notify(message.decode()) + finally: + writer.close() + await writer.wait_closed() + + async with await asyncio.start_unix_server(connection, path) as srv: + _ServiceGroup.notify_ready() + await srv.serve_forever() + + async def _journal_service(self, path: Path) -> None: + intern_table: dict[str, str] = {} + + def intern(bval: bytes) -> str: + # The spec says that binary is "rare", so let's do strings, but use + # surrogateescape to leave the door open to having a way back. + sval = bval.decode(errors="surrogateescape") + return intern_table.setdefault(sval, sval) + + async def connection( + reader: asyncio.StreamReader, writer: asyncio.StreamWriter + ) -> None: + assert self._journal is not False - if public is not None: - creds["ssh.ephemeral-authorized_keys-all"] = public + print("JOURNAL CON") + print("JOURNAL CON") + print("JOURNAL CON") + # https://systemd.io/JOURNAL_EXPORT_FORMATS/#journal-export-format + try: + entry: dict[str, str] = {} + + while line := await reader.readline(): + if line := line.rstrip(b"\n"): + key, eq, value = line.partition(b"=") + + if not eq: # no equal sign? + # 64bit le size, followed by data, followed by newline + size = await reader.readexactly(8) + value = await reader.readexactly( + int.from_bytes(size, "little") + ) + _ = await reader.readexactly(1) + + # NB: no multiple-entries supported :( + entry[intern(key)] = intern(value) + + else: + if self._journal is True or self._journal(entry): + self.journal.append(entry) + entry = {} + finally: + writer.close() + + async with await asyncio.start_unix_server(connection, path) as srv: + _ServiceGroup.notify_ready() + await srv.serve_forever() + + async def _qemu(self, creds: Mapping[str, str]) -> None: snap = "on" if self._snapshot else "off" drives = [f"file={self.image},format=qcow2,discard=unmap,snapshot={snap}"] @@ -725,7 +879,9 @@ async def _qemu( cloud_init = self._ipc / "cloud-init" cloud_init.mkdir() (cloud_init / "meta-data").touch() - (cloud_init / "user-data").write_text("#cloud-config\n" + json.dumps(self._cloud_init_user_data) + "\n") + (cloud_init / "user-data").write_text( + "#cloud-config\n" + json.dumps(self._cloud_init_user_data) + "\n" + ) drives.append(f"driver=vvfat,dir={cloud_init},readonly=on,label=CIDATA") args = ( @@ -733,6 +889,7 @@ async def _qemu( "-nodefaults", ("-object", f"memory-backend-memfd,share=on,id=mem0,size={self._memory}"), ("-bios", _find_ovmf()) if self._boot == "efi" else (), + ("-boot", "menu=on"), ("-machine", "q35,accel=kvm,memory-backend=mem0"), ("-cpu", "host"), ("-smp", f"{self._cpus}"), @@ -743,7 +900,10 @@ async def _qemu( ("-device", "vhost-user-vsock-pci,chardev=vsock"), # Console stuff... ("-device", "virtio-serial-pci"), - ("-chardev", f"socket,path={self._ipc}/vsock_1111,id=tt-notify,reconnect-ms=1"), + ( + "-chardev", + f"socket,path={self._ipc}/vsock_1111,id=tt-notify,reconnect-ms=1", + ), ("-device", "virtserialport,chardev=tt-notify,name=tt-notify"), ("-serial", "chardev:console"), *( @@ -761,15 +921,21 @@ async def _qemu( ) ), *(("-drive", f"{drive},if=virtio,media=disk") for drive in drives), - *(("-nic", net.to_qemu() + ",model=virtio-net-pci") for net in self._networks), + *( + ("-nic", net.to_qemu() + ",model=virtio-net-pci") + for net in self._networks + ), # Credentials - *(("-smbios", f"type=11,value=io.systemd.credential:{k}={v}") for k, v in creds.items()), + *( + ("-smbios", f"type=11,value=io.systemd.credential:{k}={v}") + for k, v in creds.items() + ), ) qemu = None try: self._ui.print_status("Waiting for guest") - qemu = await self._ui.spawn(*args) + qemu = await self._ui.spawn(*args, stdin=None) returncode = await qemu.wait() if not self._shutdown_ok: raise SubprocessError(args, returncode) @@ -794,7 +960,7 @@ async def _ssh_control(self) -> None: try: assert self.ssh_direct_args is not None - self._ui.print_status("ssh control socket: Connecting") + self._ui.print_status("ssh control socket: connecting via vsock") control_socket = self._ipc / "ssh" @@ -822,8 +988,7 @@ async def _ssh_control(self) -> None: # we're online! self._ssh_control_ready.set() - self._ui.print_status("ssh control socket: Connected") - self._ui.clear_status_message() + self._ui.print_status("ssh control socket: connected.") returncode = await ssh.wait() if not self._shutdown_ok: @@ -839,135 +1004,143 @@ async def _ssh_control(self) -> None: self._ssh_control_ready.clear() self._ssh_control_task = None - async def __aenter__(self) -> Self: - """Start the virtual machine.""" - await self._tasks.__aenter__() - - self.ssh_args = ( - ("-F", "none"), # don't use the user's config - ("-o", f"ControlPath={self._ipc}/ssh"), # connect via the control socket - ) - + def _get_console_log(self) -> Sequence[str]: try: - if self._identity is None: - self._identity = await self._ssh_keygen() - - private, public = self._identity - identities = [private] - - for nr, identity in enumerate(self._identities): - if isinstance(identity, str): - path = self._ipc / f"id.{nr}" - with path.open("x") as extra: - extra.write(identity) - path.chmod(0o400) - identities.append(path) - else: - identities.append(identity) + log = (self._ipc / "console").read_text(errors="replace") + except FileNotFoundError: + log = "" + + # Remove ANSI escapes, control characters, extra newlines + log_lines = re.sub( + r"\x1b\[[ -?]*[@-~]|" # CSI: ESC [ + params/interms + final + r"\x1b\][^\a\x1b]*|" # OSC: ESC ] + everything to \a or ESC + r"\x1b[ ()O].|\x1b.|" # two- and one-character escapes + r"[\x00-\b\v-\x1f\x7f]|" # all control chars but [\t\n] + r"(<=\n)\n", # extra newlines + "", + log, + ).splitlines() + + return ["Console log:" if log_lines else "Console log unavailable.", *log_lines] + + async def _start_services( + self, services: _ServiceGroup, creds: dict[str, str] + ) -> None: + if self._identity is None: + services.add_service(self._ssh_keygen_service()) - self.ssh_direct_args = _ssh_direct_args(identities, self._ipc / "vsock") + services.add_service(self._sd_notify_service(self._ipc / "vsock_1111")) + creds["vmm.notify_socket"] = "vsock-stream:2:1111" - # It goes like this: - # - we start listening on the sd-notify socket - # - we start qemu - # - at some point the guest will notify that ssh is ready - # - this causes us to spawn the ssh control task - # - once connected, _ssh_control_ready gets set - # - we wait for that, so when it's done, we're done + if self._journal: + services.add_service(self._journal_service(self._ipc / "vsock_1112")) + creds["journal.forward_to_socket"] = "vsock-stream:2:1112" - self._vsock_helper = await self._run_helper( + services.add_service( + self._run_helper( "vhost-device-vsock", ("--socket", self._ipc / "vsock-device"), ("--uds-path", self._ipc / "vsock"), ready_when=self._ipc / "vsock-device", ) + ) - server = await asyncio.start_unix_server(self._sd_notify_connection, self._ipc / "vsock_1111") - self._notify_server_task = self._tasks.create_task(server.serve_forever()) + def _setup_identities(self, creds: dict[str, str]) -> Iterable[Path]: + assert self._identity is not None + private, public = self._identity + yield private - self._qemu_task = self._tasks.create_task(self._qemu(port=1111, public=public)) + if public is not None: + creds["ssh.ephemeral-authorized_keys-all"] = public + if self._provision_ssh_key: + creds["ssh.authorized_keys.root"] = public + + for nr, identity in enumerate(self._identities): + if isinstance(identity, str): + path = self._ipc / f"id.{nr}" + with path.open("x") as extra: + extra.write(identity) + path.chmod(0o400) + yield path + else: + yield identity - # the notify socket server will create the ssh control socket task - # which, in turn, sets this ready once it's online. - try: - await asyncio.wait_for(self._ssh_control_ready.wait(), self._timeout) - except TimeoutError as exc: - self._ui.clear_status_message() + @contextlib.asynccontextmanager + async def _run(self) -> AsyncIterator[Self]: + # It goes like this: + # - we start listening on the sd-notify socket + # - we start qemu + # - at some point the guest will notify that ssh is ready + # - this causes us to spawn the ssh control task + # - once connected, _ssh_control_ready gets set + # - we wait for that, so when it's done, we're done + creds = {**self._credentials} - try: - log = (self._ipc / "console").read_text(errors="replace") - except FileNotFoundError: - log = "" - - # Remove ANSI escapes and weird/excessive line endings - log = re.sub(r"\033\[[0-9;]*[A-Za-z]", "", log) - log_lines = re.split(r"[\r\n]+", log) - - lines = [ - "Timed out waiting for the VM to start (after {self._timeout}s).", - "Console log:" if log_lines else "Console log unavailable.", - *log_lines, - ] - - raise TimeoutError("\n".join(lines) + "\n") from exc - - except: - exc_info = sys.exc_info() - logger.debug("Startup failed: %r", exc_info) - await self._tasks.__aexit__(*exc_info) - raise - else: - logger.debug("Startup was successful") - return self - finally: - logger.debug("Startup is done") - self._ui.clear_status_message() + self.ssh_args = ( + ("-F", "none"), # don't use the user's config + ("-o", f"ControlPath={self._ipc}/ssh"), # connect via the control socket + ) - async def __aexit__( - self, - et: type[BaseException] | None, - exc: BaseException | None, - tb: TracebackType | None, - ) -> None: - """Stop the virtual machine.""" - self._shutdown_ok = True + async with _ServiceGroup() as services: + # Start all of the background services in parallel + await self._start_services(services, creds) - try: - if self._sit and exc is not None: - current = asyncio.current_task() - cancelled = current and current.cancelled() - - # This will report exceptions thrown from inside of the context managed - # block but won't work properly in case we were cancelled due to an - # error elsewhere (ie: qemu unexpectedly exiting) because at that point - # the main task will have been cancelled and we won't be able to block - # while waiting to read from stdin. That's ok: the goal if this is to - # catch test failures, not infra failures. - if not cancelled: - await self._ui.sit(exc=exc) - - if self._ssh_control_task is not None: - task = self._ssh_control_task - task.cancel() - await task # Cancellation is async, so wait - - # If we're in snapshot mode then we don't have to do a clean shutdown - with contextlib.suppress(FileNotFoundError): - await self.qmp("quit" if self._snapshot else "system_powerdown") - await self._qemu_exited.wait() - - assert self._vsock_helper is not None - self._vsock_helper.cancel() - - assert self._notify_server_task is not None - self._notify_server_task.cancel() + # ...and wait for them to be ready + await services.wait_all_ready() - finally: - # If the TaskGroup fails (possibly by way of 'exc') and shutdown - # also failed then the error from the shutdown will be suppressed. - await self._tasks.__aexit__(et, exc, tb) - if exc is not None: - raise exc + # we should have our ssh key by now, so deal with that + identities = tuple(self._setup_identities(creds)) + self.ssh_direct_args = _ssh_direct_args(identities, self._ipc / "vsock") + + async with self._tasks: + # start QEMU + self._tasks.create_task(self._qemu(creds)) + + # the notify socket server will create the ssh control socket task + # which, in turn, sets this ready once it's online. + try: + await asyncio.wait_for( + self._ssh_control_ready.wait(), self._timeout + ) + except TimeoutError as exc: + lines = [ + "Timed out waiting for the VM to start" + f" (after {self._timeout}s).", + *self._get_console_log(), + ] + raise TimeoutError("\n".join(lines) + "\n") from exc + + # we're online + try: + yield self + except asyncio.CancelledError: + raise + except Exception as exc: + if self._sit: + await self._ui.sit(vm_id=self.get_id(), exc=exc) + raise + + # time to shutdown. + self._shutdown_ok = True + + # start with the control channel. + if self._ssh_control_task is not None: + task = self._ssh_control_task + task.cancel() + await task # Cancellation is async, so wait + + # next, qemu: if we're in snapshot mode then we don't have to do a + # clean shutdown + with contextlib.suppress(FileNotFoundError): + await self.qmp("quit" if self._snapshot else "system_powerdown") + + # finally, all background services + await services.cancel_all() + + async def __aenter__(self) -> Self: + """Start the virtual machine.""" + await super().__aenter__() + return await self.enter_async_context(self._run()) def get_id(self) -> str: """Get the machine identifier like `tt.0`, `tt.1`, etc.""" @@ -993,19 +1166,6 @@ async def cancel_port(self, *args: tuple[str, ...]) -> None: """Cancel a previous forward.""" return await self._ssh_cmd(("-O", "cancel"), *args) - async def scp(self, *args: str | pathlib.Path, direct: bool = False) -> None: - """Do a file transfer with scp. - - The hostname is ignored, so for paths on the guest use something like - `vm:/tmp/path`. - """ - assert self.ssh_direct_args is not None - await self._ui.run( - "scp", - *(self.ssh_direct_args if direct else self.ssh_args), - args, - ) - async def wait_boot(self) -> None: """Wait for the machine to be fully-booted.""" await self.execute("systemctl", "is-system-running", "--wait") @@ -1018,6 +1178,7 @@ async def execute( direct: bool = False, input: bytes | str | None = b"", # noqa:A002 # shadows `input()` but so does subprocess module environment: Mapping[str, str] = {}, + stdin: int | None = asyncio.subprocess.PIPE, stdout: int | None = asyncio.subprocess.PIPE, ) -> str: """Execute a command on the guest. @@ -1039,11 +1200,7 @@ async def execute( cmd, ) - ssh = await self._ui.spawn( - *full_command, - stdin=asyncio.subprocess.PIPE, - stdout=stdout, - ) + ssh = await self._ui.spawn(*full_command, stdin=stdin, stdout=stdout) input_bytes = input.encode() if isinstance(input, str) else input output, _ = await ssh.communicate(input_bytes) returncode = await ssh.wait() @@ -1087,6 +1244,43 @@ async def write( if perm: await dest.chmod(perm) + async def scp(self, *args: str | pathlib.Path, direct: bool = False) -> None: + """Do a file transfer with scp. + + The hostname is ignored, so for paths on the guest use something like + `vm:/tmp/path`. + + All arguments that are given in the form of `pathlib.Path` are passed to + `scp` in absolute form, avoiding worries about `:` characters, but also + making it impossible to use certain scp features (such as the special + treatment of `.` — which pathlib collapses anyway). Use string form if + you need this, and worry about the escaping yourself. + """ + assert self.ssh_direct_args is not None + await self._ui.run( + "scp", + *(self.ssh_direct_args if direct else self.ssh_args), + tuple(p.absolute() if isinstance(p, pathlib.Path) else p for p in args), + ) + + async def upload( + self, *args: str | pathlib.Path, target_directory: str | GuestPath | None = None + ) -> None: + """Upload files to the guest. + + This works similarly to `cp --target-directory` (`-t`). + + All arguments are interpreted as local paths. The target directory is + a directory on the remote system (defaulting to root's home directory) + which will be created (`mkdir -p`) if it doesn't exist. Each argument + is recursively copied into that directory by its basename. + """ + if target_directory is not None: + await self.execute("mkdir", "-p", target_directory) + await self.scp( + "-r", *map(pathlib.Path, args), f"{self.get_id()}:{target_directory or ''}" + ) + @contextlib.asynccontextmanager async def disconnected(self) -> AsyncGenerator[None]: """Temporarily disconnect the control socket. @@ -1145,11 +1339,16 @@ def __init__( self.output = output if returncode < 0: - msg = f"Subprocess terminated by {signal.Signals(-returncode).name}" + msg = f"Subprocess terminated by {signal.Signals(-returncode).name}\n" else: msg = f"Subprocess exited unexpectedly with return code {returncode}:\n" - super().__init__(f"{msg}\n{_pretty_print_args(*args)}\n\n") + if self.output: + out = "\n🗯️ Output:\n\n" + self.output.decode(errors="replace") + else: + out = "" + + super().__init__(f"{msg}\n{_pretty_print_args(*args)}\n{out}\n") def cleanup_on_signal() -> None: @@ -1169,23 +1368,60 @@ def _term(*args: object) -> Never: signal.signal(signal.SIGTERM, _term) +async def _ssh_properly_configured() -> bool: + proc = None + try: + proc = await asyncio.subprocess.create_subprocess_exec( + *("ssh", "-G", "tt.n"), + stdout=asyncio.subprocess.PIPE, + ) + stdout, _stderr = await proc.communicate() + except OSError: + return False + else: + return b"/test.thing/tt.n/ssh\n" in stdout + finally: + if proc is not None: + await proc.wait() + + +async def _show_ssh_hints(ui: UI, vm_id: str) -> None: + if not _stderr_is_tty(): + return + + if await _ssh_properly_configured(): + ui.print(f"\n🍓 VM running. Connect with: \033[1mssh {vm_id}\033[0m\n") + else: + ui.print(f""" +Please consider adding this stanza to your SSH config: + +Host tt.* + ControlPath ${{XDG_RUNTIME_DIR}}/test.thing/%h/ssh + +At which point you can connect to the VM using \033[1mssh {vm_id}\033[0m\n""") + + @contextlib.contextmanager def cli_helper() -> Iterator[None]: - """A context manager for using test.thing from CLI tools. + """Help use test.thing from CLI tools. This installs a signal handler for clean exit on SIGHUP and SIGTERM and - catches SubprocessError and TimeoutError, printing a message to stderr and - calling sys.exit().""" + catches SubprocessError, TimeoutError, and KeyboardInterrupt, printing a + message to stderr and calling sys.exit(). + + Because of how cancellation is used internally to handle KeyboardInterrupt, + you should use this *outside* of asyncio.run(). + """ cleanup_on_signal() try: yield - except* (SubprocessError, TimeoutError) as eg: + except* (SubprocessError, TimeoutError, KeyboardInterrupt) as eg: for exc in eg.exceptions: - sys.stderr.write(f"🤦 {exc}\n") + sys.stderr.write(f"\n🤦 [{exc.__class__.__name__}] {exc}\n\n") sys.exit("I'm sorry it didn't work out.") -def _main() -> None: # noqa: C901 +def _main() -> None: class AppendTuple(argparse.Action): def __call__( self, @@ -1199,19 +1435,31 @@ def __call__( fwds = (*fwds, (option_string, values)) setattr(namespace, self.dest, fwds) - parser = argparse.ArgumentParser(description="test.thing - a simple modern VM runner") - parser.add_argument("--maintain", "-m", action="store_true", help="Changes are permanent") - parser.add_argument("--attach", "-a", action="store_true", help="Attach to the VM console") - parser.add_argument("--sit", action="store_true", help="Wait for enter key on exceptions") - parser.add_argument("--debug", "-d", action="store_true", help="Enable debug output") - parser.add_argument("--cloud-init", action="store_true", help="Add Cockpit ssh key via cloud-init") + parser = argparse.ArgumentParser( + description="test.thing - a simple modern VM runner" + ) + parser.add_argument( + "--maintain", "-m", action="store_true", help="Changes are permanent" + ) + parser.add_argument( + "--attach", "-a", action="store_true", help="Attach to the VM console" + ) + parser.add_argument( + "--sit", action="store_true", help="Wait for enter key on exceptions" + ) + parser.add_argument( + "--debug", "-d", action="store_true", help="Enable debug output" + ) + parser.add_argument("--root-pw", help="Set root's password") parser.add_argument( "--boot", choices=("efi", "mbr"), default="efi", help="How to boot the image (default: efi)", ) - parser.add_argument("--verbose", "-v", action="store_true", help="Print verbose output") + parser.add_argument( + "--verbose", "-v", action="store_true", help="Print verbose output" + ) parser.add_argument( "--ssh-key", "-i", @@ -1219,8 +1467,12 @@ def __call__( action="append", help="Path to SSH private key (default: generate)", ) - parser.add_argument("--timeout", type=float, help="For startup, in seconds, or 'inf' (default: 30)") - parser.add_argument("--no-network", action="store_true", help="Isolate the VM from the Internet") + parser.add_argument( + "--timeout", type=float, help="For startup, in seconds, or 'inf' (default: 30)" + ) + parser.add_argument( + "--no-network", action="store_true", help="Isolate the VM from the Internet" + ) parser.add_argument( "-L", "-R", @@ -1230,7 +1482,6 @@ def __call__( action=AppendTuple, help="Setup an SSH-style port forward", ) - parser.add_argument("--no-wait", action="store_true", help="Don't wait for ENTER before exiting") parser.add_argument( "--script", "-c", @@ -1244,33 +1495,35 @@ def __call__( metavar="UNIT", action="append", dest="script", - type=(lambda s: f"systemctl enable --now {s!s}"), + type=(lambda s: f"systemctl enable --now {shlex.quote(s)}"), help="Start this systemd unit", ) parser.add_argument("image", type=Path, help="The path to a qcow2 VM image to run") - args = parser.parse_args() - - if args.cloud_init: - user_data = {"users": [{"name": "root", "ssh_authorized_keys": [COCKPIT_TEST_IDENTITY_PUB]}]} - else: - user_data = None + parser.add_argument("cmd", nargs="*") + args = parser.parse_intermixed_args() async def _async_main() -> None: with cli_helper(), IpcDirectory() as ipc: + ui = UI(status_messages=not args.attach, verbose=args.verbose) + async with VirtualMachine( args.image, ipc=ipc, attach_console=args.attach, boot=args.boot, - cloud_init_user_data=user_data, + cloud_init_user_data={ + "chpasswd": {"list": "root:foobar", "expire": False}, + "ssh_pwauth": True, + }, identities=args.ssh_key or (COCKPIT_TEST_IDENTITY,), + journal=print, networks=(() if args.no_network else (Network.user(),)), + provision_ssh_key=not args.maintain, sit=args.sit, snapshot=not args.maintain, - status_messages=not args.attach, timeout=args.timeout, - verbose=args.verbose, + ui=ui, ) as vm: for spec in args.fwd_spec: await vm.forward_port(spec) @@ -1280,8 +1533,11 @@ async def _async_main() -> None: if args.attach: await vm.wait_exit() - elif not args.no_wait: - await vm._ui.sit(f"VM running: {vm.get_id()}") + elif args.cmd: + await vm.execute(*args.cmd, stdin=None, stdout=None) + else: + await _show_ssh_hints(ui, vm.get_id()) + await ui.run("ssh", *vm.ssh_args, vm.get_id(), stdin=None) if args.debug: logging.basicConfig(level=logging.DEBUG)