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
18 changes: 18 additions & 0 deletions src/cmcp_gateway/tee/opaque.py
Original file line number Diff line number Diff line change
@@ -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")
102 changes: 102 additions & 0 deletions src/cmcp_gateway/tee/sev_snp.py
Original file line number Diff line number Diff line change
@@ -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("<I", buf, 0)[0]
if status != 0:
raise RuntimeError(f"SEV-SNP attestation failed: ioctl status={status:#x}")

# Report starts at offset _SNP_RESP_HEADER_SIZE
raw_evidence = bytes(buf[_SNP_RESP_HEADER_SIZE : _SNP_RESP_HEADER_SIZE + _SNP_REPORT_SIZE])

# Measurement = SHA-384 of the measurement field within the SNP report
measurement_bytes = raw_evidence[_SNP_MEASUREMENT_OFFSET:_SNP_MEASUREMENT_END]
measurement = "sha384:" + hashlib.sha384(measurement_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,
)
85 changes: 85 additions & 0 deletions src/cmcp_gateway/tee/tdx.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,85 @@
"""Intel TDX TEE provider — implements issue #93."""

from __future__ import annotations

import hashlib
import sys
from datetime import UTC, datetime
from pathlib import Path

from cmcp_gateway.tee.base import AttestationReport, TEEProvider

_TDX_GUEST_DEVICE = Path("/dev/tdx_guest")

# TDX_CMD_GET_REPORT0 ioctl: 0xC0884000
# Derived from: _IOWR(0x40, 0x00, struct tdx_report_req) where req is 0x88 bytes.
_TDX_CMD_GET_REPORT0 = 0xC0884000

# TDREPORT size: 1024 bytes
_TDREPORT_SIZE = 1024

# REPORTDATA input: 64 bytes (placed at start of ioctl buffer)
_REPORTDATA_SIZE = 64

# MRTD field in TDREPORT: bytes 0x90..0xC0 (48 bytes)
_MRTD_OFFSET = 0x90
_MRTD_END = 0xC0


class TDXProvider(TEEProvider):
"""Intel TDX attestation provider using the /dev/tdx_guest ioctl interface."""

def provider_name(self) -> 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,
)
Loading
Loading