diff --git a/src/cmcp_gateway/audit/chain.py b/src/cmcp_gateway/audit/chain.py index 10df0ded..0e8664af 100644 --- a/src/cmcp_gateway/audit/chain.py +++ b/src/cmcp_gateway/audit/chain.py @@ -22,6 +22,7 @@ "suspicious_call_sequence", "attestation_stale", "catalog_drift", + "break_glass_used", ] PolicyDecision = Literal["allow", "deny", "redact", "advisory_deny", "fault", "n/a"] diff --git a/src/cmcp_gateway/catalog/loader.py b/src/cmcp_gateway/catalog/loader.py index c11391cd..578af8b0 100644 --- a/src/cmcp_gateway/catalog/loader.py +++ b/src/cmcp_gateway/catalog/loader.py @@ -5,6 +5,7 @@ import hashlib import json from dataclasses import dataclass, field +from datetime import UTC, datetime from pathlib import Path from typing import Any, Literal @@ -52,10 +53,21 @@ class CatalogEntry: schema_validation_mode: Literal["redact", "strict", "log"] = field(default="redact") +@dataclass +class CatalogException: + """Metadata for a runtime break-glass exception entry.""" + + tool_name: str + reason: str + authorized_by: str + added_at: str # ISO 8601 UTC + + @dataclass class ToolCatalog: entries: dict[str, CatalogEntry] # tool_name -> entry - catalog_hash: str # sha256: measured into the TEE report + catalog_hash: str # sha256: measured into the TEE report; never mutated + exceptions: list[CatalogException] = field(default_factory=list) def lookup(self, tool_name: str) -> CatalogEntry | None: return self.entries.get(tool_name) @@ -66,6 +78,28 @@ def require(self, tool_name: str) -> CatalogEntry: raise ToolNotInCatalog(f"Tool '{tool_name}' not in attested catalog") return entry + def add_exception( + self, + entry: CatalogEntry, + reason: str, + authorized_by: str, + ) -> None: + """Add a runtime catalog exception without modifying catalog_hash. + + The sealed catalog_hash reflects the original measured catalog only. + Exception entries are visible in the TRACE Claim under gateway.catalog_exceptions. + """ + entry.catalog_exception = True + self.entries[entry.tool_name] = entry + self.exceptions.append( + CatalogException( + tool_name=entry.tool_name, + reason=reason, + authorized_by=authorized_by, + added_at=datetime.now(UTC).isoformat(), + ) + ) + def _sha256_hex(data: bytes) -> str: return hashlib.sha256(data).hexdigest() diff --git a/src/cmcp_gateway/mcp/proxy.py b/src/cmcp_gateway/mcp/proxy.py index 291d362a..29616dec 100644 --- a/src/cmcp_gateway/mcp/proxy.py +++ b/src/cmcp_gateway/mcp/proxy.py @@ -281,6 +281,25 @@ async def call_tool( audit_entry_hash=self._audit.chain_tip, ) + # Step 1b: break-glass warning — log and audit every call via an exception entry + if entry.catalog_exception: + logger.warning( + "BREAK_GLASS_ACTIVE: tool=%s call_id=%s server=%s", + tool_name, + call_id, + entry.server.url, + ) + self._audit.append( + "break_glass_used", + call_id=call_id, + tool_name=tool_name, + server_identity=entry.server.url, + policy_decision="allow", + session_sensitivity_before=sensitivity_before, + session_sensitivity_after=self._session.max_sensitivity, + workflow_id=workflow_id, + ) + # Step 2: Cedar policy evaluation cedar_context = self._build_cedar_context(tool_name, arguments, workflow_id) policy_rule: str | None = None diff --git a/src/cmcp_gateway/mcp/server.py b/src/cmcp_gateway/mcp/server.py index fc831b40..e8fa9887 100644 --- a/src/cmcp_gateway/mcp/server.py +++ b/src/cmcp_gateway/mcp/server.py @@ -24,6 +24,7 @@ from starlette.responses import JSONResponse, Response from starlette.routing import Route +from cmcp_gateway.catalog.loader import ApprovedDefinition, CatalogEntry, ServerIdentity from cmcp_gateway.mcp.proxy import CMCPProxy if TYPE_CHECKING: @@ -128,6 +129,7 @@ def __init__( self._session_reset, methods=["POST"], ), + Route("/catalog/exception", self._catalog_exception, methods=["POST"]), ], middleware=middleware, exception_handlers={Exception: _unhandled_error_handler}, @@ -362,6 +364,110 @@ async def _audit_export(self, request: Request) -> Response: ) return JSONResponse(bundle) + async def _catalog_exception(self, request: Request) -> Response: + """POST /catalog/exception — add a break-glass catalog exception at runtime. + + The exception is visible in the TRACE Claim but does NOT modify catalog_hash. + Requires the same bearer token as all other operator endpoints. + """ + try: + body = await request.body() + data = json.loads(body) + except (json.JSONDecodeError, UnicodeDecodeError): + return JSONResponse( + {"error": "invalid JSON body", "error_code": "PARSE_ERROR"}, + status_code=400, + ) + + reason: str | None = data.get("reason") + authorized_by: str | None = data.get("authorized_by") + tool_names: list[str] | None = data.get("tool_names") + server_identity_raw: dict[str, Any] | None = data.get("server_identity") + + if not reason or not isinstance(reason, str): + return JSONResponse( + {"error": "'reason' is required", "error_code": "MISSING_FIELD"}, + status_code=422, + ) + if not authorized_by or not isinstance(authorized_by, str): + return JSONResponse( + {"error": "'authorized_by' is required", "error_code": "MISSING_FIELD"}, + status_code=422, + ) + if not tool_names or not isinstance(tool_names, list) or not all(isinstance(n, str) for n in tool_names): + return JSONResponse( + {"error": "'tool_names' must be a non-empty list of strings", "error_code": "MISSING_FIELD"}, + status_code=422, + ) + if not server_identity_raw or not isinstance(server_identity_raw, dict): + return JSONResponse( + {"error": "'server_identity' is required", "error_code": "MISSING_FIELD"}, + status_code=422, + ) + + required_si_fields = ("display_name", "url", "tls_fingerprint") + missing = [f for f in required_si_fields if not server_identity_raw.get(f)] + if missing: + return JSONResponse( + { + "error": f"server_identity missing fields: {missing}", + "error_code": "MISSING_FIELD", + }, + status_code=422, + ) + + try: + server = ServerIdentity( + display_name=server_identity_raw["display_name"], + url=server_identity_raw["url"], + tls_fingerprint=server_identity_raw["tls_fingerprint"], + spiffe_id=server_identity_raw.get("spiffe_id"), + transport=server_identity_raw.get("transport", "http-sse"), + rotation_mode=server_identity_raw.get("rotation_mode", "key-pinned"), + ) + except (KeyError, TypeError) as exc: + return JSONResponse( + {"error": f"invalid server_identity: {exc}", "error_code": "INVALID_FIELD"}, + status_code=422, + ) + + added: list[str] = [] + for tool_name in tool_names: + entry = CatalogEntry( + tool_name=tool_name, + server=server, + approved_definition=ApprovedDefinition( + description=f"Break-glass exception: {reason}", + input_schema={}, + output_schema=None, + ), + definition_hash="sha256:" + "0" * 64, + compliance_domain="external", + requires_baa=False, + sensitivity_level="public", + added_at="", + approved_by=authorized_by, + ) + self._proxy._catalog.add_exception(entry, reason=reason, authorized_by=authorized_by) + added.append(tool_name) + + logger.warning( + "BREAK_GLASS_EXCEPTION_ADDED: tools=%s reason=%r authorized_by=%r", + added, + reason, + authorized_by, + ) + + return JSONResponse( + { + "status": "ok", + "added_tools": added, + "reason": reason, + "authorized_by": authorized_by, + }, + status_code=201, + ) + async def _session_reset(self, request: Request) -> Response: """POST /sessions/{session_id}/reset — operator-only session sensitivity reset.""" if self._session is None or self._audit_chain is None: diff --git a/src/cmcp_gateway/session/manager.py b/src/cmcp_gateway/session/manager.py index 12d7f6a5..df4455fa 100644 --- a/src/cmcp_gateway/session/manager.py +++ b/src/cmcp_gateway/session/manager.py @@ -97,11 +97,15 @@ def close_session( ) catalog = ctx.catalog - # Detect catalog exceptions — entries where catalog_exception=True. + # Collect catalog exceptions from the runtime exception list (richer metadata). catalog_exceptions: list[dict[str, str]] = [ - {"tool_name": name} - for name, entry in catalog.entries.items() - if entry.catalog_exception + { + "tool_name": exc.tool_name, + "reason": exc.reason, + "authorized_by": exc.authorized_by, + "added_at": exc.added_at, + } + for exc in catalog.exceptions ] catalog_info = ToolCatalogInfo( hash=catalog.catalog_hash, diff --git a/tests/unit/test_break_glass.py b/tests/unit/test_break_glass.py new file mode 100644 index 00000000..e0c0dc56 --- /dev/null +++ b/tests/unit/test_break_glass.py @@ -0,0 +1,383 @@ +"""Tests for break-glass catalog exception handling (issue #76).""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from starlette.testclient import TestClient + +from cmcp_gateway.audit.chain import AuditChain +from cmcp_gateway.catalog.loader import ( + ApprovedDefinition, + CatalogEntry, + CatalogException, + ServerIdentity, + ToolCatalog, +) +from cmcp_gateway.config import AttestationConfig, Config, EnforcementMode +from cmcp_gateway.mcp.server import MCPServer +from cmcp_gateway.policy.evaluator import PolicyDecision, PolicyEvaluator +from cmcp_gateway.session.state import SessionState + + +# ── Helpers ─────────────────────────────────────────────────────────────────── + + +def _make_server_identity(url: str = "https://emergency.example.com/mcp") -> ServerIdentity: + return ServerIdentity( + display_name="Emergency Server", + url=url, + tls_fingerprint="SHA256:EMER/GENCY==", + spiffe_id=None, + transport="http-sse", + rotation_mode="key-pinned", + ) + + +def _make_catalog(*tools: str) -> ToolCatalog: + entries = {} + for t in (tools or ("normal.tool",)): + entries[t] = CatalogEntry( + tool_name=t, + server=_make_server_identity("https://normal.example.com/mcp"), + approved_definition=ApprovedDefinition(description="normal", input_schema={}, output_schema=None), + definition_hash="sha256:" + "0" * 64, + compliance_domain="external", + requires_baa=False, + sensitivity_level="public", + added_at="2026-06-01T00:00:00Z", + approved_by="test", + ) + return ToolCatalog(entries=entries, catalog_hash="sha256:" + "a" * 64) + + +def _make_proxy(catalog: ToolCatalog) -> MagicMock: + evaluator = MagicMock(spec=PolicyEvaluator) + decision = PolicyDecision( + allowed=True, + enforcement_mode=EnforcementMode.ENFORCING, + rule_matched=None, + advice={}, + evaluation_ms=0.1, + would_have_denied=False, + ) + evaluator.evaluate.return_value = decision + evaluator.authorize_egress.return_value = decision + + proxy = MagicMock() + proxy._catalog = catalog + proxy.call_tool = AsyncMock(return_value=MagicMock( + allowed=True, + deny_reason=None, + response="ok", + audit_entry_hash="sha256:" + "0" * 64, + would_have_denied=False, + latency_us=100, + )) + return proxy + + +def _make_server(catalog: ToolCatalog | None = None, bearer_token: str | None = None) -> MCPServer: + if catalog is None: + catalog = _make_catalog() + proxy = _make_proxy(catalog) + with patch("cmcp_gateway.mcp.server.StatelessKernel"): + return MCPServer(proxy, bearer_token=bearer_token) + + +# ── Unit: ToolCatalog.add_exception ────────────────────────────────────────── + + +def test_add_exception_does_not_change_catalog_hash(): + catalog = _make_catalog("normal.tool") + original_hash = catalog.catalog_hash + + exc_entry = CatalogEntry( + tool_name="emergency.tool", + server=_make_server_identity(), + approved_definition=ApprovedDefinition(description="emergency", input_schema={}, output_schema=None), + definition_hash="sha256:" + "0" * 64, + compliance_domain="external", + requires_baa=False, + sensitivity_level="public", + added_at="", + approved_by="ops-team", + ) + catalog.add_exception(exc_entry, reason="P0 incident", authorized_by="ops@example.com") + + assert catalog.catalog_hash == original_hash + + +def test_add_exception_makes_entry_callable(): + catalog = _make_catalog() + exc_entry = CatalogEntry( + tool_name="emergency.tool", + server=_make_server_identity(), + approved_definition=ApprovedDefinition(description="emergency", input_schema={}, output_schema=None), + definition_hash="sha256:" + "0" * 64, + compliance_domain="external", + requires_baa=False, + sensitivity_level="public", + added_at="", + approved_by="ops@example.com", + ) + catalog.add_exception(exc_entry, reason="incident", authorized_by="ops@example.com") + + found = catalog.lookup("emergency.tool") + assert found is not None + assert found.catalog_exception is True + + +def test_add_exception_records_metadata(): + catalog = _make_catalog() + exc_entry = CatalogEntry( + tool_name="emergency.tool", + server=_make_server_identity(), + approved_definition=ApprovedDefinition(description="emergency", input_schema={}, output_schema=None), + definition_hash="sha256:" + "0" * 64, + compliance_domain="external", + requires_baa=False, + sensitivity_level="public", + added_at="", + approved_by="ops@example.com", + ) + catalog.add_exception(exc_entry, reason="P0 outage", authorized_by="on-call@example.com") + + assert len(catalog.exceptions) == 1 + exc = catalog.exceptions[0] + assert isinstance(exc, CatalogException) + assert exc.tool_name == "emergency.tool" + assert exc.reason == "P0 outage" + assert exc.authorized_by == "on-call@example.com" + assert exc.added_at # non-empty ISO timestamp + + +def test_multiple_exceptions_tracked(): + catalog = _make_catalog() + for i in range(3): + e = CatalogEntry( + tool_name=f"emergency.tool.{i}", + server=_make_server_identity(), + approved_definition=ApprovedDefinition(description="x", input_schema={}, output_schema=None), + definition_hash="sha256:" + "0" * 64, + compliance_domain="external", + requires_baa=False, + sensitivity_level="public", + added_at="", + approved_by="ops", + ) + catalog.add_exception(e, reason=f"reason {i}", authorized_by="ops") + + assert len(catalog.exceptions) == 3 + assert len(catalog.entries) == 4 # 1 normal + 3 exceptions + + +# ── Integration: POST /catalog/exception endpoint ──────────────────────────── + + +_SI_PAYLOAD = { + "display_name": "Emergency Server", + "url": "https://emergency.example.com/mcp", + "tls_fingerprint": "SHA256:EMER/GENCY==", + "transport": "http-sse", + "rotation_mode": "key-pinned", +} + + +def test_catalog_exception_endpoint_adds_tools(): + catalog = _make_catalog() + server = _make_server(catalog) + client = TestClient(server.app, raise_server_exceptions=False) + + resp = client.post("/catalog/exception", json={ + "server_identity": _SI_PAYLOAD, + "reason": "P0 incident — fallback server required", + "authorized_by": "on-call@example.com", + "tool_names": ["emergency.search", "emergency.fetch"], + }) + + assert resp.status_code == 201 + body = resp.json() + assert body["status"] == "ok" + assert "emergency.search" in body["added_tools"] + assert "emergency.fetch" in body["added_tools"] + assert catalog.lookup("emergency.search") is not None + assert catalog.lookup("emergency.fetch") is not None + + +def test_catalog_exception_preserves_hash(): + catalog = _make_catalog() + original_hash = catalog.catalog_hash + server = _make_server(catalog) + client = TestClient(server.app, raise_server_exceptions=False) + + client.post("/catalog/exception", json={ + "server_identity": _SI_PAYLOAD, + "reason": "emergency", + "authorized_by": "ops@example.com", + "tool_names": ["emergency.tool"], + }) + + assert catalog.catalog_hash == original_hash + + +def test_catalog_exception_endpoint_missing_reason(): + server = _make_server() + client = TestClient(server.app, raise_server_exceptions=False) + + resp = client.post("/catalog/exception", json={ + "server_identity": _SI_PAYLOAD, + "authorized_by": "ops@example.com", + "tool_names": ["t"], + }) + + assert resp.status_code == 422 + assert resp.json()["error_code"] == "MISSING_FIELD" + + +def test_catalog_exception_endpoint_missing_authorized_by(): + server = _make_server() + client = TestClient(server.app, raise_server_exceptions=False) + + resp = client.post("/catalog/exception", json={ + "server_identity": _SI_PAYLOAD, + "reason": "incident", + "tool_names": ["t"], + }) + + assert resp.status_code == 422 + + +def test_catalog_exception_endpoint_missing_tool_names(): + server = _make_server() + client = TestClient(server.app, raise_server_exceptions=False) + + resp = client.post("/catalog/exception", json={ + "server_identity": _SI_PAYLOAD, + "reason": "incident", + "authorized_by": "ops@example.com", + "tool_names": [], + }) + + assert resp.status_code == 422 + + +def test_catalog_exception_endpoint_missing_server_identity(): + server = _make_server() + client = TestClient(server.app, raise_server_exceptions=False) + + resp = client.post("/catalog/exception", json={ + "reason": "incident", + "authorized_by": "ops@example.com", + "tool_names": ["t"], + }) + + assert resp.status_code == 422 + + +def test_catalog_exception_endpoint_bad_json(): + server = _make_server() + client = TestClient(server.app, raise_server_exceptions=False) + + resp = client.post( + "/catalog/exception", + content=b"not json", + headers={"Content-Type": "application/json"}, + ) + + assert resp.status_code == 400 + assert resp.json()["error_code"] == "PARSE_ERROR" + + +# ── Integration: proxy logs BREAK_GLASS_ACTIVE ─────────────────────────────── + + +def _make_real_proxy_for_break_glass(): + """Build a minimal CMCPProxy with one exception entry to test audit logging.""" + from cmcp_gateway.mcp.proxy import CMCPProxy + + catalog = _make_catalog() + exc_entry = CatalogEntry( + tool_name="emergency.tool", + server=_make_server_identity(), + approved_definition=ApprovedDefinition(description="emergency", input_schema={}, output_schema=None), + definition_hash="sha256:" + "0" * 64, + compliance_domain="external", + requires_baa=False, + sensitivity_level="public", + added_at="", + approved_by="ops@example.com", + ) + catalog.add_exception(exc_entry, reason="test incident", authorized_by="ops@example.com") + + evaluator = MagicMock(spec=PolicyEvaluator) + decision = PolicyDecision( + allowed=True, + enforcement_mode=EnforcementMode.ENFORCING, + rule_matched=None, + advice={}, + evaluation_ms=0.1, + would_have_denied=False, + ) + evaluator.evaluate.return_value = decision + evaluator.authorize_egress.return_value = decision + evaluator.bundle_hash = "sha256:" + "0" * 64 + evaluator.enforcement_mode = EnforcementMode.ENFORCING + + session = SessionState(session_id="test-session-id") + chain = AuditChain(session_id="test-session-id") + config = Config( + attestation=AttestationConfig( + provider="software-only", + enforcement_mode=EnforcementMode.ENFORCING, + ) + ) + + mock_agt_result = MagicMock(sensitivity_tags=[], injection_detected=False, modified_response=b"ok") + + with patch("cmcp_gateway.mcp.proxy.MCPGateway"), \ + patch("cmcp_gateway.mcp.proxy.MCPResponseScanner"): + proxy = CMCPProxy( + catalog=catalog, + policy_evaluator=evaluator, + session=session, + audit_chain=chain, + config=config, + ) + proxy._mcp_gateway = MagicMock() + proxy._mcp_gateway.call_tool = AsyncMock(return_value=mock_agt_result) + + return proxy, chain + + +@pytest.mark.asyncio +async def test_break_glass_call_logs_audit_entry(): + """Calls routed via a catalog exception produce a break_glass_used audit entry.""" + proxy, chain = _make_real_proxy_for_break_glass() + await proxy.call_tool("call-1", "emergency.tool", {}) + + entry_types = [e.entry_type for e in chain.entries] + assert "break_glass_used" in entry_types + + +@pytest.mark.asyncio +async def test_break_glass_call_logs_warning(caplog): + """BREAK_GLASS_ACTIVE is logged as a warning.""" + import logging + proxy, _ = _make_real_proxy_for_break_glass() + + with caplog.at_level(logging.WARNING, logger="cmcp_gateway.mcp.proxy"): + await proxy.call_tool("call-1", "emergency.tool", {}) + + assert any("BREAK_GLASS_ACTIVE" in r.message for r in caplog.records) + + +@pytest.mark.asyncio +async def test_normal_tool_no_break_glass_audit_entry(): + """Normal catalog entries do not produce a break_glass_used audit entry.""" + proxy, chain = _make_real_proxy_for_break_glass() + await proxy.call_tool("call-1", "normal.tool", {}) + + entry_types = [e.entry_type for e in chain.entries] + assert "break_glass_used" not in entry_types