feat(catalog): break-glass catalog exception handling (#76)

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"]

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:<hex> measured into the TEE report
+    catalog_hash: str  # sha256:<hex> 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()

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

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:

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,