diff --git a/images/host-centos-10/mkosi.conf b/images/host-centos-10/mkosi.conf index 5d257020..62fbb4e6 100644 --- a/images/host-centos-10/mkosi.conf +++ b/images/host-centos-10/mkosi.conf @@ -22,5 +22,6 @@ Packages= xxd python3 python3-pip + python3-cryptography jq avahi diff --git a/images/host-debian-13/mkosi.conf b/images/host-debian-13/mkosi.conf index 14676970..60c72577 100644 --- a/images/host-debian-13/mkosi.conf +++ b/images/host-debian-13/mkosi.conf @@ -24,6 +24,7 @@ Packages= xxd python3 python3-pip + python3-cryptography python3-emoji jq avahi-daemon diff --git a/images/host-debian-forky/mkosi.conf b/images/host-debian-forky/mkosi.conf index 2c1288db..339bf0ba 100644 --- a/images/host-debian-forky/mkosi.conf +++ b/images/host-debian-forky/mkosi.conf @@ -25,6 +25,7 @@ Packages= python3 python3-dev python3-pip + python3-cryptography python3-emoji g++ jq diff --git a/images/host-fedora-41/mkosi.conf b/images/host-fedora-41/mkosi.conf index 6a20017b..6d7ad213 100644 --- a/images/host-fedora-41/mkosi.conf +++ b/images/host-fedora-41/mkosi.conf @@ -21,6 +21,7 @@ Packages= xxd python3 python3-pip + python3-cryptography python3-emoji jq avahi diff --git a/images/host-opensuse-16.0/mkosi.conf b/images/host-opensuse-16.0/mkosi.conf index 73b1edd7..02f76904 100644 --- a/images/host-opensuse-16.0/mkosi.conf +++ b/images/host-opensuse-16.0/mkosi.conf @@ -28,6 +28,7 @@ Packages= xxd python3 python3-pip + python3-cryptography python3-emoji jq avahi diff --git a/images/host-rocky-10/mkosi.conf b/images/host-rocky-10/mkosi.conf index c138358c..9730fd41 100644 --- a/images/host-rocky-10/mkosi.conf +++ b/images/host-rocky-10/mkosi.conf @@ -20,5 +20,6 @@ Packages= xxd python3 python3-pip + python3-cryptography jq avahi diff --git a/images/host-ubuntu-25.04/mkosi.conf b/images/host-ubuntu-25.04/mkosi.conf index 0d1c74f5..393122dd 100644 --- a/images/host-ubuntu-25.04/mkosi.conf +++ b/images/host-ubuntu-25.04/mkosi.conf @@ -23,6 +23,7 @@ Packages= xxd python3 python3-pip + python3-cryptography python3-emoji jq apt diff --git a/images/host-ubuntu-25.10/mkosi.conf b/images/host-ubuntu-25.10/mkosi.conf index 68989453..90d14036 100644 --- a/images/host-ubuntu-25.10/mkosi.conf +++ b/images/host-ubuntu-25.10/mkosi.conf @@ -23,6 +23,7 @@ Packages= xxd python3 python3-pip + python3-cryptography python3-emoji jq apt diff --git a/images/host-ubuntu-26.04/mkosi.conf b/images/host-ubuntu-26.04/mkosi.conf index 26ac6d91..27bd8571 100644 --- a/images/host-ubuntu-26.04/mkosi.conf +++ b/images/host-ubuntu-26.04/mkosi.conf @@ -24,6 +24,7 @@ Packages= python3 python3-dev python3-pip + python3-cryptography python3-emoji jq apt diff --git a/pyproject.toml b/pyproject.toml index 219735b5..96225539 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -6,6 +6,13 @@ build-backend = "setuptools.build_meta" name = "sev-verify" version = "0.1.0" requires-python = ">=3.11" +dependencies = [ + # Ephemeral P-384 key generation for ID blocks (sev_verify.cvm_props). + # snpguest signs the ID block but cannot generate the keys, so this is + # needed on the host running the harness. Host images install the distro + # package (python3-cryptography) — see images/host-*/mkosi.conf. + "cryptography", +] [project.scripts] sev-verify = "sev_verify.cli:main" diff --git a/sev_verify/README.md b/sev_verify/README.md index 95026c61..2f3893d2 100644 --- a/sev_verify/README.md +++ b/sev_verify/README.md @@ -32,7 +32,12 @@ python3 -m sev_verify /path/to/guest.efi --output-dir /data/sev-artifacts -v 3.0 ## How it works -1. Discover manifests at `cert_tests/*/manifest.toml`. Each manifest declares test entries (name, scope, module path). +1. Discover manifests at `cert_tests/*/manifest.toml`. Each manifest declares test entries: + + - **`name`**, **`description`**, **`module`** — identity and the dotted path to the test module. + - **`scope`** — `host`, `guest`, or `mixed`. Anything other than `host` causes a `VMProfile` to be built (see step 4). + - **`level`** — certification level, e.g. `3.0.0-1`. Several tests may share one level. Also selects the artifacts directory (see [Artifacts directory](#artifacts-directory)). + - **`host_changes`** — set `true` when the test may alter host state that outlives it, such as `snphost commit` advancing the committed TCB floor. Such tests are listed at startup and gated on `--allow-host-changes`. Launching a guest does not count; changing platform configuration does. 2. For each test, import its Python module and call `steps()` to get the ordered list of **`BaseStep`** records. Each has a **`kind`** field (`host`, `guest`, `vm_launch`, …). Define steps with **`Step`** either **chained** (``Step(...).host(command=...)``, …) or **in one call** with ``Step.for_host(...)``, ``Step.for_callable(...)``, etc., so your editor shows every required parameter for that shape. Only the fields relevant to ``kind`` may be set; invalid combinations are rejected at construction. @@ -69,6 +74,11 @@ sev_verify/ Harness package runner.py load_test_execution_plan, run_step, run_vm_launch_step, … vm_profile.py VMProfile, QEMU argv, vm_launch / stop_vm guest_vsock.py vsock command channel to the guest + attestation_report.py Parse report.bin; TCB layout varies by CPU generation + cvm_props.py Measurement + ID block generation shared across tests + environment.py Host component versions recorded in the result + os_info.py Host and guest OS identity (guest read over vsock) + output.py JSON and Markdown result writers cert_tests/ Certification levels common/ Shared test modules snp_ok.py Example host-only test @@ -81,7 +91,25 @@ results/ Output (gitignored) ## Requirements -Python 3.11+ (uses `tomllib` from stdlib). No external packages. +Python 3.11+ (uses `tomllib` from stdlib). + +One external package: **`cryptography`**, used by the ID block tests to generate +the ephemeral P-384 key pairs that sign an ID block. `snpguest` signs and +computes key digests but cannot generate keys, so this cannot be delegated to +the tooling. + +Install it from the distribution, not with pip — the harness runs from the +source tree, so `pyproject.toml`'s dependency list is never consulted unless the +project is actually installed. + +``` +apt install python3-cryptography # Debian / Ubuntu +dnf install python3-cryptography # Fedora / RHEL / CentOS / Rocky +zypper install python3-cryptography # openSUSE +``` + +Host images install it through `Packages=` in `images/host-*/mkosi.conf`, so a +freshly built image needs no extra step. ## Flags diff --git a/sev_verify/attestation_report.py b/sev_verify/attestation_report.py new file mode 100644 index 00000000..a3f9cf35 --- /dev/null +++ b/sev_verify/attestation_report.py @@ -0,0 +1,519 @@ +"""Parse the SEV-SNP ATTESTATION_REPORT binary structure. + +Reading ``report.bin`` directly, rather than regexing ``snpguest display +report``, removes a dependency on a CLI's human-readable output format. The +binary layout is fixed by the SEV-SNP ABI and, unlike the CLI text, the report +is *self-describing*: VERSION is the first four bytes, so every report states +which layout it uses and can be checked on the spot. + +Layout is version-dependent in practice — the report version tracks firmware, +which tracks CPU generation — but versions have only ever *appended* fields. +Everything below 0x188 is common to v2 and v3; the CPUID family/model/stepping +triple at 0x188 exists only in v3+. + +**TCB_VERSION is the exception, and it is not version-dependent but +*generation*-dependent.** The ``sev`` crate decodes it two different ways +(``from_legacy_bytes`` vs ``from_turin_bytes``), and the layouts are +incompatible and indistinguishable from the eight bytes alone: + +=========== ===================== ==================== +byte Milan / Genoa Turin / Venice +=========== ===================== ==================== +0 BOOT_LOADER FMC +1 TEE BOOT_LOADER +2 -- TEE +3 -- SNP +6 SNP -- +7 MICROCODE MICROCODE +=========== ===================== ==================== + +Decoding therefore requires knowing the processor generation. The authoritative +source is the **host's** CPUID — a report parsed here was produced by a guest on +this machine, and unlike the report's own CPUID copy it is present regardless of +report version. :func:`host_generation` reads it; :func:`parse` takes the result +as an argument so the parser itself stays a pure function of its input and can +be unit-tested without hardware. + +When the report is v3+ it also carries a CPUID copy, which :func:`parse` uses as +a cross-check — but only when it can be resolved. Firmware does not always fill +it in: SEV firmware 1.55 build 38 leaves all three bytes zero in version-3 +reports, and build 39 populates them. A report like that is perfectly decodable +using the host's generation, so it is decoded, and the failed cross-check is +recorded in ``cpuid_note`` rather than raised. Refusing it would reject a usable +report over a field the platform declined to fill. + +The error is reserved for the case that actually indicates a problem: both the +host's and the report's CPUID resolve to validated generations, and they +disagree. Then the report did not come from this machine and neither layout can +be trusted for it. Where no generation is supplied at all, the report's own +CPUID is used if it resolves; if it does not, TCB_VERSION is left undecoded, +matching what a v2 report — which carries no CPUID — already does. + +An unrecognised processor still raises when it is the *only* source, since +guessing a layout would produce plausible-looking but wrong values with no error. + +Report *versions* are treated more leniently than processors, deliberately. A +newer version is decoded with the newest validated offsets and the assumption +recorded in ``version_note``; only versions older than the validated range are +refused. The asymmetry is the point: misreading a generation corrupts values +silently, whereas an unrecognised version at worst leaves new fields unread, +and the version — unlike the generation — is stated in the report itself. + +Offsets are confirmed against real reports rather than read off a spec. The +first such validation used a v3 report from an EPYC 9654 (Genoa, CPUID +19h/11h), cross-checked against independently known values: +GUEST_SVN/POLICY/FAMILY_ID/IMAGE_ID against the values the ID block was built +with, REPORTED_TCB against ``snphost ok``, AUTHOR_KEY_DIGEST against the known +all-zero author key, and CPUID against the CPU model. + +The second used a **version 5** report from an EPYC 9575F (Turin, CPUID +1Ah/02h), which validated the Turin TCB layout for the first time. Its +REPORTED_TCB bytes were ``0103020600000062``, decoding under the Turin layout to +``bootloader=3 tee=2 snp=6 microcode=98 fmc=1`` — matching ``snphost show tcb`` +exactly. Decoded under the legacy layout the same bytes give +``bootloader=1 tee=3 snp=0 microcode=98`` with no FMC: plausible values, silently +wrong, which is precisely the failure this generation gate exists to prevent. +That report also confirmed v5 moved none of the fields read here. + +As further processors are exercised, extend :data:`SUPPORTED_GENERATIONS` and +record the validation here. + +.. note:: + + This module is a **short-term stand-in**. The right long-term source is + ``snpguest``, which already parses reports correctly and generation-aware via + the ``sev`` crate but currently exposes them only as human-readable text + (``println!("{}", att_report)``). Once snpguest gains machine-readable + output, this module should be replaced by consuming that output rather than + extended to cover further processor generations. +""" + +from __future__ import annotations + +import struct +from dataclasses import dataclass +from pathlib import Path + +#: ATTESTATION_REPORT is a fixed-size structure. +REPORT_SIZE = 1184 + +#: Report versions whose layout we read. v3 and v5 are verified on hardware; +#: v2 shares the same layout for every field below 0x188. +#: +#: Membership here means *validated*, not *accepted*. A newer version is still +#: decoded — with these offsets, and a note on the result recording the +#: assumption — because versions have only ever appended fields, so the cost of +#: being wrong is missing something new rather than misreading something old. +#: Only versions *older* than this set are refused, where fields may genuinely +#: not exist. See :func:`parse` for why that is a weaker stance than the one +#: :data:`SUPPORTED_GENERATIONS` takes. +#: +#: Note this is an axis independent of processor generation. The version decides +#: which fields exist and where; the generation decides how TCB_VERSION's eight +#: bytes are ordered. A v3 report can come from either a legacy-layout or a +#: Turin-layout processor, so both gates are needed and neither implies the +#: other. +#: +#: v5 was added after a real v5 report from an EPYC 9575F decoded correctly at +#: these offsets — REPORTED_TCB matched ``snphost show tcb`` and CPUID matched +#: the host's, confirming its additions moved nothing we read. v4 has never been +#: seen; it decodes on the append-only assumption and says so. To promote a +#: version to validated: +#: +#: 1. Check whether it shares framing with a version already listed. The +#: ``sev`` crate's ``ReportVariant`` mapping groups versions by layout — +#: currently ``2 => V2``, ``3 | 4 => V3``, ``_ => V5`` — so a version +#: sharing a variant with one listed here reads identically. That makes v4 +#: the cheap case and v5 the one needing real scrutiny. +#: 2. Remember additions are not always new offsets. v5 adds +#: ``page_swap_disabled`` to GuestPolicy and SEV-TIO to PlatformInfo, which +#: are new *bits in existing fields* and move nothing. +#: 3. Parse a real report of that version and check decoded values against +#: independently known ones, as the module docstring records for v3. +#: 4. Add the version here and record the validation in the docstring. +KNOWN_VERSIONS = frozenset({2, 3, 5}) + +#: TCB_VERSION byte layouts. See the module docstring for the two orderings. +TCB_LAYOUT_LEGACY = "legacy" +TCB_LAYOUT_TURIN = "turin" + +#: Processor generations this module has been **validated against**, keyed by +#: CPUID family and an inclusive model range. +#: +#: This is deliberately a record of what has been exercised on real hardware, +#: not of what we believe we could decode. A certification harness reporting a +#: pass on silicon it has never run on is the failure this gate exists to +#: prevent, so an unrecognised processor raises rather than being decoded on +#: the assumption that a transcribed layout is right. +#: +#: Keyed on family/model *pairs*, not family/model/stepping triples: AMD scopes +#: SEV firmware images by family and model only — ``amd_sev_fam19h_model1xh``, +#: ``amd_sev_fam1ah_model0xh`` — and stepping appears nowhere in that +#: partitioning. snpguest's ``get_processor_model`` (``src/fetch.rs``) splits +#: the same way for VCEK lookup. +#: +#: To add a generation: run the ID block test on that hardware, confirm the +#: decoded fields against ``snphost show tcb`` and the values the ID block was +#: built with, then add the entry and note the validation in the docstring. +#: The layouts for generations not yet exercised here, taken from the ``sev`` +#: crate, are: +#: +#: 0x19 / 0x00-0x0F Milan legacy +#: 0x19 / 0xA0-0xAF Bergamo/Siena legacy +SUPPORTED_GENERATIONS: tuple[tuple[int, range, str, str], ...] = ( + # (cpuid_family, model range, name, TCB layout) + (0x19, range(0x10, 0x20), "Genoa", TCB_LAYOUT_LEGACY), # EPYC 9654, v3 reports + (0x1A, range(0x00, 0x12), "Turin", TCB_LAYOUT_TURIN), # EPYC 9575F, v5 reports +) + +# Field offsets. See module docstring for how these were validated. +_OFF_VERSION = 0x000 +_OFF_GUEST_SVN = 0x004 +_OFF_POLICY = 0x008 +_OFF_FAMILY_ID = 0x010 +_OFF_IMAGE_ID = 0x020 +_OFF_VMPL = 0x030 +_OFF_REPORT_DATA = 0x050 +_OFF_MEASUREMENT = 0x090 +_OFF_HOST_DATA = 0x0C0 +_OFF_ID_KEY_DIGEST = 0x0E0 +_OFF_AUTHOR_KEY_DIGEST = 0x110 +_OFF_REPORT_ID = 0x140 +_OFF_REPORTED_TCB = 0x180 +_OFF_CPUID_FAM = 0x188 # v3+ + +_LEN_ID = 16 +_LEN_MEASUREMENT = 48 +_LEN_DIGEST = 48 +_LEN_REPORT_DATA = 64 +_LEN_HOST_DATA = 32 +_LEN_REPORT_ID = 32 + + +class ReportError(Exception): + """Base class for attestation report problems.""" + + +class ReportMalformed(ReportError): + """The file is not a well-formed ATTESTATION_REPORT.""" + + +class ReportUnsupportedVersion(ReportError): + """The report declares a version whose layout we have not validated.""" + + +class ReportUnsupportedCpu(ReportError): + """The report comes from a processor this module has not been validated on. + + Raised rather than decoding on the assumption that a transcribed layout is + correct — TCB_VERSION in particular is laid out differently on Turin, so a + wrong guess yields plausible values rather than an error. + """ + + +@dataclass(frozen=True) +class TcbVersion: + """Decoded SNP TCB_VERSION — the same values ``snphost ok`` prints. + + ``fmc`` exists only on Turin and later; it is ``None`` elsewhere, matching + the ``sev`` crate's ``Option``. + """ + + bootloader: int + tee: int + snp: int + microcode: int + fmc: int | None = None + + @classmethod + def from_bytes(cls, raw: bytes, layout: str) -> TcbVersion: + """Decode the 8-byte TCB_VERSION using the given generation layout.""" + if layout == TCB_LAYOUT_TURIN: + # byte 0 FMC, 1 BOOT_LOADER, 2 TEE, 3 SNP, 7 MICROCODE. + return cls( + fmc=raw[0], + bootloader=raw[1], + tee=raw[2], + snp=raw[3], + microcode=raw[7], + ) + if layout == TCB_LAYOUT_LEGACY: + # byte 0 BOOT_LOADER, 1 TEE, bytes 2-5 reserved, 6 SNP, 7 MICROCODE. + return cls(bootloader=raw[0], tee=raw[1], snp=raw[6], microcode=raw[7]) + raise ReportUnsupportedCpu(f"unknown TCB layout {layout!r}") + + def __str__(self) -> str: + base = ( + f"bootloader={self.bootloader} tee={self.tee} " + f"snp={self.snp} microcode={self.microcode}" + ) + return base if self.fmc is None else f"fmc={self.fmc} {base}" + + +@dataclass(frozen=True) +class AttestationReport: + """The fields of an ATTESTATION_REPORT that we read.""" + + version: int + guest_svn: int + policy: int + family_id: bytes + image_id: bytes + vmpl: int + report_data: bytes + measurement: bytes + host_data: bytes + id_key_digest: bytes + author_key_digest: bytes + report_id: bytes + #: ``None`` when the processor generation was unknown, since the byte + #: layout differs between generations and cannot be guessed. + reported_tcb: TcbVersion | None + #: (family, model, stepping) — v3+ only, None on older reports. + cpuid: tuple[int, int, int] | None + #: Validated processor generation this report was decoded as, or "unknown". + generation: str + #: Set when the report's own CPUID could not be used and the host's was + #: preferred — for instance when firmware leaves those bytes zero. ``None`` + #: when the report's CPUID was absent by design (v2) or agreed with the host. + cpuid_note: str | None = None + #: Set when the report declared a version newer than any validated here and + #: was decoded with the newest known field offsets. ``None`` when the + #: version was one this parser has been checked against. + version_note: str | None = None + + @property + def id_block_used(self) -> bool: + """True when ID_KEY_DIGEST is set, i.e. the guest launched with an ID block.""" + return any(self.id_key_digest) + + +def resolve_generation(family: int, model: int) -> tuple[str, str]: + """Return ``(name, tcb_layout)`` for a CPUID family/model pair. + + Raises: + ReportUnsupportedCpu: the pair is not in :data:`SUPPORTED_GENERATIONS`. + """ + for fam, models, name, layout in SUPPORTED_GENERATIONS: + if family == fam and model in models: + return name, layout + + validated = ", ".join( + f"{name} (family 0x{fam:02X} model 0x{models[0]:02X}-0x{models[-1]:02X})" + for fam, models, name, _ in SUPPORTED_GENERATIONS + ) + raise ReportUnsupportedCpu( + f"CPUID family 0x{family:02X} model 0x{model:02X} has not been validated " + f"against. Validated: {validated}. TCB_VERSION is laid out differently " + f"across processor generations, so decoding anyway would produce " + f"plausible but wrong values. See SUPPORTED_GENERATIONS in " + f"sev_verify/attestation_report.py." + ) + + +def host_generation() -> tuple[str, str]: + """Return ``(name, tcb_layout)`` for the CPU this process is running on. + + Reads ``/proc/cpuinfo``. This is the authoritative source when parsing a + report produced by a guest on this machine: unlike the report's CPUID copy + it is present regardless of report version. + + Raises: + ReportUnsupportedCpu: family/model unreadable, or not validated. + """ + family = model = None + try: + with open("/proc/cpuinfo", encoding="utf-8") as f: + for line in f: + key, _, value = line.partition(":") + key = key.strip() + if key == "cpu family" and family is None: + family = int(value.strip()) + elif key == "model" and model is None: + model = int(value.strip()) + if family is not None and model is not None: + break + except OSError as exc: + raise ReportUnsupportedCpu(f"could not read /proc/cpuinfo: {exc}") from exc + + if family is None or model is None: + raise ReportUnsupportedCpu( + "could not determine CPU family/model from /proc/cpuinfo" + ) + return resolve_generation(family, model) + + +def parse( + data: bytes, *, generation: tuple[str, str] | None = None +) -> AttestationReport: + """Parse raw report bytes. + + Args: + data: the raw 1184-byte ATTESTATION_REPORT. + generation: ``(name, tcb_layout)``, normally from :func:`host_generation`. + Decides the TCB_VERSION byte layout. If omitted, it is taken from + the report's own CPUID when present (v3+); a v2 report then leaves + ``reported_tcb`` as ``None`` rather than guessing a layout. + + Raises: + ReportMalformed: wrong size. + ReportUnsupportedVersion: layout not validated for that version. + ReportUnsupportedCpu: the report's CPUID disagrees with *generation*, or + names a processor that has not been validated against. + """ + if len(data) != REPORT_SIZE: + raise ReportMalformed( + f"expected a {REPORT_SIZE}-byte ATTESTATION_REPORT, got {len(data)} bytes" + ) + + (version,) = struct.unpack_from(" bytes: + return data[off:off + length] + + cpuid = None + if version >= 3: + cpuid = ( + data[_OFF_CPUID_FAM], + data[_OFF_CPUID_FAM + 1], + data[_OFF_CPUID_FAM + 2], + ) + + # Resolve the generation that decides the TCB_VERSION layout. + # + # The host's CPUID is authoritative when supplied: it describes the silicon + # this code is running on, which is the thing the layout actually depends + # on. The report's copy is a cross-check, and only a useful one when it can + # be resolved — firmware does not always populate it. Observed on SEV + # firmware 1.55 build 38, which leaves all three bytes zero in version-3 + # reports; build 39 fills them in. Refusing such a report would reject a + # decodable one over a field the platform declined to fill. + cpuid_note: str | None = None + if generation is not None and cpuid is not None: + try: + report_gen = resolve_generation(cpuid[0], cpuid[1]) + except ReportUnsupportedCpu: + # Unresolvable, so it contradicts nothing. Decode with the host's + # generation and record that the cross-check could not be made. + cpuid_note = ( + f"report CPUID family 0x{cpuid[0]:02X} model 0x{cpuid[1]:02X} " + f"stepping 0x{cpuid[2]:02X} does not resolve to a known " + f"generation; decoded as {generation[0]} from the host instead" + ) + else: + if report_gen != generation: + # Both resolve, and disagree: the report is not from this + # machine, and neither layout can be trusted for it. + # + # Note this branch is only reachable once SUPPORTED_GENERATIONS + # holds more than one entry. With a single validated generation + # every disagreeing CPUID is unresolvable instead, and takes the + # branch above. That is the conservative order: a report is only + # called foreign when both generations are ones we have actually + # validated against. + raise ReportUnsupportedCpu( + f"report CPUID family 0x{cpuid[0]:02X} model " + f"0x{cpuid[1]:02X} resolves to {report_gen[0]}, but this " + f"host is {generation[0]}. The report does not appear to " + f"come from this machine." + ) + elif generation is None and cpuid is not None: + # No host generation to fall back on. An unresolvable CPUID then leaves + # nothing to choose a layout with, so the TCB is left undecoded — the + # same outcome as a v2 report, which carries no CPUID at all — rather + # than raising for a v3 report where v2 would have been tolerated. + try: + generation = resolve_generation(cpuid[0], cpuid[1]) + except ReportUnsupportedCpu: + cpuid_note = ( + f"report CPUID family 0x{cpuid[0]:02X} model 0x{cpuid[1]:02X} " + f"stepping 0x{cpuid[2]:02X} does not resolve to a known " + f"generation and no host generation was supplied; " + f"TCB_VERSION left undecoded" + ) + + if generation is not None: + gen_name, tcb_layout = generation + reported_tcb = TcbVersion.from_bytes( + data[_OFF_REPORTED_TCB:_OFF_REPORTED_TCB + 8], tcb_layout + ) + else: + # v2 report and no generation supplied — the TCB layout is unknowable, + # so leave it undecoded rather than assume one. + gen_name, reported_tcb = "unknown", None + + return AttestationReport( + version=version, + guest_svn=guest_svn, + policy=policy, + family_id=field(_OFF_FAMILY_ID, _LEN_ID), + image_id=field(_OFF_IMAGE_ID, _LEN_ID), + vmpl=vmpl, + report_data=field(_OFF_REPORT_DATA, _LEN_REPORT_DATA), + measurement=field(_OFF_MEASUREMENT, _LEN_MEASUREMENT), + host_data=field(_OFF_HOST_DATA, _LEN_HOST_DATA), + id_key_digest=field(_OFF_ID_KEY_DIGEST, _LEN_DIGEST), + author_key_digest=field(_OFF_AUTHOR_KEY_DIGEST, _LEN_DIGEST), + report_id=field(_OFF_REPORT_ID, _LEN_REPORT_ID), + reported_tcb=reported_tcb, + cpuid=cpuid, + generation=gen_name, + cpuid_note=cpuid_note, + version_note=version_note, + ) + + +def read( + path: Path, *, generation: tuple[str, str] | None = None +) -> AttestationReport: + """Read and parse an ATTESTATION_REPORT file. + + Args: + path: the report file. + generation: forwarded to :func:`parse`; see its docstring. + + Raises: + ReportMalformed: file missing or wrong size. + ReportUnsupportedVersion: layout not validated for that version. + ReportUnsupportedCpu: CPUID mismatch, or processor not validated. + """ + try: + data = path.read_bytes() + except FileNotFoundError as exc: + raise ReportMalformed(f"{path.name} not found") from exc + return parse(data, generation=generation) diff --git a/sev_verify/cert_tests/c3_0/c3_0_0_0/attestation_test.py b/sev_verify/cert_tests/c3_0/c3_0_0_0/attestation_test.py index 4dc18888..8e596f46 100644 --- a/sev_verify/cert_tests/c3_0/c3_0_0_0/attestation_test.py +++ b/sev_verify/cert_tests/c3_0/c3_0_0_0/attestation_test.py @@ -18,62 +18,23 @@ """ import subprocess -from pathlib import Path +# calculate_measurement is imported, not redefined: it is shared with the ID +# block test via cvm_props, and the handler for the step below resolves by name +# on this module, so the import is what makes it available. +from sev_verify.cvm_props import ( + MeasurementError, + calculate_measurement, + read_measurement, +) from sev_verify.models import BaseStep, Step, StepContext, StepHandlerResult -from sev_verify.vm_profile import VMProfile, VMProfileError +from sev_verify.vm_profile import VMProfile vm_profile = VMProfile( image_path="", memory_mb=4096, ) -def calculate_measurement(ctx: StepContext) -> StepHandlerResult: - """ - Calculate expected measurement using ``snpguest generate measurement``. - - Searches for an AMD SEV-compatible OVMF binary and runs snpguest to - produce a hex measurement of the guest image, stored in - ``ctx.expected_measurement`` for later attestation comparison. - """ - measurement_file = ctx.artifact_dir / "guest_measurement.txt" - ovmf_path = None - - try: - ovmf_path = Path(ctx.profile.resolved_ovmf_path()) - except VMProfileError as e: - return StepHandlerResult( - exit_code=1, - stderr=str(e), - ) - - result = subprocess.run( - [ - "snpguest", "generate", "measurement", - "--vcpu-type", "EPYC-v4", - "--ovmf", str(ovmf_path), - "--kernel", str(ctx.guest_path), - "--output-format", "hex", - "--measurement-file", str(measurement_file), - ], - capture_output=True, - text=True, - check=False, - ) - if result.returncode != 0: - return StepHandlerResult( - exit_code=result.returncode, - stdout=result.stdout, - stderr=result.stderr, - ) - - expected_measurement = measurement_file.read_text().strip() - return StepHandlerResult( - exit_code=0, - stdout=f"Calculated expected measurement: {expected_measurement}", - ) - - def verify_report_fields(ctx: StepContext) -> StepHandlerResult: """ Example callable step: validate ``report.bin`` after ``guest_pull``. @@ -82,10 +43,13 @@ def verify_report_fields(ctx: StepContext) -> StepHandlerResult: to values computed in earlier ``callable`` or ``host`` steps. """ report_file = ctx.artifact_dir / "report.bin" - measurement_file = ctx.artifact_dir / "guest_measurement.txt" request_file = ctx.artifact_dir / "request.bin" - expected_measurement = measurement_file.read_text().strip() + try: + expected_measurement = f"0x{read_measurement(ctx.artifact_dir)}" + except MeasurementError as exc: + return StepHandlerResult(exit_code=1, stderr=str(exc)) + request_data = "0x" + str(request_file.read_bytes().hex()) result = subprocess.run( [ diff --git a/sev_verify/cert_tests/c3_0/c3_0_0_2/__init__.py b/sev_verify/cert_tests/c3_0/c3_0_0_2/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/sev_verify/cert_tests/c3_0/c3_0_0_2/id_block_test.py b/sev_verify/cert_tests/c3_0/c3_0_0_2/id_block_test.py new file mode 100644 index 00000000..6d7bd129 --- /dev/null +++ b/sev_verify/cert_tests/c3_0/c3_0_0_2/id_block_test.py @@ -0,0 +1,475 @@ +"""id_block_test: Verify ID block acceptance, report field binding, and rejection. + +Positive path: launch an SEV-SNP guest with a valid ID block, fetch the +attestation report, and verify that the hardware report reflects the ID block +fields (guest_svn, policy, family_id, image_id). + +Negative path: attempt three launches that must fail: + 1. ID block with a corrupted measurement (digest mismatch) + 2. Policy incompatible with the platform (SMT=0 on an SMT-active host) + 3. Impossibly high ABI major version (ABI_MAJOR=255) +""" + +from __future__ import annotations + +import subprocess +import tempfile +from dataclasses import replace +from pathlib import Path + +from cryptography.hazmat.primitives.asymmetric import ec +from cryptography.hazmat.primitives.serialization import ( + Encoding, + NoEncryption, + PrivateFormat, +) + +from sev_verify import attestation_report +from sev_verify.cvm_props import ( + IdBlockMetadataError, + MeasurementError, + calculate_measurement, + generate_id_block, + read_id_block_metadata, + read_measurement, +) +from sev_verify.models import BaseStep, Step, StepContext, StepHandlerResult +from sev_verify.vm_profile import VMProfile + +vm_profile = VMProfile( + image_path="", + memory_mb=2048, +) + + +# ── Report field verification ───────────────────────────────────────────────── + + +def verify_id_block_fields(ctx: StepContext) -> StepHandlerResult: + """Compare ID block fields in the hardware attestation report to expectations. + + Reads report.bin directly (see :mod:`sev_verify.attestation_report`) rather + than parsing ``snpguest display report`` output, so the check does not + depend on a CLI's human-readable formatting. + + The host's processor generation is passed in so the parser can cross-check + it against the CPUID the report carries, and so TCB_VERSION — whose byte + layout differs by generation — is never decoded on a guess. + """ + try: + report = attestation_report.read( + ctx.artifact_dir / "report.bin", + generation=attestation_report.host_generation(), + ) + except attestation_report.ReportError as exc: + return StepHandlerResult(exit_code=1, stderr=str(exc)) + + # Read through the same helper the generator used, so the expectations here + # cannot drift from the values the ID block was actually built with. + try: + meta = read_id_block_metadata() + except IdBlockMetadataError as exc: + return StepHandlerResult(exit_code=1, stderr=str(exc)) + + errors = [] + if report.guest_svn != meta.guest_svn: + errors.append(f"guest_svn: expected {meta.guest_svn}, got {report.guest_svn}") + if report.policy != meta.policy: + errors.append( + f"policy: expected {hex(meta.policy)}, got {hex(report.policy)}" + ) + if report.family_id != meta.family_id_bytes: + errors.append( + f"family_id: expected {meta.family_id_bytes.hex()}, " + f"got {report.family_id.hex()}" + ) + if report.image_id != meta.image_id_bytes: + errors.append( + f"image_id: expected {meta.image_id_bytes.hex()}, " + f"got {report.image_id.hex()}" + ) + # An all-zero ID_KEY_DIGEST means the guest launched without an ID block at + # all. The four comparisons above would then all fail with zeros, which is + # a confusing way to report "no ID block was used". + if not report.id_block_used: + errors.append( + "id_key_digest is all zero — the guest launched without an ID block" + ) + + if errors: + return StepHandlerResult(exit_code=1, stderr="\n".join(errors)) + return StepHandlerResult( + exit_code=0, + stdout=( + f"All ID block fields match: svn={meta.guest_svn} " + f"policy={hex(meta.policy)} family_id={meta.family_id!r} " + f"image_id={meta.image_id!r}\n" + f" report v{report.version} vmpl={report.vmpl} " + f"cpuid={report.cpuid} gen={report.generation} " + f"tcb=({report.reported_tcb})\n" + f" id_key_digest={report.id_key_digest.hex()[:32]}..." + ), + ) + + +# ── Negative-test profile mutation helpers ──────────────────────────────────── + + +def _regenerate_id_block( + ctx: StepContext, measurement: str, policy: int, +) -> StepHandlerResult: + """Generate a fresh ID block with the given measurement and policy, update ctx.profile. + + ``measurement`` must be in snpguest's input form — 0x-prefixed hex. An + unprefixed string is decoded as base64, not hex. + + Only the policy varies between the negative cases; the identifying fields + come from the same validated source the original ID block was built from. + """ + try: + meta = read_id_block_metadata() + except IdBlockMetadataError as exc: + return StepHandlerResult(exit_code=1, stderr=str(exc)) + + id_key = ec.generate_private_key(ec.SECP384R1()) + auth_key = ec.generate_private_key(ec.SECP384R1()) + + id_block_file = ctx.artifact_dir / "neg-id-block.b64" + id_auth_file = ctx.artifact_dir / "neg-id-auth.b64" + + with tempfile.TemporaryDirectory() as tmpdir: + id_key_path = Path(tmpdir) / "id-key.pem" + auth_key_path = Path(tmpdir) / "auth-key.pem" + id_key_path.write_bytes( + id_key.private_bytes(Encoding.PEM, PrivateFormat.TraditionalOpenSSL, NoEncryption()) + ) + auth_key_path.write_bytes( + auth_key.private_bytes(Encoding.PEM, PrivateFormat.TraditionalOpenSSL, NoEncryption()) + ) + + result = subprocess.run( + [ + "snpguest", "generate", "id-block", + str(id_key_path), str(auth_key_path), + measurement, + "--family-id", meta.family_id, + "--image-id", meta.image_id, + "--svn", str(meta.guest_svn), + "--policy", hex(policy), + "--id-file", str(id_block_file), + "--auth-file", str(id_auth_file), + ], + capture_output=True, text=True, check=False, + ) + + if result.returncode != 0: + return StepHandlerResult( + exit_code=1, + stderr=f"snpguest generate id-block failed:\n{result.stderr}", + ) + + ctx.profile = replace( + ctx.profile, + id_block=id_block_file.read_text().strip(), + id_auth=id_auth_file.read_text().strip(), + policy=policy, + ) + return StepHandlerResult(exit_code=0) + + +def set_bad_measurement(ctx: StepContext) -> StepHandlerResult: + """Regenerate the ID block with a corrupted measurement to cause digest mismatch.""" + try: + real = read_measurement(ctx.artifact_dir) + except MeasurementError as exc: + return StepHandlerResult(exit_code=1, stderr=str(exc)) + + try: + meta = read_id_block_metadata() + except IdBlockMetadataError as exc: + return StepHandlerResult(exit_code=1, stderr=str(exc)) + + # Flip the first byte of the digest + flipped_byte = "00" if real[:2].lower() != "00" else "ff" + flipped = flipped_byte + real[2:] + + hr = _regenerate_id_block(ctx, f"0x{flipped}", meta.policy) + if hr.exit_code != 0: + return hr + return StepHandlerResult( + exit_code=0, + stdout=f"Set bad measurement: {flipped[:16]}... (real: {real[:16]}...)", + ) + + +_SMT_ACTIVE = Path("/sys/devices/system/cpu/smt/active") +_SMT_CONTROL = Path("/sys/devices/system/cpu/smt/control") + +#: smt/control values that explain *why* SMT is inactive. The first two mean it +#: was switched off, the last two that the capability is absent — a distinction +#: worth preserving in the report, since only the former could have been on. +_SMT_CONTROL_REASONS = { + "off": "SMT is disabled on this host", + "forceoff": "SMT is force-disabled and cannot be re-enabled without a reboot", + "notsupported": "this processor does not support SMT", + "notimplemented": "this architecture does not implement SMT runtime control", +} + + +def _read_sysfs(path: Path) -> str | None: + """Return the stripped contents of *path*, or None if it cannot be read.""" + try: + return path.read_text().strip() + except OSError: + return None + + +def _smt_status() -> tuple[bool, str]: + """Return whether host SMT is active, with a reason suitable for reporting. + + ``smt/active`` decides: it reports whether sibling threads are online right + now. ``smt/control`` is consulted only to explain why they are not, and is + deliberately not used as the decision — besides the four names above it can + also read as a thread count on architectures with partial SMT states. + """ + active = _read_sysfs(_SMT_ACTIVE) + control = _read_sysfs(_SMT_CONTROL) + + if active == "1": + return True, "SMT is active on this host" + + reason = _SMT_CONTROL_REASONS.get(control or "") + if reason is not None: + return False, reason + if active is None: + return False, ( + f"{_SMT_ACTIVE} is not present, so SMT state cannot be determined" + ) + return False, "SMT is not active on this host" + + +def smt_case_not_applicable(ctx: StepContext) -> StepHandlerResult: + """Record that the SMT policy case was left out of this run. + + Emitted in place of the SMT steps when the host cannot exercise them, so + that the omission appears in the results rather than the case simply being + absent. + """ + _, reason = _smt_status() + return StepHandlerResult( + exit_code=0, + stdout=f"SMT policy rejection case not run: {reason}", + ) + + +def set_incompatible_policy(ctx: StepContext) -> StepHandlerResult: + """Regenerate the ID block with a policy the platform cannot satisfy. + + Regenerates the ID block (and QEMU launch policy) with SMT=0 — the firmware + must reject because the platform cannot guarantee single-threaded execution. + + Only reached on an SMT-active host; steps() omits this case otherwise. The + check is repeated here so the handler is correct on its own terms rather + than relying on the caller. + """ + smt_active, reason = _smt_status() + if not smt_active: + return StepHandlerResult( + exit_code=1, + stderr=f"Cannot test SMT policy incompatibility: {reason}", + ) + + try: + measurement = read_measurement(ctx.artifact_dir) + except MeasurementError as exc: + return StepHandlerResult(exit_code=1, stderr=str(exc)) + + try: + meta = read_id_block_metadata() + except IdBlockMetadataError as exc: + return StepHandlerResult(exit_code=1, stderr=str(exc)) + + # Clear SMT bit (16) — guest demands no SMT, but host has SMT active + incompatible_policy = meta.policy & ~(1 << 16) + + hr = _regenerate_id_block(ctx, f"0x{measurement}", incompatible_policy) + if hr.exit_code != 0: + return hr + return StepHandlerResult( + exit_code=0, + stdout=f"Set incompatible policy {hex(incompatible_policy)} (SMT=0, host SMT active)", + ) + + +def set_bad_abi_version(ctx: StepContext) -> StepHandlerResult: + """Regenerate the ID block with an impossibly high ABI major version. + + The policy's ABI_MAJOR field (bits 15:8) specifies the minimum firmware + ABI version required. Setting it to 255 guarantees the firmware cannot + satisfy the requirement on any current platform. + """ + try: + measurement = read_measurement(ctx.artifact_dir) + except MeasurementError as exc: + return StepHandlerResult(exit_code=1, stderr=str(exc)) + + try: + meta = read_id_block_metadata() + except IdBlockMetadataError as exc: + return StepHandlerResult(exit_code=1, stderr=str(exc)) + + # Set ABI_MAJOR (bits 15:8) to 255 + bad_policy = (meta.policy & ~0xFF00) | (0xFF << 8) + + hr = _regenerate_id_block(ctx, f"0x{measurement}", bad_policy) + if hr.exit_code != 0: + return hr + return StepHandlerResult( + exit_code=0, + stdout=f"Set policy {hex(bad_policy)} (ABI_MAJOR=255)", + ) + + +# ── Steps ───────────────────────────────────────────────────────────────────── + + +def steps() -> list[BaseStep]: + smt_active, _ = _smt_status() + + steps_list: list[BaseStep] = [ + # ── Positive: launch with valid ID block, verify report fields ── + Step.for_callable( + name="Calculate measurement", + type="setup", + handler="calculate_measurement", + timeout=60, + ), + Step.for_callable( + name="Generate ID block", + type="setup", + handler="generate_id_block", + timeout=30, + ), + Step.for_vm_launch( + name="Launch with valid ID block", + type="setup", + timeout=300, + ).add_hint( + "Address already in use", + "A previous VM may still be running. " + "Try: sudo kill $(pgrep -f 'qemu.*guest-cid')", + ), + Step.for_guest( + name="Get attestation report", + type="required", + command="snpguest report report.bin request.bin --random", + timeout=60, + ), + Step.for_guest_pull( + name="Pull attestation report", + type="required", + guest_src="report.bin", + host_dest="report.bin", + timeout=120, + ), + Step.for_vm_stop( + name="Stop VM", + type="info", + timeout=60, + ), + Step.for_callable( + name="Verify ID block fields in report", + type="required", + handler="verify_id_block_fields", + timeout=30, + ), + + # ── Negative: bad measurement (digest mismatch) ── + Step.for_callable( + name="Set bad measurement in ID block", + type="required", + handler="set_bad_measurement", + timeout=30, + ), + Step.for_vm_launch( + name="Launch with bad measurement (expect rejection)", + type="required", + # Assert the firmware's own reason, not merely that something + # failed: exit_code:1 alone is also satisfied by a boot timeout, so + # it cannot distinguish a real rejection from a hung guest. + expected_result="stdout_contains:Bad measurement", + timeout=300, + ), + Step.for_vm_stop( + name="Stop VM (after bad measurement)", + type="info", + timeout=60, + ), + ] + + # ── Negative: incompatible policy (SMT=0 on SMT-active host) ── + # + # Clearing the SMT bit only produces a rejection on a host where SMT is + # actually active, so elsewhere there is nothing to assert. The case is + # left out of the step list rather than run and failed, with an info step + # in its place: a case that vanishes silently is indistinguishable from one + # that passed. Reporting it as "pass" does overload that outcome — a + # first-class per-step "not applicable on this platform" result would say + # so plainly, and is the better home for this once one exists. + if smt_active: + steps_list += [ + Step.for_callable( + name="Set incompatible policy (SMT)", + type="required", + handler="set_incompatible_policy", + timeout=30, + ), + Step.for_vm_launch( + name="Launch with SMT-incompatible policy (expect rejection)", + type="required", + # This one is refused by KVM before the firmware sees it + # (SNP_LAUNCH_START ret=-22 fw_error=0 ''), so there is no + # firmware string to match. Assert the rejection happened at + # launch-start, which still rules out a boot timeout. + expected_result="stdout_contains:SNP_LAUNCH_START", + timeout=300, + ), + Step.for_vm_stop( + name="Stop VM (after SMT policy)", + type="info", + timeout=60, + ), + ] + else: + steps_list.append( + Step.for_callable( + name="SMT policy case not applicable", + type="info", + handler="smt_case_not_applicable", + timeout=10, + ) + ) + + steps_list += [ + # ── Negative: impossible ABI version ── + Step.for_callable( + name="Set impossible ABI version", + type="required", + handler="set_bad_abi_version", + timeout=30, + ), + Step.for_vm_launch( + name="Launch with impossible ABI version (expect rejection)", + type="required", + # Firmware rejects this one: SNP_LAUNCH_START fw_error=7. + expected_result="stdout_contains:Policy is not allowed", + timeout=300, + ), + Step.for_vm_stop( + name="Stop VM (after ABI version)", + type="info", + timeout=60, + ), + ] + + return steps_list diff --git a/sev_verify/cert_tests/c3_0/manifest.toml b/sev_verify/cert_tests/c3_0/manifest.toml index 22b5dee2..97b69076 100644 --- a/sev_verify/cert_tests/c3_0/manifest.toml +++ b/sev_verify/cert_tests/c3_0/manifest.toml @@ -20,3 +20,10 @@ module = "cert_tests.c3_0.c3_0_0_1.snphost_config_commit" scope = "mixed" level = "3.0.0-1" host_changes = true + +[[tests]] +name = "id-block-test" +description = "Verify ID block acceptance, report field binding, and launch rejection" +module = "cert_tests.c3_0.c3_0_0_2.id_block_test" +scope = "mixed" +level = "3.0.0-1" diff --git a/sev_verify/cli.py b/sev_verify/cli.py index b523e814..175684b1 100644 --- a/sev_verify/cli.py +++ b/sev_verify/cli.py @@ -422,6 +422,7 @@ def execute_test( # 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. + # e.g. generate_id_block setting id_block/id_auth/policy. profile = ctx.profile ctx.launch = launch diff --git a/sev_verify/cvm_props.py b/sev_verify/cvm_props.py new file mode 100644 index 00000000..f128cfd0 --- /dev/null +++ b/sev_verify/cvm_props.py @@ -0,0 +1,319 @@ +"""Shared callables for ID block generation in sev_verify test modules. + +A test module that requires an ID block includes these steps in its steps() +list, in order, before vm_launch: + + Step.for_callable(name="Calculate measurement", type="setup", + handler="calculate_measurement", timeout=60), + Step.for_callable(name="Generate ID block", type="setup", + handler="generate_id_block", timeout=30), + +The calculate_measurement step writes guest_measurement.txt to ctx.artifact_dir. +The generate_id_block step reads it, generates ephemeral P-384 key pairs, calls +snpguest to produce id-block.b64 and id-auth.b64, and updates ctx.profile so +that the subsequent vm_launch step passes the ID block to QEMU. + +Both steps follow the additive principle: if OVMF is absent (no measurement +possible), calculate_measurement returns a non-zero exit code and — because it +is typed "setup" — the remaining steps are skipped cleanly. +""" + +from __future__ import annotations + +import os +import string +import subprocess +import tempfile +from dataclasses import dataclass, replace +from pathlib import Path + +from cryptography.hazmat.primitives.asymmetric import ec +from cryptography.hazmat.primitives.serialization import ( + Encoding, + NoEncryption, + PrivateFormat, +) + +from .models import StepContext, StepHandlerResult +# from .vm_profile import VMProfile, VMProfileError +from .vm_profile import VMProfileError + +_MEASUREMENT_FILE = "guest_measurement.txt" +_ID_BLOCK_FILE = "id-block.b64" +_ID_AUTH_FILE = "id-auth.b64" + +DEFAULT_FAMILY_ID = "sev-certify-fam0" +DEFAULT_IMAGE_ID = "sev-certify-img0" +DEFAULT_GUEST_SVN = "48" +DEFAULT_POLICY = "0xb0000" + +# The SNP attestation report MEASUREMENT field is 48 bytes, so the hex form is +# 96 characters. Fixed by the SNP spec, not by configuration. +MEASUREMENT_HEX_LEN = 96 + +# FAMILY_ID and IMAGE_ID are 16-byte fields in both the ID block and the +# attestation report. Fixed by the SNP spec, not by configuration. +ID_FIELD_SIZE = 16 + + +class IdBlockMetadataError(Exception): + """An ID_BLOCK_* environment variable holds a value that cannot be used.""" + + +@dataclass(frozen=True) +class IdBlockMetadata: + """The ID block's identifying fields, validated and in usable form. + + Read once and shared between the step that builds an ID block and the step + that checks the resulting report, so the two cannot disagree about what was + asked for. Deriving expectations separately from the environment would let + the check pass against values the ID block was never built with. + """ + + family_id: str + image_id: str + guest_svn: int + policy: int + + @property + def family_id_bytes(self) -> bytes: + """FAMILY_ID as it appears in the report: ASCII, NUL-padded to 16 bytes.""" + return self.family_id.encode("ascii").ljust(ID_FIELD_SIZE, b"\x00") + + @property + def image_id_bytes(self) -> bytes: + """IMAGE_ID as it appears in the report: ASCII, NUL-padded to 16 bytes.""" + return self.image_id.encode("ascii").ljust(ID_FIELD_SIZE, b"\x00") + + +def _read_id_field(var: str, default: str) -> str: + """Read a 16-byte ID field, rejecting values that cannot encode into one. + + ``ljust`` pads but never truncates, so an over-long value would otherwise + produce an expectation longer than the report field and fail to match every + time, with a byte-diff that does not say why. + """ + value = os.environ.get(var, default) + try: + encoded = value.encode("ascii") + except UnicodeEncodeError as exc: + raise IdBlockMetadataError( + f"{var}: must be ASCII; {value!r} is not ({exc})" + ) from exc + if len(encoded) > ID_FIELD_SIZE: + raise IdBlockMetadataError( + f"{var}: must be at most {ID_FIELD_SIZE} bytes to fit the SNP field; " + f"{value!r} is {len(encoded)}" + ) + return value + + +def _read_int(var: str, default: str, *, base: int) -> int: + """Read an integer-valued variable, failing with the variable's name.""" + raw = os.environ.get(var, default) + try: + parsed = int(raw, base) + except (TypeError, ValueError) as exc: + raise IdBlockMetadataError( + f"{var}: expected an integer, got {raw!r}" + ) from exc + if parsed < 0: + raise IdBlockMetadataError(f"{var}: must not be negative, got {parsed}") + return parsed + + +def read_id_block_metadata() -> IdBlockMetadata: + """Read and validate the ID_BLOCK_* environment variables. + + Raises: + IdBlockMetadataError: a variable is set to something unusable. Raised + rather than allowed to surface as a ValueError or UnicodeEncodeError + from deep in a handler, so the step fails with a message naming the + variable at fault. + """ + return IdBlockMetadata( + family_id=_read_id_field("ID_BLOCK_FAMILY_ID", DEFAULT_FAMILY_ID), + image_id=_read_id_field("ID_BLOCK_IMAGE_ID", DEFAULT_IMAGE_ID), + guest_svn=_read_int("ID_BLOCK_GUEST_SVN", DEFAULT_GUEST_SVN, base=10), + # base=0 so 0x-prefixed, decimal and octal forms are all accepted. + policy=_read_int("ID_BLOCK_POLICY", DEFAULT_POLICY, base=0), + ) + + +class MeasurementError(Exception): + """Base class for problems reading guest_measurement.txt.""" + + +class MeasurementMissing(MeasurementError): + """guest_measurement.txt does not exist.""" + + +class MeasurementMalformed(MeasurementError): + """guest_measurement.txt exists but does not hold a 48-byte hex digest.""" + + +def read_measurement(artifact_dir: Path) -> str: + """Read guest_measurement.txt and return the bare (unprefixed) hex digest. + + Validation lives here, at the point of use, rather than in + calculate_measurement. A check at write time says nothing about what a + later step is about to read: the steps are separated in time, so the file + can change (or be replaced) in between. + + snpguest writes the digest 0x-prefixed under ``--output-format hex``. The + prefix is stripped here so callers operate on a bare body; re-add it with + ``f"0x{...}"`` when handing the value back to snpguest, which decodes an + unprefixed string as base64 rather than hex. + + Raises: + MeasurementMissing: the file is absent. + MeasurementMalformed: the file is present but not a 48-byte hex digest. + """ + measurement_file = artifact_dir / _MEASUREMENT_FILE + try: + raw = measurement_file.read_text().strip() + except FileNotFoundError as exc: + raise MeasurementMissing(f"{_MEASUREMENT_FILE} not found") from exc + + body = raw[2:] if raw[:2].lower() == "0x" else raw + if len(body) != MEASUREMENT_HEX_LEN: + raise MeasurementMalformed( + f"{_MEASUREMENT_FILE}: expected a {MEASUREMENT_HEX_LEN}-character hex " + f"digest (48 bytes), got {len(body)} characters" + ) + if not all(c in string.hexdigits for c in body): + raise MeasurementMalformed( + f"{_MEASUREMENT_FILE}: contains non-hex characters" + ) + return body + + +def calculate_measurement(ctx: StepContext) -> StepHandlerResult: + """Calculate the expected guest launch measurement via snpguest. + + Resolves the OVMF path from ctx.profile, runs snpguest generate + measurement against the guest image, and writes the result to + guest_measurement.txt in ctx.artifact_dir. + """ + try: + ovmf_path = Path(ctx.profile.resolved_ovmf_path()) + except VMProfileError as exc: + return StepHandlerResult(exit_code=1, stderr=str(exc)) + + measurement_file = ctx.artifact_dir / _MEASUREMENT_FILE + result = subprocess.run( + [ + "snpguest", "generate", "measurement", + "--vcpu-type", "EPYC-v4", + "--ovmf", str(ovmf_path), + "--kernel", str(ctx.guest_path), + "--output-format", "hex", + "--measurement-file", str(measurement_file), + ], + capture_output=True, + text=True, + check=False, + ) + if result.returncode != 0: + return StepHandlerResult( + exit_code=result.returncode, + stdout=result.stdout, + stderr=result.stderr, + ) + measurement = measurement_file.read_text().strip() + return StepHandlerResult( + exit_code=0, + stdout=f"Measurement: {measurement}", + ) + + +def generate_id_block(ctx: StepContext) -> StepHandlerResult: + """Generate an ID block and auth block for the current guest measurement. + + Reads guest_measurement.txt from ctx.artifact_dir (written by + calculate_measurement). Generates two ephemeral P-384 key pairs, invokes + snpguest generate id-block, and updates ctx.profile with the resulting + id_block and id_auth values so that vm_launch passes them to QEMU. + + ID block metadata is read from environment variables, falling back to the + DEFAULT_* constants in this module: + ID_BLOCK_FAMILY_ID, ID_BLOCK_IMAGE_ID, ID_BLOCK_GUEST_SVN, ID_BLOCK_POLICY + + If guest_measurement.txt is absent (calculate_measurement was skipped or + failed), this step exits 0 and leaves ctx.profile unchanged, so vm_launch + proceeds without an ID block. + + A file that is present but malformed is a different case and fails the + step: absence is an expected configuration, corruption is not. + """ + try: + meta = read_id_block_metadata() + except IdBlockMetadataError as exc: + return StepHandlerResult(exit_code=1, stderr=str(exc)) + + try: + measurement = read_measurement(ctx.artifact_dir) + except MeasurementMissing as exc: + return StepHandlerResult( + exit_code=0, + stdout=f"INFO: {exc} — skipping ID block generation", + ) + except MeasurementMalformed as exc: + return StepHandlerResult(exit_code=1, stderr=str(exc)) + + id_key = ec.generate_private_key(ec.SECP384R1()) + auth_key = ec.generate_private_key(ec.SECP384R1()) + + id_block_file = ctx.artifact_dir / _ID_BLOCK_FILE + id_auth_file = ctx.artifact_dir / _ID_AUTH_FILE + + with tempfile.TemporaryDirectory() as tmpdir: + id_key_path = Path(tmpdir) / "id-key.pem" + auth_key_path = Path(tmpdir) / "auth-key.pem" + id_key_path.write_bytes( + id_key.private_bytes(Encoding.PEM, PrivateFormat.TraditionalOpenSSL, NoEncryption()) + ) + auth_key_path.write_bytes( + auth_key.private_bytes(Encoding.PEM, PrivateFormat.TraditionalOpenSSL, NoEncryption()) + ) + + result = subprocess.run( + [ + "snpguest", "generate", "id-block", + str(id_key_path), + str(auth_key_path), + f"0x{measurement}", + "--family-id", meta.family_id, + "--image-id", meta.image_id, + "--svn", str(meta.guest_svn), + "--policy", hex(meta.policy), + "--id-file", str(id_block_file), + "--auth-file", str(id_auth_file), + ], + capture_output=True, + text=True, + check=False, + ) + + if result.returncode != 0: + return StepHandlerResult( + exit_code=result.returncode, + stdout=result.stdout, + stderr=result.stderr, + ) + + id_block_b64 = id_block_file.read_text().strip() + id_auth_b64 = id_auth_file.read_text().strip() + + ctx.profile = replace( + ctx.profile, id_block=id_block_b64, id_auth=id_auth_b64, policy=meta.policy + ) + + return StepHandlerResult( + exit_code=0, + stdout=( + f"Generated ID block for measurement {measurement[:16]}...\n" + f" family_id={meta.family_id} image_id={meta.image_id} " + f"svn={meta.guest_svn} policy={hex(meta.policy)}" + ), + ) diff --git a/sev_verify/vm_profile.py b/sev_verify/vm_profile.py index d042efb3..1ebe3c9d 100644 --- a/sev_verify/vm_profile.py +++ b/sev_verify/vm_profile.py @@ -109,6 +109,9 @@ class VMProfile: policy: str | int | None = None auth_key_enabled: bool = False kernel_hashes: bool = True + # ID block parameters — set by generate_id_block() in sev_verify.cvm_props. + id_block: str | None = None + id_auth: str | None = None # Fixed SEV-SNP parameters used by the existing launch scripts. cbitpos: int = 51 reduced_phys_bits: int = 1 @@ -298,6 +301,9 @@ def _build_sev_snp_guest_object(profile: VMProfile) -> str: parts.append(f"policy={_format_policy(profile.policy)}") if profile.auth_key_enabled: parts.append("author-key-enabled=true") + if profile.id_block and profile.id_auth: + parts.append(f"id-block={profile.id_block}") + parts.append(f"id-auth={profile.id_auth}") return ",".join(parts)