diff --git a/sev_verify/cli.py b/sev_verify/cli.py index 1e7677eb..81db41d5 100644 --- a/sev_verify/cli.py +++ b/sev_verify/cli.py @@ -378,7 +378,11 @@ def execute_test( for i, step in enumerate(steps): is_last = i == total_steps - 1 - ctx.profile = profile + # Re-read the profile from the context rather than overwriting it. + # A callable step may replace ctx.profile (dataclasses.replace on a + # frozen VMProfile yields a new object); assigning the stale local + # back over it silently discarded that change for every later step. + profile = ctx.profile ctx.launch = launch if _IS_TTY: @@ -492,6 +496,25 @@ def execute_test( stop_vm(launch) +#: Ordering used to fold many test results into one certification result. +#: A later test must never mask a worse earlier one. "skip" is included for +#: forward compatibility — StepResult already has that state, and a test-level +#: skip belongs between pass and fail. +_RESULT_SEVERITY = {"pass": 0, "skip": 1, "fail": 2, "error": 3} + + +def _worse_result(current: str, candidate: str) -> str: + """Return whichever of the two results is more severe. + + An unrecognised state ranks above every known one, so a typo surfaces + loudly rather than silently masking a real failure. + """ + unknown = max(_RESULT_SEVERITY.values()) + 1 + current_severity = _RESULT_SEVERITY.get(current, unknown) + candidate_severity = _RESULT_SEVERITY.get(candidate, unknown) + return candidate if candidate_severity > current_severity else current + + def execute_certification( cert: CertificationDefinition, guest_path: Path, @@ -529,8 +552,7 @@ def execute_certification( environment=environment, ) test_results.append(tr) - if tr.result != "pass": - overall = tr.result + overall = _worse_result(overall, tr.result) _flush("") icon = _RESULT_LABEL.get(overall, "????") diff --git a/sev_verify/guest_vsock.py b/sev_verify/guest_vsock.py index 6604a95d..c8cf6464 100644 --- a/sev_verify/guest_vsock.py +++ b/sev_verify/guest_vsock.py @@ -14,6 +14,7 @@ import re import shlex import socket +import subprocess import time from pathlib import Path from typing import TYPE_CHECKING, Any @@ -142,9 +143,15 @@ def wait_for_guest( *, timeout: float | None = None, poll_interval: float = 2.0, + process: subprocess.Popen[bytes] | None = None, ) -> None: """ Block until the guest vsock agent responds or timeout is reached. + + Pass ``process`` (the QEMU handle) to stop waiting the moment QEMU exits. + Without it, a guest that never starts — or one the firmware refuses to + launch — costs the full boot timeout and is then reported as an + indistinguishable "agent not ready", discarding the reason QEMU gave. """ boot_timeout = timeout if timeout is not None else profile.vsock_boot_timeout deadline = time.monotonic() + boot_timeout @@ -156,6 +163,11 @@ def wait_for_guest( return except GuestVsockError as exc: last_error = str(exc) + if process is not None and process.poll() is not None: + raise GuestVsockError( + f"QEMU exited with code {process.returncode} before the " + f"guest became ready" + ) from exc time.sleep(poll_interval) raise GuestVsockError( @@ -169,12 +181,20 @@ def check_guest_ready( *, timeout: float | None = None, poll_interval: float = 2.0, + process: subprocess.Popen[bytes] | None = None, ) -> tuple[bool, str]: """ Return whether the guest vsock agent responds. + + ``process`` is forwarded to :func:`wait_for_guest`; see its docstring. """ try: - wait_for_guest(profile, timeout=timeout, poll_interval=poll_interval) + wait_for_guest( + profile, + timeout=timeout, + poll_interval=poll_interval, + process=process, + ) except GuestVsockError as exc: return False, str(exc) diff --git a/sev_verify/runner.py b/sev_verify/runner.py index d019bda8..9a8fd56e 100644 --- a/sev_verify/runner.py +++ b/sev_verify/runner.py @@ -175,10 +175,16 @@ def run_vm_launch_step( launch = profile.vm_launch() except VMLaunchError as exc: duration_ms = int((time.monotonic() - start) * 1000) + # A launch that fails hard is still a launch outcome, so honour the + # step's expected_result. Reporting "error" unconditionally made it + # impossible to write a step that expects a launch to be refused — + # QEMU exiting immediately raises rather than returning ok=False. + passed = _check_expected_values(step, 1, str(exc)) return ( StepResult( step=step, - result="error", + result="pass" if passed else "error", + exit_code=1, stderr=str(exc), duration_ms=duration_ms, ), diff --git a/sev_verify/vm_profile.py b/sev_verify/vm_profile.py index ed500ac5..d042efb3 100644 --- a/sev_verify/vm_profile.py +++ b/sev_verify/vm_profile.py @@ -228,10 +228,20 @@ def vm_launch( # Import here to avoid circular import with guest_vsock. from .guest_vsock import check_guest_ready - booted, boot_error = check_guest_ready(self) + # Passing the process lets the wait abort as soon as QEMU dies, + # instead of polling vsock until the boot timeout expires against + # a VM that no longer exists. + booted, boot_error = check_guest_ready(self, process=process) if not booted: ok = False message = f"Guest did not boot: {boot_error}" + if process.poll() is not None: + # QEMU's own diagnosis is far more useful than "agent not + # ready" — for a rejected launch it names the firmware + # error, e.g. "SNP_LAUNCH_FINISH ... 'Bad measurement'". + stderr_tail = _read_guest_errors(self.guest_error_log).strip() + if stderr_tail: + message = f"{message}\nQEMU stderr:\n{stderr_tail}" elif message == "VM launch verified": message = "VM launched and guest booted"