diff --git a/src/cmcp_gateway/startup.py b/src/cmcp_gateway/startup.py index 5f66519b..ed1bdbad 100644 --- a/src/cmcp_gateway/startup.py +++ b/src/cmcp_gateway/startup.py @@ -21,6 +21,7 @@ from cmcp_gateway.policy.bundle import PolicyStore, load_policy_bundle from cmcp_gateway.tee.base import AttestationReport, TEEProvider from cmcp_gateway.tee.detect import detect_provider +from cmcp_gateway.tee.nras import AppraisalResult, try_appraise from cmcp_gateway.tee.spiffe import SpiffeClientResult, fetch_svid logger = logging.getLogger(__name__) @@ -48,6 +49,7 @@ class GatewayContext: policy_bundle: PolicyStore catalog: ToolCatalog spiffe: SpiffeClientResult | None = None + nras_appraisal: AppraisalResult | None = None def _fatal(code: str, message: str, **fields: Any) -> None: @@ -245,6 +247,10 @@ def run_startup(config_path: str) -> GatewayContext: spiffe_result.failure_reason, ) + # Step 5c: NRAS post-attestation appraisal (non-fatal, Phase 2 / v0.2 -- issue #125). + # CMCP_NRAS_API_KEY missing -> skip with warning; any NRAS error -> skip with warning. + nras_appraisal = try_appraise(attestation_report) + return GatewayContext( config=config, tee_provider=tee_provider, @@ -253,4 +259,5 @@ def run_startup(config_path: str) -> GatewayContext: policy_bundle=policy_store, catalog=catalog, spiffe=spiffe_result, + nras_appraisal=nras_appraisal, ) diff --git a/src/cmcp_gateway/tee/nras.py b/src/cmcp_gateway/tee/nras.py new file mode 100644 index 00000000..21b37200 --- /dev/null +++ b/src/cmcp_gateway/tee/nras.py @@ -0,0 +1,215 @@ +"""NVIDIA Remote Attestation Service (NRAS) client -- Phase 2 / v0.2. + +Implements issue #125: post-attestation appraisal via the NRAS REST API. +After get_attestation_report() completes, callers may optionally submit +the raw evidence to NRAS for hardware-level appraisal. The result is stored +in GatewayContext.nras_appraisal and written into the TRACE Trust Record +appraisal field. + +Integration is opt-in: if CMCP_NRAS_API_KEY is absent the step is skipped +with a WARNING log and GatewayContext.nras_appraisal is None. + +Phase 2 / v0.2 -- target: Q3 2026 (Berlin demo). +""" + +from __future__ import annotations + +import base64 +import logging +import os +from dataclasses import dataclass +from datetime import UTC, datetime +from typing import Any + +import httpx + +from cmcp_gateway.tee.base import AttestationReport + +logger = logging.getLogger(__name__) + +NRAS_ENDPOINT: str = "https://nras.nvidia.com/v1/attestation/gpu" +_DEFAULT_TIMEOUT_SECONDS: float = 10.0 +_ENV_API_KEY: str = "CMCP_NRAS_API_KEY" +_VALID_STATUSES: frozenset[str] = frozenset({"affirming", "warning", "contraindicated"}) + + +@dataclass +class AppraisalResult: + """EAR (Evidence Appraisal Result) returned by NRAS. + + Attributes: + status: EAR appraisal status -- one of affirming, warning, contraindicated. + verifier: NRAS verifier identifier string from verifier-identifier. + timestamp: ISO 8601 UTC timestamp of when the appraisal was received. + ear_raw: Full EAR JSON payload, preserved verbatim for audit. + """ + + status: str + verifier: str + timestamp: str + ear_raw: dict[str, Any] + + +class NRASError(Exception): + """Base class for all NRAS client errors.""" + + +class NRASAuthError(NRASError): + """NRAS returned HTTP 401 -- API key invalid or missing.""" + + +class NRASAppraisalError(NRASError): + """NRAS rejected the attestation evidence (4xx other than 401).""" + + def __init__(self, status_code: int, body: str) -> None: + self.status_code = status_code + super().__init__(f"NRAS appraisal failed: HTTP {status_code} -- {body[:200]}") + + +class NRASClient: + """Client for the NVIDIA Remote Attestation Service. + + Phase 2 / v0.2 -- implements issue #125. + + Args: + api_key: NVIDIA developer API key (CMCP_NRAS_API_KEY). + endpoint: NRAS appraisal endpoint URL. Override for testing only. + timeout: HTTP request timeout in seconds. + http_client: Optional pre-built httpx.Client; injected in tests. + """ + + def __init__( + self, + api_key: str, + *, + endpoint: str = NRAS_ENDPOINT, + timeout: float = _DEFAULT_TIMEOUT_SECONDS, + http_client: httpx.Client | None = None, + ) -> None: + self._api_key = api_key + self._endpoint = endpoint + self._timeout = timeout + self._http_client = http_client + + def appraise(self, report: AttestationReport) -> AppraisalResult: + """Submit attestation evidence to NRAS and return the EAR result. + + The nonce field is taken from report.report_data (hex-encoded SHA-256 nonce). + attestation_report is the base64-encoded raw_evidence blob; if raw_evidence + is None (software-only dev mode) the measurement bytes are encoded instead. + + Raises: + NRASAuthError: HTTP 401 from NRAS. + NRASAppraisalError: Any other 4xx response. + NRASError: Network / timeout errors or unexpected response. + """ + nonce_b64 = base64.b64encode(bytes.fromhex(report.report_data)).decode() + + if report.raw_evidence is not None: + evidence_b64 = base64.b64encode(report.raw_evidence).decode() + else: + evidence_b64 = base64.b64encode(report.measurement.encode()).decode() + + payload: dict[str, Any] = { + "nonce": nonce_b64, + "attestation_report": evidence_b64, + } + headers = { + "Authorization": f"Bearer {self._api_key}", + "Content-Type": "application/json", + "Accept": "application/json", + } + + try: + ear = self._post(payload, headers) + except (NRASAuthError, NRASAppraisalError, NRASError): + raise + except httpx.TimeoutException as exc: + raise NRASError(f"NRAS request timed out after {self._timeout}s") from exc + except httpx.HTTPError as exc: + raise NRASError(f"NRAS HTTP transport error: {exc}") from exc + + return self._parse_ear(ear) + + def _post(self, payload: dict[str, Any], headers: dict[str, str]) -> dict[str, Any]: + """POST to the NRAS endpoint and return the parsed JSON body.""" + client = self._http_client + if client is not None: + response = client.post(self._endpoint, json=payload, headers=headers) + else: + with httpx.Client(timeout=self._timeout) as c: + response = c.post(self._endpoint, json=payload, headers=headers) + + if response.status_code == 401: + raise NRASAuthError("NRAS rejected the API key (HTTP 401)") + if response.status_code >= 400: + raise NRASAppraisalError(response.status_code, response.text) + + try: + return response.json() + except Exception as exc: + raise NRASError(f"NRAS response is not valid JSON: {exc}") from exc + @staticmethod + def _parse_ear(ear: dict[str, Any]) -> AppraisalResult: + """Validate and extract fields from a raw EAR JSON payload.""" + status = ear.get("status", "") + if status not in _VALID_STATUSES: + raise NRASError( + f"NRAS returned unrecognised EAR status {status!r}. Expected one of {sorted(_VALID_STATUSES)}." + ) + + verifier = ear.get("verifier-identifier", "") + if not isinstance(verifier, str) or not verifier: + verifier = "nras.nvidia.com" + + timestamp = datetime.now(tz=UTC).isoformat() + + return AppraisalResult( + status=status, + verifier=verifier, + timestamp=timestamp, + ear_raw=ear, + ) + + +def try_appraise(report: AttestationReport) -> AppraisalResult | None: + """Attempt NRAS appraisal using the CMCP_NRAS_API_KEY env var. + + Returns None (and logs a WARNING) when: + - CMCP_NRAS_API_KEY is not set. + - The NRAS call fails for any reason. + + This is the integration point called from startup.run_startup() after + get_attestation_report() succeeds. It must never raise -- a missing or + failed appraisal is non-fatal per issue #125. + + Phase 2 / v0.2 -- implements issue #125. + """ + api_key = os.environ.get(_ENV_API_KEY) + if not api_key: + logger.warning( + "CMCP_NRAS_API_KEY is not set -- skipping NRAS post-attestation appraisal. " + "The TRACE Trust Record appraisal field will be empty. " + "Set CMCP_NRAS_API_KEY to enable hardware appraisal (Phase 2 / v0.2)." + ) + return None + + client = NRASClient(api_key=api_key) + try: + result = client.appraise(report) + logger.info( + "NRAS appraisal complete: status=%s verifier=%s", + result.status, + result.verifier, + ) + return result + except NRASAuthError: + logger.warning( + "NRAS appraisal skipped: API key was rejected (HTTP 401). " + "Check CMCP_NRAS_API_KEY." + ) + except NRASAppraisalError as exc: + logger.warning("NRAS appraisal rejected evidence: %s", exc) + except NRASError as exc: + logger.warning("NRAS appraisal failed (network/timeout): %s", exc) + return None diff --git a/tests/unit/test_nras_client.py b/tests/unit/test_nras_client.py new file mode 100644 index 00000000..c83fec8b --- /dev/null +++ b/tests/unit/test_nras_client.py @@ -0,0 +1,307 @@ +from __future__ import annotations + +import json +from datetime import UTC, datetime +from unittest.mock import patch + +import httpx +import pytest + +from cmcp_gateway.tee.base import AttestationReport +from cmcp_gateway.tee.nras import ( + NRAS_ENDPOINT, + AppraisalResult, + NRASAppraisalError, + NRASAuthError, + NRASClient, + NRASError, + try_appraise, +) + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture() +def sw_report() -> AttestationReport: + return AttestationReport( + provider="software-only", + measurement="DEVELOPMENT_ONLY_NOT_FOR_PRODUCTION", + report_data="aa" * 32, + raw_evidence=None, + attestation_generated_at=datetime.now(tz=UTC), + attestation_validity_seconds=86400, + ) + + +@pytest.fixture() +def hw_report() -> AttestationReport: + return AttestationReport( + provider="software-only", + measurement="sha256:" + "ab" * 32, + report_data="cd" * 32, + raw_evidence=bytes.fromhex("deadbeef") * 64, + attestation_generated_at=datetime.now(tz=UTC), + attestation_validity_seconds=86400, + ) + + +def _make_ear(status: str = "affirming") -> dict: + return { + "eat_profile": "tag:nvidia.com,2024:nras-v1", + "status": status, + "verifier-identifier": "nras.nvidia.com", + } + + +def _mock_client(status_code: int, body: object) -> httpx.Client: + content = json.dumps(body).encode() + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(status_code, content=content) + + return httpx.Client(transport=httpx.MockTransport(handler)) + + +def _mock_text_client(status_code: int, text: str) -> httpx.Client: + content = text.encode() + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(status_code, content=content) + + return httpx.Client(transport=httpx.MockTransport(handler)) + + +# --------------------------------------------------------------------------- +# NRASClient.appraise -- happy paths +# --------------------------------------------------------------------------- + + +def test_appraise_affirming_status(sw_report): + ear = _make_ear("affirming") + client = NRASClient(api_key="test-key", http_client=_mock_client(200, ear)) + result = client.appraise(sw_report) + assert isinstance(result, AppraisalResult) + assert result.status == "affirming" + assert result.verifier == "nras.nvidia.com" + assert result.ear_raw == ear + assert result.timestamp + + +def test_appraise_warning_status(sw_report): + ear = _make_ear("warning") + client = NRASClient(api_key="test-key", http_client=_mock_client(200, ear)) + result = client.appraise(sw_report) + assert result.status == "warning" + + +def test_appraise_contraindicated_status(sw_report): + ear = _make_ear("contraindicated") + client = NRASClient(api_key="test-key", http_client=_mock_client(200, ear)) + result = client.appraise(sw_report) + assert result.status == "contraindicated" + + +def test_appraise_uses_raw_evidence_when_present(hw_report): + import base64 + + captured: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + captured.append(request) + return httpx.Response(200, content=json.dumps(_make_ear()).encode()) + + http = httpx.Client(transport=httpx.MockTransport(handler)) + client = NRASClient(api_key="test-key", http_client=http) + client.appraise(hw_report) + + body = json.loads(captured[0].content) + expected = base64.b64encode(hw_report.raw_evidence).decode() + assert body["attestation_report"] == expected + + +def test_appraise_uses_measurement_when_no_raw_evidence(sw_report): + import base64 + + captured: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + captured.append(request) + return httpx.Response(200, content=json.dumps(_make_ear()).encode()) + + http = httpx.Client(transport=httpx.MockTransport(handler)) + client = NRASClient(api_key="k", http_client=http) + client.appraise(sw_report) + + body = json.loads(captured[0].content) + expected = base64.b64encode(sw_report.measurement.encode()).decode() + assert body["attestation_report"] == expected + + +def test_appraise_sends_bearer_auth(sw_report): + captured: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + captured.append(request) + return httpx.Response(200, content=json.dumps(_make_ear()).encode()) + + http = httpx.Client(transport=httpx.MockTransport(handler)) + client = NRASClient(api_key="my-secret-key", http_client=http) + client.appraise(sw_report) + assert captured[0].headers["authorization"] == "Bearer my-secret-key" + + +def test_appraise_posts_to_nras_endpoint(sw_report): + captured: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + captured.append(request) + return httpx.Response(200, content=json.dumps(_make_ear()).encode()) + + http = httpx.Client(transport=httpx.MockTransport(handler)) + client = NRASClient(api_key="k", http_client=http) + client.appraise(sw_report) + assert str(captured[0].url) == NRAS_ENDPOINT + assert captured[0].method == "POST" + + +def test_verifier_defaults_when_missing_from_ear(sw_report): + ear = {"status": "affirming"} + client = NRASClient(api_key="k", http_client=_mock_client(200, ear)) + result = client.appraise(sw_report) + assert result.verifier == "nras.nvidia.com" + + +# --------------------------------------------------------------------------- +# NRASClient.appraise -- error paths +# --------------------------------------------------------------------------- + + +def test_appraise_raises_auth_error_on_401(sw_report): + client = NRASClient(api_key="bad", http_client=_mock_text_client(401, "Unauthorized")) + with pytest.raises(NRASAuthError): + client.appraise(sw_report) + + +def test_appraise_raises_appraisal_error_on_422(sw_report): + client = NRASClient(api_key="k", http_client=_mock_text_client(422, "invalid evidence")) + with pytest.raises(NRASAppraisalError) as exc_info: + client.appraise(sw_report) + assert exc_info.value.status_code == 422 + + +def test_appraise_raises_appraisal_error_on_400(sw_report): + client = NRASClient(api_key="k", http_client=_mock_text_client(400, "bad request")) + with pytest.raises(NRASAppraisalError) as exc_info: + client.appraise(sw_report) + assert exc_info.value.status_code == 400 + + +def test_appraise_raises_nras_error_on_unknown_ear_status(sw_report): + ear = {"status": "unknown-future-status", "verifier-identifier": "nras.nvidia.com"} + client = NRASClient(api_key="k", http_client=_mock_client(200, ear)) + with pytest.raises(NRASError, match="unrecognised EAR status"): + client.appraise(sw_report) + + +def test_appraise_raises_nras_error_on_timeout(sw_report): + def timeout_handler(request: httpx.Request) -> httpx.Response: + raise httpx.TimeoutException("timed out", request=request) + + http = httpx.Client(transport=httpx.MockTransport(timeout_handler)) + client = NRASClient(api_key="k", http_client=http) + with pytest.raises(NRASError, match="timed out"): + client.appraise(sw_report) + + +def test_appraise_raises_nras_error_on_non_json_response(sw_report): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, content=b"not json at all") + + http = httpx.Client(transport=httpx.MockTransport(handler)) + client = NRASClient(api_key="k", http_client=http) + with pytest.raises(NRASError, match="not valid JSON"): + client.appraise(sw_report) + + +# --------------------------------------------------------------------------- +# try_appraise -- non-fatal wrapper +# --------------------------------------------------------------------------- + + +def test_try_appraise_returns_none_when_no_api_key(sw_report, monkeypatch): + monkeypatch.delenv("CMCP_NRAS_API_KEY", raising=False) + result = try_appraise(sw_report) + assert result is None + + +def test_try_appraise_returns_result_on_success(sw_report, monkeypatch): + monkeypatch.setenv("CMCP_NRAS_API_KEY", "valid-key") + ear = _make_ear("affirming") + mock_instance = NRASClient(api_key="valid-key", http_client=_mock_client(200, ear)) + + with patch("cmcp_gateway.tee.nras.NRASClient", return_value=mock_instance): + result = try_appraise(sw_report) + + assert result is not None + assert result.status == "affirming" + + +def test_try_appraise_returns_none_on_auth_failure(sw_report, monkeypatch): + monkeypatch.setenv("CMCP_NRAS_API_KEY", "bad-key") + mock_instance = NRASClient( + api_key="bad-key", + http_client=_mock_text_client(401, "Unauthorized"), + ) + + with patch("cmcp_gateway.tee.nras.NRASClient", return_value=mock_instance): + result = try_appraise(sw_report) + + assert result is None + + +def test_try_appraise_returns_none_on_appraisal_rejection(sw_report, monkeypatch): + monkeypatch.setenv("CMCP_NRAS_API_KEY", "k") + mock_instance = NRASClient( + api_key="k", + http_client=_mock_text_client(422, "bad evidence"), + ) + + with patch("cmcp_gateway.tee.nras.NRASClient", return_value=mock_instance): + result = try_appraise(sw_report) + + assert result is None + + +def test_try_appraise_returns_none_on_timeout(sw_report, monkeypatch): + monkeypatch.setenv("CMCP_NRAS_API_KEY", "k") + + def timeout_handler(request: httpx.Request) -> httpx.Response: + raise httpx.TimeoutException("timed out", request=request) + + http = httpx.Client(transport=httpx.MockTransport(timeout_handler)) + mock_instance = NRASClient(api_key="k", http_client=http) + + with patch("cmcp_gateway.tee.nras.NRASClient", return_value=mock_instance): + result = try_appraise(sw_report) + + assert result is None + + +def test_try_appraise_logs_warning_when_no_key(sw_report, monkeypatch, caplog): + import logging + monkeypatch.delenv("CMCP_NRAS_API_KEY", raising=False) + with caplog.at_level(logging.WARNING, logger="cmcp_gateway.tee.nras"): + try_appraise(sw_report) + assert any("CMCP_NRAS_API_KEY" in r.message for r in caplog.records) + + +# --------------------------------------------------------------------------- +# GatewayContext: nras_appraisal field +# --------------------------------------------------------------------------- + + +def test_gateway_context_nras_appraisal_defaults_none(): + from cmcp_gateway.startup import GatewayContext + assert GatewayContext.__dataclass_fields__["nras_appraisal"].default is None