review: fix shlex escaping, port comment, and CI cache poisoning

Address three issues raised in code review:

- driver.py: use shlex.quote() when building the chpasswd runcmd so
  special characters in username/password cannot break the shell command.

- driver_test.py: clarify port allocation comment; the probe socket is
  released before power.on() which is when QEMU actually binds the port.

- python-tests.yaml: split actions/cache into separate restore and save
  steps; the save step only runs on non-pull_request events, preventing
  untrusted PR code from poisoning the shared cache.

Author: mangelajo

.github/workflows/python-tests.yaml

@@ -120,15 +120,7 @@ jobs:
         run: |
           brew install renode/tap/renode
 
-      - name: Cache Alpine cloud images
-        id: cache-alpine-cloud-images
-        uses: actions/cache@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5
-        with:
-          path: python/packages/jumpstarter-driver-qemu/images
-          key: alpine-3.22.4-uefi-tiny
-
       - name: Download Alpine cloud images
-        if: steps.cache-alpine-cloud-images.outputs.cache-hit != 'true'
         run: |
           for arch in aarch64 x86_64; do
             curl -L --fail --output "python/packages/jumpstarter-driver-qemu/images/nocloud_alpine-3.22.4-${arch}-uefi-tiny-r0.qcow2" \

python/packages/jumpstarter-driver-qemu/jumpstarter_driver_qemu/client.py

@@ -4,6 +4,7 @@
 from contextlib import contextmanager
 
 import click
+from fabric import Connection
 from jumpstarter_driver_composite.client import CompositeClient
 from jumpstarter_driver_network.adapters import FabricAdapter, NovncAdapter
 
@@ -75,12 +76,25 @@ def novnc(self):
 
     @contextmanager
     def shell(self):
-        with FabricAdapter(
-            client=self.ssh,
-            user=self.username,
-            connect_kwargs={"password": self.password},
-        ) as conn:
-            yield conn
+        # If the driver has an 'ssh' hostfwd entry, fetch the actual host port
+        # (resolving any port=0 assignment) and connect directly over TCP.
+        # Otherwise fall back to tunnelling through the jumpstarter stream (vsock).
+        try:
+            port = int(self.call("get_hostfwd_port", "ssh"))
+            with Connection(
+                host="127.0.0.1",
+                port=port,
+                user=self.username,
+                connect_kwargs={"password": self.password},
+            ) as conn:
+                yield conn
+        except KeyError:
+            with FabricAdapter(
+                client=self.ssh,
+                user=self.username,
+                connect_kwargs={"password": self.password},
+            ) as conn:
+                yield conn
 
     def cli(self):
         # Get the base group from CompositeClient which includes all child commands

python/packages/jumpstarter-driver-qemu/jumpstarter_driver_qemu/driver.py

@@ -5,6 +5,7 @@
 import logging
 import os
 import platform
+import shlex
 import shutil
 from collections.abc import AsyncGenerator
 from dataclasses import dataclass, field
@@ -381,6 +382,26 @@ async def on(self) -> None:  # noqa: C901
         Path(self.parent._pty).unlink(missing_ok=True)
         Path(self.parent._pty).symlink_to(pty)
 
+        # Resolve any hostport=0 hostfwd entries to the actual port QEMU chose.
+        # Parse 'info usernet': lines look like "TCP[HOST_FORWARD] fd addr port addr port ..."
+        # Store resolved ports on the parent so get_hostfwd_port() can return them to clients.
+        zero_fwds = {k: v for k, v in self.parent.hostfwd.items() if v.hostport == 0}
+        if zero_fwds:
+            usernet = await qmp.execute("human-monitor-command", {"command-line": "info usernet"})
+            self.logger.debug("info usernet output:\n%s", usernet)
+            for line in usernet.splitlines():
+                parts = line.split()
+                if len(parts) >= 6 and "HOST_FORWARD" in parts[0]:
+                    # parts: Protocol[State] fd hostaddr hostport guestaddr guestport ...
+                    actual_hostaddr, actual_hostport, actual_guestport = parts[2], int(parts[3]), int(parts[5])
+                    for k, v in zero_fwds.items():
+                        if v.hostaddr == actual_hostaddr and v.guestport == actual_guestport:
+                            self.logger.info(
+                                "hostfwd '%s': resolved port 0 -> %s:%d (guest port %d)",
+                                k, actual_hostaddr, actual_hostport, actual_guestport,
+                            )
+                            self.parent._resolved_hostports[k] = actual_hostport
+
         await qmp.execute("system_reset")
         await qmp.disconnect()
 
@@ -410,7 +431,7 @@ def close(self):
 class Hostfwd(BaseModel):
     protocol: Literal["tcp"] = "tcp"
     hostaddr: str = "127.0.0.1"
-    hostport: int = Field(ge=1, le=65535)
+    hostport: int = Field(ge=0, le=65535)  # 0 = let QEMU pick a free port
     guestport: int = Field(ge=1, le=65535)
 
 
@@ -440,6 +461,8 @@ class Qemu(Driver):
     flash_timeout: int = field(default=30 * 60)  # 30 minutes
 
     _tmp_dir: TemporaryDirectory = field(init=False, default_factory=TemporaryDirectory)
+    # Maps hostfwd key -> actual host port after QEMU resolves port 0 assignments
+    _resolved_hostports: dict[str, int] = field(init=False, default_factory=dict)
 
     @classmethod
     def client(cls) -> str:
@@ -533,15 +556,26 @@ def cidata(self) -> TemporaryDirectory:
                     # that do not support plain_text_passwd (e.g. Alpine's tiny-cloud).
                     # cloud-init ignores runcmd entries it doesn't understand, so this
                     # is safe to include unconditionally.
+                    # shlex.quote ensures special characters in credentials are safe.
                     "runcmd": [
-                        f"echo '{self.username}:{self.password}' | chpasswd",
+                        f"printf %s {shlex.quote(f'{self.username}:{self.password}')} | chpasswd",
                     ],
                 }
             )
         )
 
         return tmp
 
+    @export
+    @validate_call(validate_return=True)
+    def get_hostfwd_port(self, key: str) -> int:
+        """Return the actual host port for a hostfwd entry (resolves port 0 assignments)."""
+        if key in self._resolved_hostports:
+            return self._resolved_hostports[key]
+        if key in self.hostfwd:
+            return self.hostfwd[key].hostport
+        raise KeyError(f"hostfwd key {key!r} not found")
+
     @export
     @validate_call(validate_return=True)
     def get_hostname(self) -> str:

python/packages/jumpstarter-driver-qemu/jumpstarter_driver_qemu/driver_test.py

@@ -2,7 +2,6 @@
 import json
 import os
 import platform
-import socket
 import sys
 import tarfile
 from pathlib import Path
@@ -61,20 +60,17 @@ def test_driver_qemu(tmp_path, ovmf):
     arch, ovmf_arch = get_native_arch_config()
 
     # Alpine uses OpenRC (not systemd), so systemd-ssh-generator does not run
-    # and sshd never binds to AF_VSOCK. Use a TCP hostfwd instead so that
-    # qemu.shell() connects over standard TCP SSH, which always works.
-    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as _s:
-        _s.bind(("127.0.0.1", 0))
-        ssh_host_port = _s.getsockname()[1]
-
+    # and sshd never binds to AF_VSOCK. Use a TCP hostfwd with hostport=0 so
+    # QEMU picks a free port automatically; the driver resolves the actual port
+    # from QMP after startup and updates the ssh child accordingly.
     with serve(
         Qemu(
             arch=arch,
             default_partitions={
                 "OVMF_CODE.fd": ovmf / ovmf_arch / "code.fd",
                 "OVMF_VARS.fd": ovmf / ovmf_arch / "vars.fd",
             },
-            hostfwd={"ssh": {"protocol": "tcp", "hostaddr": "127.0.0.1", "hostport": ssh_host_port, "guestport": 22}},
+            hostfwd={"ssh": {"protocol": "tcp", "hostaddr": "127.0.0.1", "hostport": 0, "guestport": 22}},
         )
     ) as qemu:
         hostname = qemu.hostname
@@ -97,11 +93,15 @@ def test_driver_qemu(tmp_path, ovmf):
 
         with qemu.console.pexpect() as p:
             p.logfile = sys.stdout.buffer
-            # Press Enter if GRUB is waiting; if it already booted, go straight to login
-            idx = p.expect_exact(["automatically in ", f"{hostname} login:"], timeout=60)
+            # Press Enter if GRUB is waiting. Both the countdown and bootstrap_complete
+            # can appear before the login prompt, so match whichever comes first.
+            idx = p.expect_exact(["automatically in ", "bootstrap_complete: done"], timeout=600)
             if idx == 0:
+                # GRUB countdown: skip it, then wait for cloud-init to finish
                 p.sendline("")
-                p.expect_exact(f"{hostname} login:", timeout=600)
+                p.expect_exact("bootstrap_complete: done", timeout=600)
+            # tiny-cloud finished: password is set, sshd is ready
+            p.expect_exact(f"{hostname} login:", timeout=60)
             p.sendline(username)
             p.expect_exact("Password:")
             p.sendline(password)