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
2 changes: 1 addition & 1 deletion src/cmcp_runtime/__init__.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,3 @@
"""cMCP Runtime: hardware-attested MCP runtime."""

__version__ = "0.1.0"
__version__ = "0.3.0"
2 changes: 1 addition & 1 deletion src/cmcp_runtime/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -67,7 +67,7 @@ class Config:
attestation: AttestationConfig = field(default_factory=AttestationConfig)
agent_manifest: AgentManifestConfig = field(default_factory=AgentManifestConfig)
kill_switch: KillSwitchConfig = field(default_factory=KillSwitchConfig)
policy_bundle_path: str = "policy/"
policy_bundle_path: str = "policies/"
catalog_path: str = "catalog.json"
listen_addr: str = "0.0.0.0:8443"
max_response_size_bytes: int = 2 * 1024 * 1024 # 2MB
Expand Down
6 changes: 6 additions & 0 deletions src/cmcp_runtime/mcp/proxy.py
Original file line number Diff line number Diff line change
Expand Up @@ -375,6 +375,12 @@ def _build_cedar_context(
entry = self._catalog.lookup(tool_name)
ctx: dict[str, Any] = {
"tool_name": tool_name,
# Cedar resource entity: the backend builds Resource::"<resource>" from
# this, so policies can match a tool by name, e.g.
# forbid(principal, action, resource == Resource::"salesforce.contacts");
# Without it the resource defaults to Resource::"default" and no
# resource-scoped policy can ever match.
"resource": tool_name,
"arguments": _cedar_safe(arguments),
"server_identity": entry.server.url if entry else "",
"compliance_domain": entry.compliance_domain if entry else "external",
Expand Down
10 changes: 9 additions & 1 deletion src/cmcp_runtime/mcp/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -528,7 +528,15 @@ async def _session_close(self, request: Request) -> Response:
session_id: str = request.path_params["session_id"]
if session_id != self._session.session_id:
return JSONResponse(
{"error": f"unknown or already closed session_id={session_id}"},
{
"error": "session_not_found",
"message": (
f"No open session with id '{session_id}'. It may already be "
"closed, or you passed the _cmcp.session_id label instead of "
"the internal session id. Look up the internal id via "
"GET /audit/export?session_id=<label>."
),
},
status_code=404,
)

Expand Down
4 changes: 4 additions & 0 deletions src/cmcp_runtime/policy/evaluator.py
Original file line number Diff line number Diff line change
Expand Up @@ -219,6 +219,10 @@ def authorize_egress(
self._maybe_reload()
context: dict[str, Any] = {
"tool_name": tool_name,
# Same resource entity as ingress (see PolicyProxy._build_cedar_context)
# so a resource-scoped permit matches at egress too; without it an
# allowed tool response would be denied on the way back.
"resource": tool_name,
"egress": True,
"sensitivity_level": SENSITIVITY_ORDER.get(session.max_sensitivity, 0),
"injection_events": len(session.injection_events),
Expand Down
2 changes: 1 addition & 1 deletion src/cmcp_runtime/tee/detect.py
Original file line number Diff line number Diff line change
Expand Up @@ -64,7 +64,7 @@ def detect_provider(config: Config) -> TEEProvider:
Detect and return the active TEE provider.

Follows the probe order from docs/spec/attestation.md §1.1:
tpm -> sev-snp -> tdx
azure-cvm -> tpm -> sev-snp -> tdx

If no hardware provider is found:
- CMCP_DEV_MODE=1: returns SoftwareOnlyProvider with a WARN log
Expand Down
120 changes: 120 additions & 0 deletions tests/unit/test_resource_scoped_policy.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,120 @@
"""Resource-scoped Cedar policies must actually enforce.

Regression guard: the Cedar backend builds the request resource from the
`resource` key in the evaluation context. If the proxy/evaluator omit it, the
resource defaults to Resource::"default" and a policy like

forbid(principal, action, resource == Resource::"salesforce.contacts");

never matches, so every call silently default-denies (or, in advisory mode,
forwards). These tests pin the context shape and the end-to-end decision.
"""

from __future__ import annotations

import pytest

from cmcp_runtime.config import AttestationConfig, Config, EnforcementMode
from cmcp_runtime.errors import PolicyDeny
from cmcp_runtime.policy.bundle import PolicyBundle, PolicyManifest
from cmcp_runtime.policy.evaluator import PolicyEvaluator
from cmcp_runtime.session.state import SessionState
from tests.unit.test_call_log_integration import _make_catalog, _make_proxy

RESOURCE_POLICY = """
// Permit calls from the demo-agent workflow.
permit (principal, action, resource) when { context.workflow_id == "demo-agent" };

// Block one tool by resource name. forbid overrides permit.
forbid (principal, action, resource == Resource::"salesforce.contacts");
"""


def _bundle(policy_content: str) -> PolicyBundle:
return PolicyBundle(
manifest=PolicyManifest(
version="1.0.0",
authored_at="2026-06-10T00:00:00Z",
author_identity="test",
commit_sha="abc",
),
policy_files={"demo.cedar": policy_content},
schema_content='{"cMCP": {}}',
bundle_hash="sha256:" + "0" * 64,
)


def _evaluator(mode: EnforcementMode = EnforcementMode.ENFORCING) -> PolicyEvaluator:
cfg = Config(attestation=AttestationConfig(enforcement_mode=mode))
return PolicyEvaluator(_bundle(RESOURCE_POLICY), cfg)


# ── the proxy must put the tool name in the Cedar resource slot ───────────────


def test_build_cedar_context_sets_resource_to_tool_name():
proxy, _, _ = _make_proxy(catalog=_make_catalog("salesforce.contacts"))
ctx = proxy._build_cedar_context("salesforce.contacts", {}, "demo-agent")
assert ctx["resource"] == "salesforce.contacts"


# ── ingress: resource-scoped forbid actually denies ──────────────────────────


def test_resource_forbid_denies_sensitive_tool():
ev = _evaluator()
ctx = {
"tool_name": "salesforce.contacts",
"resource": "salesforce.contacts",
"session_max_sensitivity": "public",
"workflow_id": "demo-agent",
}
with pytest.raises(PolicyDeny):
ev.evaluate(ctx)


def test_permitted_tool_is_allowed():
ev = _evaluator()
decision = ev.evaluate(
{
"tool_name": "echo",
"resource": "echo",
"session_max_sensitivity": "public",
"workflow_id": "demo-agent",
}
)
assert decision.allowed is True
assert decision.would_have_denied is False


def test_wrong_workflow_is_denied():
ev = _evaluator()
with pytest.raises(PolicyDeny):
ev.evaluate(
{
"tool_name": "echo",
"resource": "echo",
"session_max_sensitivity": "public",
"workflow_id": "stranger",
}
)


# ── egress: the allowed tool's response must not be denied on the way back ────


def test_egress_allows_permitted_tool():
ev = _evaluator()
session = SessionState(session_id="sess-egress")
decision = ev.authorize_egress("echo", b"{}", session, workflow_id="demo-agent")
assert decision.allowed is True
assert decision.would_have_denied is False


def test_egress_denies_forbidden_resource():
ev = _evaluator()
session = SessionState(session_id="sess-egress")
with pytest.raises(PolicyDeny):
ev.authorize_egress(
"salesforce.contacts", b"{}", session, workflow_id="demo-agent"
)