diff --git a/src/cmcp_gateway/tee/opaque.py b/src/cmcp_gateway/tee/opaque.py new file mode 100644 index 00000000..a44a7d0b --- /dev/null +++ b/src/cmcp_gateway/tee/opaque.py @@ -0,0 +1,18 @@ +"""Opaque Systems TEE provider stub — not yet implemented.""" + +from __future__ import annotations + +from cmcp_gateway.tee.base import AttestationReport, TEEProvider + + +class OpaqueProvider(TEEProvider): + """Placeholder for the Opaque Systems TEE provider (not yet implemented).""" + + def provider_name(self) -> str: + return "opaque" + + def detect(self) -> bool: + return False + + def get_attestation_report(self, nonce: bytes) -> AttestationReport: + raise NotImplementedError("Opaque provider not yet implemented") diff --git a/src/cmcp_gateway/tee/sev_snp.py b/src/cmcp_gateway/tee/sev_snp.py new file mode 100644 index 00000000..cdbd0fd7 --- /dev/null +++ b/src/cmcp_gateway/tee/sev_snp.py @@ -0,0 +1,102 @@ +"""AMD SEV-SNP TEE provider — implements issue #89.""" + +from __future__ import annotations + +import hashlib +import struct +import sys +from datetime import UTC, datetime +from pathlib import Path + +from cmcp_gateway.tee.base import AttestationReport, TEEProvider + +_SEV_GUEST_DEVICE = Path("/dev/sev-guest") + +# SNP_GET_REPORT ioctl number: 0xC0A01181 +# Derived from: _IOWR(0x11, 0x01, struct snp_report_req) where req is 0xA0 bytes. +_SNP_GET_REPORT = 0xC0A01181 + +# SNP attestation report is 0x4A0 (1184) bytes. +_SNP_REPORT_SIZE = 0x4A0 + +# Request structure: 96-byte user_data + 4-byte vmpl + 28-byte reserved = 128 bytes total +_SNP_REQ_USER_DATA_SIZE = 96 +_SNP_REQ_SIZE = 128 + +# Response structure: 4-byte status + 4-byte report_size + 24-byte reserved + report +_SNP_RESP_HEADER_SIZE = 32 +_SNP_RESP_SIZE = _SNP_RESP_HEADER_SIZE + _SNP_REPORT_SIZE + +# Measurement field offset and size in SNP report +_SNP_MEASUREMENT_OFFSET = 0x60 +_SNP_MEASUREMENT_END = 0x90 # 48 bytes = SHA-384 + + +class SEVSNPProvider(TEEProvider): + """AMD SEV-SNP attestation provider using the /dev/sev-guest ioctl interface.""" + + def provider_name(self) -> str: + return "sev-snp" + + def detect(self) -> bool: + """Return True if /dev/sev-guest exists (Linux only).""" + try: + if sys.platform != "linux": + return False + return _SEV_GUEST_DEVICE.exists() + except Exception: # noqa: BLE001 + return False + + def get_attestation_report(self, nonce: bytes) -> AttestationReport: + """ + Request an SNP attestation report via the SNP_GET_REPORT ioctl. + + The nonce is placed in the 96-byte user_data field (first 64 bytes used, + zero-padded to 96). + """ + try: + import fcntl # available on Linux only + except ImportError as exc: + raise RuntimeError(f"SEV-SNP attestation failed: {exc}") from exc + + # Build request: 96-byte user_data (nonce truncated/padded) + vmpl=0 + reserved + user_data = (nonce[:64] + b"\x00" * 96)[:96] + vmpl = 0 + # struct: 96s user_data, I vmpl, 28s reserved + req = struct.pack("96sI28s", user_data, vmpl, b"\x00" * 28) + + # Response buffer: 4-byte status + 4-byte report_size + 24-byte reserved + report + resp = bytearray(_SNP_RESP_SIZE) + # Place request into response buffer (ioctl arg is a combined req/resp struct) + # The kernel driver takes a pointer to snp_guest_request_ioctl which contains + # pointers; however the simplified /dev/sev-guest interface accepts the request + # directly. We use a single buffer of max(req, resp) size. + buf = bytearray(max(len(req), _SNP_RESP_SIZE)) + buf[: len(req)] = req + + try: + with open(_SEV_GUEST_DEVICE, "rb") as fd: + fcntl.ioctl(fd, _SNP_GET_REPORT, buf) # type: ignore[attr-defined] + except OSError as exc: + raise RuntimeError(f"SEV-SNP attestation failed: {exc}") from exc + + # Extract status (first 4 bytes) + status = struct.unpack_from(" str: + return "tdx" + + def detect(self) -> bool: + """Return True if /dev/tdx_guest exists.""" + try: + if sys.platform != "linux": + return False + return _TDX_GUEST_DEVICE.exists() + except Exception: # noqa: BLE001 + return False + + def get_attestation_report(self, nonce: bytes) -> AttestationReport: + """ + Request a TDREPORT via the TDX_CMD_GET_REPORT0 ioctl. + + The nonce is placed in the REPORTDATA field (first 64 bytes, zero-padded). + """ + try: + import fcntl # available on Linux only + except ImportError as exc: + raise RuntimeError(f"TDX attestation failed: {exc}") from exc + + # Buffer layout: 64-byte REPORTDATA followed by 1024-byte TDREPORT output + buf_size = _REPORTDATA_SIZE + _TDREPORT_SIZE + buf = bytearray(buf_size) + + # Write nonce into REPORTDATA (truncate or pad to 64 bytes) + report_data_bytes = (nonce[:_REPORTDATA_SIZE] + b"\x00" * _REPORTDATA_SIZE)[ + :_REPORTDATA_SIZE + ] + buf[:_REPORTDATA_SIZE] = report_data_bytes + + try: + with open(_TDX_GUEST_DEVICE, "rb") as fd: + fcntl.ioctl(fd, _TDX_CMD_GET_REPORT0, buf) # type: ignore[attr-defined] + except OSError as exc: + raise RuntimeError(f"TDX attestation failed: {exc}") from exc + + # TDREPORT is in the second half of the buffer + raw_evidence = bytes(buf[_REPORTDATA_SIZE : _REPORTDATA_SIZE + _TDREPORT_SIZE]) + + # MRTD field is the TD measurement equivalent + mrtd_bytes = raw_evidence[_MRTD_OFFSET:_MRTD_END] + measurement = "sha384:" + hashlib.sha384(mrtd_bytes).hexdigest() + + return AttestationReport( + provider=self.provider_name(), + measurement=measurement, + report_data=nonce.hex(), + raw_evidence=raw_evidence, + attestation_generated_at=datetime.now(tz=UTC), + attestation_validity_seconds=86400, + ) diff --git a/src/cmcp_gateway/tee/tpm.py b/src/cmcp_gateway/tee/tpm.py new file mode 100644 index 00000000..8d0faad6 --- /dev/null +++ b/src/cmcp_gateway/tee/tpm.py @@ -0,0 +1,192 @@ +"""TPM 2.0 TEE provider — implements issue #83.""" + +from __future__ import annotations + +import hashlib +import subprocess +import sys +from datetime import UTC, datetime +from pathlib import Path +from typing import TYPE_CHECKING + +from cmcp_gateway.tee.base import AttestationReport, TEEProvider + +if TYPE_CHECKING: + pass + +try: + import tpm2_pytss # type: ignore[import-not-found] + + _TSS2_AVAILABLE = True +except ImportError: + tpm2_pytss = None + _TSS2_AVAILABLE = False + +_TPM_DEVICES = [Path("/dev/tpm0"), Path("/dev/tpmrm0")] + + +class TPMProvider(TEEProvider): + """TPM 2.0 attestation provider using tpm2-pytss or subprocess fallback.""" + + def provider_name(self) -> str: + return "tpm" + + def detect(self) -> bool: + """Return True if a TPM device file exists and is readable on Linux.""" + try: + if sys.platform != "linux": + return False + for dev in _TPM_DEVICES: + if dev.exists(): + return True + return False + except Exception: # noqa: BLE001 + return False + + def get_attestation_report(self, nonce: bytes) -> AttestationReport: + """ + Produce a TPM 2.0 PCR-based attestation report. + + Tries tpm2-pytss ESAPI first, then falls back to tpm2_pcrread subprocess. + """ + if _TSS2_AVAILABLE: + return self._report_via_tss2(nonce) + return self._report_via_subprocess(nonce) + + # ── tpm2-pytss path ─────────────────────────────────────────────────────── + + def _report_via_tss2(self, nonce: bytes) -> AttestationReport: + from tpm2_pytss.ESAPI import ESAPI # type: ignore[import-not-found] + from tpm2_pytss.types import ( # type: ignore[import-not-found] + TPM2_ALG, + TPML_PCR_SELECTION, + TPM2B_DATA, + ) + + with ESAPI() as ectx: + # Try SHA-256 first; fall back to SHA-1 + measurement_note: str | None = None + raw_pcrs: list[bytes] = [] + + try: + pcr_sel = TPML_PCR_SELECTION.parse("sha256:0,1,2,3,4,5,6,7") + _, _, digests = ectx.pcr_read(pcr_sel) + for bank in digests.digests: + for digest in bank.digests: + raw_pcrs.append(bytes(digest.buffer)) + except Exception: # noqa: BLE001 + # Fall back to SHA-1 + measurement_note = "sha1-bank-fallback" + pcr_sel = TPML_PCR_SELECTION.parse("sha1:0,1,2,3,4,5,6,7") + _, _, digests = ectx.pcr_read(pcr_sel) + raw_pcrs = [] + for bank in digests.digests: + for digest in bank.digests: + raw_pcrs.append(bytes(digest.buffer)) + + # Ensure we got 8 PCRs + if len(raw_pcrs) < 8: + raise RuntimeError( + f"TPM device found but could not read PCRs: got {len(raw_pcrs)}, expected 8" + ) + + concatenated = b"".join(raw_pcrs[:8]) + measurement = "sha256:" + hashlib.sha256(concatenated).hexdigest() + + # Attempt TPM2_Quote for raw_evidence + raw_evidence: bytes | None = None + try: + qualifying_data = TPM2B_DATA(nonce[:32]) + pcr_sel_quote = TPML_PCR_SELECTION.parse("sha256:0,1,2,3,4,5,6,7") + quoted, _signature = ectx.quote( + object_handle=ectx.get_capability(TPM2_ALG.NULL), + qualifying_data=qualifying_data, + in_scheme=TPM2_ALG.NULL, + pcrselect=pcr_sel_quote, + ) + raw_evidence = bytes(quoted.attestationData) + except Exception: # noqa: BLE001 + raw_evidence = None + + return AttestationReport( + provider=self.provider_name(), + measurement=measurement, + report_data=nonce.hex(), + raw_evidence=raw_evidence, + attestation_generated_at=datetime.now(tz=UTC), + attestation_validity_seconds=3600, + measurement_note=measurement_note, + ) + + # ── subprocess fallback ─────────────────────────────────────────────────── + + def _report_via_subprocess(self, nonce: bytes) -> AttestationReport: + """Read PCRs 0-7 using tpm2_pcrread subprocess.""" + try: + result = subprocess.run( # noqa: S603 + ["tpm2_pcrread", "sha256:0,1,2,3,4,5,6,7"], # noqa: S607 + capture_output=True, + text=True, + check=False, + timeout=30, + ) + except (FileNotFoundError, subprocess.TimeoutExpired) as exc: + raise RuntimeError(f"TPM device found but could not read PCRs: {exc}") from exc + + if result.returncode != 0: + # Try SHA-1 + result = subprocess.run( # noqa: S603 + ["tpm2_pcrread", "sha1:0,1,2,3,4,5,6,7"], # noqa: S607 + capture_output=True, + text=True, + check=False, + timeout=30, + ) + if result.returncode != 0: + raise RuntimeError( + f"TPM device found but could not read PCRs: tpm2_pcrread exited " + f"{result.returncode}: {result.stderr.strip()}" + ) + measurement_note: str | None = "sha1-bank-fallback" + else: + measurement_note = None + + pcr_values = _parse_tpm2_pcrread_output(result.stdout) + if len(pcr_values) < 8: + raise RuntimeError( + f"TPM device found but could not read PCRs: parsed {len(pcr_values)} PCRs" + ) + + concatenated = b"".join(pcr_values[:8]) + measurement = "sha256:" + hashlib.sha256(concatenated).hexdigest() + + return AttestationReport( + provider=self.provider_name(), + measurement=measurement, + report_data=nonce.hex(), + raw_evidence=None, + attestation_generated_at=datetime.now(tz=UTC), + attestation_validity_seconds=3600, + measurement_note=measurement_note, + ) + + +def _parse_tpm2_pcrread_output(output: str) -> list[bytes]: + """ + Parse tpm2_pcrread YAML-ish output into a list of raw PCR bytes. + + Expected format (per PCR): + sha256: + 0 : 0xABCD... + """ + pcr_values: list[bytes] = [] + for line in output.splitlines(): + line = line.strip() + if ":" in line and line.split(":")[0].strip().isdigit(): + _, _, hex_val = line.partition(":") + hex_val = hex_val.strip().lstrip("0x").lstrip("0X") + try: + pcr_values.append(bytes.fromhex(hex_val or "00")) + except ValueError: + continue + return pcr_values diff --git a/tests/unit/test_tee_providers.py b/tests/unit/test_tee_providers.py new file mode 100644 index 00000000..1a009261 --- /dev/null +++ b/tests/unit/test_tee_providers.py @@ -0,0 +1,160 @@ +"""Tests for TPM, SEV-SNP, TDX, and Opaque TEE provider stubs.""" + +from __future__ import annotations + +import subprocess +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +from cmcp_gateway.tee.opaque import OpaqueProvider +from cmcp_gateway.tee.sev_snp import SEVSNPProvider +from cmcp_gateway.tee.tdx import TDXProvider +from cmcp_gateway.tee.tpm import TPMProvider + + +# ── OpaqueProvider ───────────────────────────────────────────────────────────── + +def test_opaque_detect_returns_false() -> None: + assert OpaqueProvider().detect() is False + + +def test_opaque_get_report_raises() -> None: + with pytest.raises(NotImplementedError): + OpaqueProvider().get_attestation_report(b"\x00" * 32) + + +def test_opaque_provider_name() -> None: + assert OpaqueProvider().provider_name() == "opaque" + + +# ── SEVSNPProvider ───────────────────────────────────────────────────────────── + +def test_sev_snp_detect_returns_false_on_non_linux() -> None: + with patch.object(sys, "platform", "win32"), \ + patch.object(Path, "exists", return_value=False): + assert SEVSNPProvider().detect() is False + + +def test_sev_snp_detect_returns_false_when_device_missing() -> None: + with patch.object(sys, "platform", "linux"), \ + patch.object(Path, "exists", return_value=False): + assert SEVSNPProvider().detect() is False + + +def test_sev_snp_detect_returns_true_when_device_present() -> None: + with patch.object(sys, "platform", "linux"), \ + patch.object(Path, "exists", return_value=True): + assert SEVSNPProvider().detect() is True + + +def test_sev_snp_provider_name() -> None: + assert SEVSNPProvider().provider_name() == "sev-snp" + + +def test_sev_snp_get_report_raises_on_ioctl_failure(monkeypatch: pytest.MonkeyPatch) -> None: + """ioctl raising OSError must surface as RuntimeError.""" + mock_fcntl = MagicMock() + mock_fcntl.ioctl = MagicMock(side_effect=OSError("ioctl failed")) + monkeypatch.setitem(sys.modules, "fcntl", mock_fcntl) + + mock_fd = MagicMock() + mock_fd.__enter__ = MagicMock(return_value=mock_fd) + mock_fd.__exit__ = MagicMock(return_value=False) + + with patch("builtins.open", return_value=mock_fd): + with pytest.raises(RuntimeError, match="SEV-SNP attestation failed"): + SEVSNPProvider().get_attestation_report(b"\x00" * 32) + + +def test_sev_snp_get_report_raises_when_fcntl_unavailable( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """When fcntl is absent (non-Linux), get_attestation_report must raise RuntimeError.""" + # Remove fcntl from sys.modules so the import inside the method fails + monkeypatch.setitem(sys.modules, "fcntl", None) # type: ignore[arg-type] + with pytest.raises(RuntimeError, match="SEV-SNP attestation failed"): + SEVSNPProvider().get_attestation_report(b"\x00" * 32) + + +# ── TDXProvider ──────────────────────────────────────────────────────────────── + +def test_tdx_detect_returns_false_when_device_missing() -> None: + with patch.object(Path, "exists", return_value=False): + assert TDXProvider().detect() is False + + +def test_tdx_detect_returns_false_on_non_linux() -> None: + with patch.object(sys, "platform", "darwin"), \ + patch.object(Path, "exists", return_value=False): + assert TDXProvider().detect() is False + + +def test_tdx_detect_returns_true_when_device_present() -> None: + with patch.object(sys, "platform", "linux"), \ + patch.object(Path, "exists", return_value=True): + assert TDXProvider().detect() is True + + +def test_tdx_provider_name() -> None: + assert TDXProvider().provider_name() == "tdx" + + +def test_tdx_get_report_raises_on_ioctl_failure(monkeypatch: pytest.MonkeyPatch) -> None: + """ioctl raising OSError must surface as RuntimeError.""" + mock_fcntl = MagicMock() + mock_fcntl.ioctl = MagicMock(side_effect=OSError("ioctl failed")) + monkeypatch.setitem(sys.modules, "fcntl", mock_fcntl) + + mock_fd = MagicMock() + mock_fd.__enter__ = MagicMock(return_value=mock_fd) + mock_fd.__exit__ = MagicMock(return_value=False) + + with patch("builtins.open", return_value=mock_fd): + with pytest.raises(RuntimeError, match="TDX attestation failed"): + TDXProvider().get_attestation_report(b"\x00" * 32) + + +# ── TPMProvider ──────────────────────────────────────────────────────────────── + +def test_tpm_detect_returns_false_when_no_device() -> None: + with patch.object(sys, "platform", "linux"), \ + patch.object(Path, "exists", return_value=False): + assert TPMProvider().detect() is False + + +def test_tpm_detect_returns_false_on_non_linux() -> None: + with patch.object(sys, "platform", "win32"): + assert TPMProvider().detect() is False + + +def test_tpm_provider_name() -> None: + assert TPMProvider().provider_name() == "tpm" + + +def test_tpm_get_report_raises_when_no_tss2(monkeypatch: pytest.MonkeyPatch) -> None: + """ + When tpm2_pytss is not importable and subprocess tpm2_pcrread returns non-zero, + get_attestation_report must raise RuntimeError. + """ + # Force the module-level flag to False (no tss2) + monkeypatch.setattr("cmcp_gateway.tee.tpm._TSS2_AVAILABLE", False) + + # Make subprocess.run return a non-zero exit code for both sha256 and sha1 probes + failed_result = MagicMock(spec=subprocess.CompletedProcess) + failed_result.returncode = 1 + failed_result.stderr = "error: cannot open /dev/tpm0" + failed_result.stdout = "" + + monkeypatch.setattr(subprocess, "run", MagicMock(return_value=failed_result)) + + with pytest.raises(RuntimeError, match="TPM device found but could not read PCRs"): + TPMProvider().get_attestation_report(b"\x00" * 32) + + +def test_tpm_detect_does_not_raise_on_exception() -> None: + """detect() must swallow all exceptions and return False.""" + with patch.object(Path, "exists", side_effect=PermissionError("no access")): + assert TPMProvider().detect() is False