Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 25 additions & 3 deletions sev_verify/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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, "????")
Expand Down
22 changes: 21 additions & 1 deletion sev_verify/guest_vsock.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
import re
import shlex
import socket
import subprocess
import time
from pathlib import Path
from typing import TYPE_CHECKING, Any
Expand Down Expand Up @@ -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
Expand All @@ -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(
Expand All @@ -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)

Expand Down
8 changes: 7 additions & 1 deletion sev_verify/runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
),
Expand Down
12 changes: 11 additions & 1 deletion sev_verify/vm_profile.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"

Expand Down
Loading