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
3 changes: 3 additions & 0 deletions src/cmcp_gateway/mcp/proxy.py
Original file line number Diff line number Diff line change
Expand Up @@ -71,6 +71,7 @@ def __init__(
attestation_generated_at: datetime | None = None,
attestation_validity_seconds: int = 86400,
catalog_hash: str | None = None,
attestation_platform: str = "unknown",
) -> None:
self._catalog = catalog
self._policy = policy_evaluator
Expand All @@ -82,6 +83,7 @@ def __init__(
self._attestation_generated_at = attestation_generated_at
self._attestation_validity_seconds = attestation_validity_seconds
self._catalog_hash = catalog_hash or catalog.catalog_hash
self._attestation_platform = attestation_platform

# Build AGT GovernancePolicy from cMCP catalog
allowed_tools = list(catalog.entries.keys())
Expand Down Expand Up @@ -158,6 +160,7 @@ def _build_cedar_context(
"baa_covered": (not entry.requires_baa) if entry else False,
"destination_class": "external",
"session_max_sensitivity": self._session.max_sensitivity,
"attestation_platform": self._attestation_platform,
}
if workflow_id is not None:
ctx["workflow_id"] = workflow_id
Expand Down
40 changes: 40 additions & 0 deletions tests/unit/test_mcp_proxy.py
Original file line number Diff line number Diff line change
Expand Up @@ -207,6 +207,46 @@ async def test_cedar_context_includes_arguments():
assert ctx["arguments"] == args


# ── POLICY-005 (issue #162): attestation_platform in Cedar context ────────────

@pytest.mark.asyncio
async def test_cedar_context_includes_attestation_platform():
"""POLICY-005 (issue #162) — attestation_platform must be in Cedar context so
policies can restrict calls to hardware-attested callers only."""
from cmcp_gateway.mcp.proxy import CMCPProxy

evaluator = _make_evaluator()
cfg = Config()
cfg.attestation = AttestationConfig(enforcement_mode=EnforcementMode.ENFORCING)
session = SessionState(session_id="sess-001")
chain = AuditChain("sess-001")

with patch("cmcp_gateway.mcp.proxy.MCPGateway"), \
patch("cmcp_gateway.mcp.proxy.MCPResponseScanner"):
proxy = CMCPProxy(
_make_catalog(), evaluator, session, chain, cfg,
attestation_platform="amd-sev-snp",
)
proxy._mcp_gateway = MagicMock()
proxy._mcp_gateway.call_tool = AsyncMock(return_value=MagicMock(
sensitivity_tags=[], injection_detected=False
))

await proxy.call_tool("c1", "test.tool", {})
ctx = evaluator.evaluate.call_args[0][0]
assert ctx["attestation_platform"] == "amd-sev-snp"


@pytest.mark.asyncio
async def test_cedar_context_attestation_platform_default_is_unknown():
"""POLICY-005 — default attestation_platform is 'unknown' when not specified."""
evaluator = _make_evaluator()
proxy, _, _ = _make_proxy(evaluator=evaluator)
await proxy.call_tool("c1", "test.tool", {})
ctx = evaluator.evaluate.call_args[0][0]
assert ctx["attestation_platform"] == "unknown"


# ── POLICY-005: request_payload_hash in all audit entries ────────────────────

@pytest.mark.asyncio
Expand Down
Loading