feat(catalog): detect upstream tool-definition drift, and stop the scanner failing open

LIMITATIONS.md

@@ -16,6 +16,15 @@ On Azure confidential VMs (`AzureCVMProvider`), SEV-SNP runs behind a Hyper-V pa
 **Operator-controlled key material**
 The TEE-sealed signing key is generated inside the enclave and cannot be extracted by a privileged operator under normal circumstances. However, if the operator can substitute the TEE firmware, modify the enclave startup measurement, or compromise the SPIRE infrastructure that issues SVIDs, they can effectively control what key material is used. The trust root is the hardware platform vendor (AMD, Intel, or the TPM manufacturer), not cMCP. Deployments that do not independently verify the attestation report before routing traffic treat attestation as post-hoc audit evidence only.
 
+**Upstream tool-definition drift: checked on first contact, not continuously**
+cMCP compares what each upstream server advertises against the approved catalog entry (threat model P4.2, rug-pull via tool-definition mutation). The comparison is a digest of the semantic triple: description, input schema, output schema. It uses only the standard library, so it does not weaken when the optional `agent-os-kernel` scanner is absent; that scanner classifies the kind of change and is not the control. A mismatch fails the call closed by default, is written to the audit chain, and sets `tool_catalog.drift_detected` in the session's TRACE Claim. Three things it does not do, all of them real gaps rather than theoretical ones:
+
+- **It runs once per server per session, on first contact.** A server that mutates a tool definition mid-session is not caught until a new session starts. Re-listing tools on every call would make the check expensive enough that operators would turn it off, which is worse than a check with a stated window.
+- **It is not driven by `notifications/tools/list_changed`.** The gateway does not subscribe to that notification. Only stdio upstreams could carry one today, since HTTP upstreams are plain request/response here, so notification-driven detection would silently cover one transport and not the other. First-contact checking covers both the same way.
+- **A server that will not answer `tools/list` is recorded as unchecked, not denied.** Nothing in MCP obliges a server to answer, and denying on silence would take out deployments whose servers simply do not implement it. `unchecked` is not a pass, and it is visible in the logs, but it does not block a call.
+
+Separately, the approved description rather than the live one is what the gateway serves to the agent on `tools/list`, so a mutated description does not reach the model through cMCP even in the windows above. That is a structural property of proxying an approved catalog, not a detection result, and it does not extend to the tool's behaviour once called.
+
 **Phase 2 completeness: server-side attestation**
 Phase 1 attests the gateway boundary. It does not attest what happens on the other side of that boundary. The `tool_transcript.hash` field in the TRACE Claim records a hash of the audit chain tip, but the tool transcript binding that ties a specific tool execution to a specific response is Phase 2 work. Phase 1 partially addresses P1.4 (transitive trust into upstream dependencies) and P4.1 (typosquatted packages added to catalog) -- both are fully closed by Phase 2. Any compliance claim that relies on server-side proof must wait for Phase 2.
 

src/cmcp_runtime/catalog/loader.py

@@ -134,6 +134,61 @@ def _compute_definition_hash(definition: dict[str, Any]) -> str:
     return f"sha256:{_sha256_hex(canonical.encode())}"
 
 
+def definition_digest(
+    description: str,
+    input_schema: dict[str, Any] | None,
+    output_schema: dict[str, Any] | None,
+) -> str:
+    """Canonical digest of the semantic triple that identifies a tool.
+
+    Deliberately distinct from ``definition_hash``. That one covers the catalog
+    entry exactly as written on disk, which is what you want for detecting an
+    edited catalog file and the wrong thing for comparing against an upstream
+    server: the server never sees our file layout, so an extra or reordered key
+    on our side would read as drift on theirs. This digest covers only what both
+    sides can be expected to agree on.
+
+    Uses the same canonical JSON form as :func:`_compute_definition_hash`, so the
+    two can never disagree about how a value is serialised.
+    """
+    canonical = json.dumps(
+        {
+            "description": description,
+            "input_schema": input_schema or {},
+            "output_schema": output_schema,
+        },
+        sort_keys=True,
+        separators=(",", ":"),
+        ensure_ascii=True,
+    )
+    return f"sha256:{_sha256_hex(canonical.encode())}"
+
+
+def approved_definition_digest(definition: ApprovedDefinition) -> str:
+    """Digest of an approved definition, for the upstream drift comparison."""
+    return definition_digest(
+        definition.description,
+        definition.input_schema,
+        definition.output_schema,
+    )
+
+
+def advertised_definition_digest(tool: dict[str, Any]) -> str:
+    """Digest of a tool as an upstream server advertises it over MCP.
+
+    MCP ``tools/list`` uses camelCase (``inputSchema``) while the catalog uses
+    snake_case. Both spellings are accepted so a server that happens to answer in
+    the catalog's shape is not reported as drifted for a naming difference.
+    """
+    input_schema = tool.get("inputSchema", tool.get("input_schema"))
+    output_schema = tool.get("outputSchema", tool.get("output_schema"))
+    return definition_digest(
+        str(tool.get("description", "")),
+        input_schema if isinstance(input_schema, dict) else {},
+        output_schema if isinstance(output_schema, dict) else None,
+    )
+
+
 def _catalog_hash(raw_entries: list[dict[str, Any]]) -> str:
     """SHA-256 of canonical JSON of entries sorted by tool_name."""
     sorted_entries = sorted(raw_entries, key=lambda e: e["tool_name"])

src/cmcp_runtime/catalog/scanner.py

@@ -29,21 +29,35 @@
 
 @dataclass
 class CatalogScanResult:
-    """Result of scanning the full tool catalog at load time."""
+    """Result of scanning the full tool catalog at load time.
+
+    ``available`` is the difference between "we looked and found nothing" and
+    "we never looked". Callers must not read ``safe`` without it: this scanner
+    is backed by an optional dependency, and a control that reports safe when it
+    is simply absent is worse than one that reports nothing at all.
+    """
 
     safe: bool
     tools_scanned: int
     tools_flagged: int
     threats: list[dict[str, str]]  # [{tool_name, threat_type, severity, description}]
+    available: bool = True
 
 
 @dataclass
 class DriftResult:
-    """Result of a rug-pull / drift check on a single tool."""
+    """Result of a rug-pull / drift check on a single tool.
+
+    As with :class:`CatalogScanResult`, ``drifted=False`` means nothing on its
+    own when ``available`` is False. The authoritative drift decision is the
+    digest comparison in the proxy, which needs no optional dependency; this
+    scanner only classifies what kind of change it was.
+    """
 
     tool_name: str
     drifted: bool
     threats: list[dict[str, str]]
+    available: bool = True
 
 
 class CatalogScanner:
@@ -84,10 +98,11 @@ def scan_catalog(self, catalog: ToolCatalog) -> CatalogScanResult:
         """
         if not self._available or self._scanner is None:
             return CatalogScanResult(
-                safe=True,
-                tools_scanned=len(catalog.entries),
+                safe=False,
+                tools_scanned=0,
                 tools_flagged=0,
                 threats=[],
+                available=False,
             )
 
         all_threats: list[dict[str, str]] = []
@@ -152,7 +167,9 @@ def check_drift(
         if the definition has changed since the catalog was sealed.
         """
         if not self._available or self._scanner is None:
-            return DriftResult(tool_name=tool_name, drifted=False, threats=[])
+            return DriftResult(
+                tool_name=tool_name, drifted=False, threats=[], available=False
+            )
 
         try:
             threats = self._scanner.check_rug_pull(

src/cmcp_runtime/cli.py

@@ -50,6 +50,7 @@ def build_server(ctx: RuntimeContext) -> MCPServer:
         attestation_generated_at=ctx.attestation_report.attestation_generated_at,
         attestation_validity_seconds=ctx.attestation_report.attestation_validity_seconds,
         attestation_platform=attestation_platform,
+        catalog_scanner=ctx.catalog_scanner,
     )
     # AUTH-001: the token validated in run_startup must reach the server, otherwise
     # every protected endpoint is reachable unauthenticated.

src/cmcp_runtime/config.py

@@ -40,6 +40,27 @@ class StalenessPolicy(StrEnum):
     WARN_ONLY = "warn_only"
 
 
+class DriftPolicy(StrEnum):
+    FAIL_CLOSED = "fail_closed"
+    WARN_ONLY = "warn_only"
+
+
+@dataclass
+class CatalogConfig:
+    """Behaviour when an upstream server stops matching its approved entry."""
+
+    drift_policy: DriftPolicy = DriftPolicy.FAIL_CLOSED
+    """What to do when a server advertises a tool whose definition differs from
+    the approved one (threat model P4.2, rug-pull via tool-definition mutation).
+
+    Fail closed by default. A mismatch means the server is not offering what was
+    reviewed, and the premise of this gateway is that the approved definition is
+    the one that runs. ``warn_only`` records the drift and routes the call anyway.
+    It exists so an operator can measure how noisy their own fleet is before
+    turning enforcement on, not as a resting state.
+    """
+
+
 @dataclass
 class KillSwitchConfig:
     enabled: bool = False
@@ -96,6 +117,7 @@ class Config:
     agent_manifest: AgentManifestConfig = field(default_factory=AgentManifestConfig)
     kill_switch: KillSwitchConfig = field(default_factory=KillSwitchConfig)
     sensitivity: SensitivityConfig = field(default_factory=SensitivityConfig)
+    catalog: CatalogConfig = field(default_factory=CatalogConfig)
     policy_bundle_path: str = "policies/"
     catalog_path: str = "catalog.json"
     listen_addr: str = "0.0.0.0:8443"
@@ -366,6 +388,15 @@ def load_config(path: str) -> Config:
             + ", ".join(_KINDS)
         )
 
+    catalog_raw = raw.get("catalog", {})
+    if not isinstance(catalog_raw, dict):
+        raise ConfigError("catalog must be a mapping")
+    try:
+        drift_policy = DriftPolicy(catalog_raw.get("drift_policy", "fail_closed"))
+    except ValueError as err:
+        valid = [d.value for d in DriftPolicy]
+        raise ConfigError(f"catalog.drift_policy must be one of {valid}") from err
+
     max_bytes = raw.get("max_response_size_bytes", 2 * 1024 * 1024)
     if not isinstance(max_bytes, int) or max_bytes <= 0:
         raise ConfigError("max_response_size_bytes must be a positive integer")
@@ -450,6 +481,7 @@ def load_config(path: str) -> Config:
             min_calls=ks_min_calls,
         ),
         sensitivity=SensitivityConfig(vocabulary=sensitivity_vocabulary),
+        catalog=CatalogConfig(drift_policy=drift_policy),
         policy_bundle_path=policy_bundle_path,
         catalog_path=catalog_path,
         listen_addr=listen_addr,

src/cmcp_runtime/mcp/proxy.py

@@ -25,8 +25,14 @@
 from agent_os.mcp_response_scanner import MCPResponseScanner
 
 from cmcp_runtime.audit.chain import AuditChain
-from cmcp_runtime.catalog.loader import CatalogEntry, ToolCatalog
-from cmcp_runtime.config import Config
+from cmcp_runtime.catalog.loader import (
+    CatalogEntry,
+    ToolCatalog,
+    advertised_definition_digest,
+    approved_definition_digest,
+)
+from cmcp_runtime.catalog.scanner import CatalogScanner
+from cmcp_runtime.config import Config, DriftPolicy
 from cmcp_runtime.errors import PolicyDeny, UpstreamToolError, UpstreamUnavailable
 from cmcp_runtime.mcp import tls_pinning
 from cmcp_runtime.mcp.stdio import StdioServer
@@ -182,6 +188,7 @@ def __init__(
         attestation_validity_seconds: int = 86400,
         catalog_hash: str | None = None,
         attestation_platform: str = "unknown",
+        catalog_scanner: CatalogScanner | None = None,
     ) -> None:
         self._catalog = catalog
         self._policy = policy_evaluator
@@ -228,6 +235,12 @@ def __init__(
         self._provenance: dict[tuple[str, ...], ProvenanceResult] = {}
         # Servers already warned about unenforceable pinning (warn once each).
         self._tls_pin_warned: set[str] = set()
+        # #521: servers whose advertised tool definitions have been compared against
+        # the catalog. Cached per server for the same reason provenance is: one
+        # tools/list round trip per server per session is affordable, one per call
+        # is not, and a check expensive enough to hurt is a check that gets disabled.
+        self._drift_checked: set[tuple[str, ...]] = set()
+        self._catalog_scanner = catalog_scanner
 
     def rebind_session(self, session: SessionState, audit_chain: AuditChain) -> None:
         """
@@ -358,6 +371,92 @@ async def _advertised_tools(self, entry: CatalogEntry) -> list[dict[str, Any]] |
         tools = result.get("tools") if isinstance(result, dict) else None
         return tools if isinstance(tools, list) else None
 
+    async def _check_upstream_drift(self, entry: CatalogEntry) -> bool:
+        """Compare what a server advertises against what we approved (P4.2).
+
+        Runs once per server per session, on first contact. Returns True when the
+        call must be denied.
+
+        The authoritative comparison is a digest of the semantic triple
+        (description, input schema, output schema), computed with the standard
+        library on both sides. That matters: the AGT scanner is an optional
+        dependency, and a control that stops working when a dependency is missing
+        is not a control. The scanner only classifies what kind of change it was.
+
+        A server that will not answer ``tools/list`` is recorded as unchecked and
+        is NOT denied. Denying would take out every deployment whose servers do
+        not implement it, and this check would be switched off within a day. That
+        is a real gap and LIMITATIONS.md says so rather than leaving it implied.
+        """
+        key = _server_provenance_key(entry)
+        if key in self._drift_checked:
+            return self._session.catalog_drift
+        self._drift_checked.add(key)
+
+        advertised = await self._advertised_tools(entry)
+        if advertised is None:
+            logger.info(
+                "upstream drift: server=%s outcome=unchecked (server would not list tools)",
+                key,
+            )
+            return self._session.catalog_drift
+
+        by_name = {
+            t.get("name"): t
+            for t in advertised
+            if isinstance(t, dict) and isinstance(t.get("name"), str)
+        }
+        drifted: list[tuple[str, str]] = []
+        for tool_name, catalog_entry in self._catalog.entries.items():
+            if _server_provenance_key(catalog_entry) != key:
+                continue
+            offered = by_name.get(tool_name)
+            if offered is None:
+                drifted.append((tool_name, "withdrawn"))
+                continue
+            if advertised_definition_digest(offered) != approved_definition_digest(
+                catalog_entry.approved_definition
+            ):
+                drifted.append((tool_name, "definition_changed"))
+
+        if not drifted:
+            logger.info("upstream drift: server=%s outcome=match", key)
+            return self._session.catalog_drift
+
+        fail_closed = self._config.catalog.drift_policy is DriftPolicy.FAIL_CLOSED
+        for tool_name, kind in drifted:
+            classification = kind
+            if self._catalog_scanner is not None and (offered := by_name.get(tool_name)):
+                result = self._catalog_scanner.check_drift(
+                    tool_name=tool_name,
+                    server_name=entry.server.display_name or entry.server.url,
+                    current_definition=offered,
+                )
+                if result.available and result.threats:
+                    classification = ";".join(
+                        t.get("threat_type", "?") for t in result.threats
+                    )
+            logger.error(
+                "UPSTREAM_CATALOG_DRIFT tool=%s server=%s kind=%s policy=%s",
+                tool_name,
+                key,
+                classification,
+                self._config.catalog.drift_policy.value,
+            )
+            if tool_name not in self._session.upstream_drift_tools:
+                self._session.upstream_drift_tools.append(tool_name)
+            self._audit.append(
+                "catalog_drift",
+                tool_name=tool_name,
+                detail={"kind": kind, "classification": classification, "source": "upstream"},
+                session_sensitivity_before=self._session.max_sensitivity,
+                session_sensitivity_after=self._session.max_sensitivity,
+            )
+
+        if fail_closed:
+            self._session.catalog_drift = True
+        return self._session.catalog_drift
+
     async def _check_provenance(self, entry: CatalogEntry) -> ProvenanceResult:
         key = _server_provenance_key(entry)
         result = self._provenance.get(key)
@@ -692,6 +791,35 @@ class above the tool's catalogued sensitivity_level. It can never lower
                 audit_entry_hash=self._audit.chain_tip,
             )
 
+        # Step 1a (#521): does this server still offer what we approved? First
+        # contact with each server only, so the cost is one tools/list per server
+        # per session. Placed after the catalog lookup because it needs the entry
+        # to know which server to ask, and before the policy decision because a
+        # server that has been swapped underneath us should not reach Cedar at all.
+        if await self._check_upstream_drift(entry):
+            elapsed_ms = (time.perf_counter() - t0) * 1000
+            self._record_call(
+                tool_name=tool_name,
+                called_at=called_at,
+                duration_ms=elapsed_ms,
+                allowed=False,
+                sensitivity_before=sensitivity_before,
+                stage_results={"catalog": "deny"},
+                call_id=call_id,
+                catalog_entry=entry,
+                policy_decision="deny",
+            )
+            return CallResult(
+                call_id=call_id,
+                tool_name=tool_name,
+                allowed=False,
+                would_have_denied=False,
+                response=None,
+                deny_reason="catalog_drift",
+                latency_us=int(elapsed_ms * 1000),
+                audit_entry_hash=self._audit.chain_tip,
+            )
+
         # #479 piece 2: this call's own class, catalog floor raised by any
         # declared_data_class. _max_sensitivity can only return the higher of
         # the two labels, ties favour the catalog value, so an unrecognised or

src/cmcp_runtime/session/manager.py

@@ -251,9 +251,14 @@ def close_session(
             }
             for exc in catalog.exceptions
         ]
+        # #521: this was hardcoded False, so the claim carried a drift field that
+        # could never be true. Both drift kinds count: our own catalog changing
+        # under a running gateway, and an upstream server advertising a tool that
+        # no longer matches its approved definition. The second stays visible here
+        # even under warn_only, where the calls were routed anyway.
         catalog_info = ToolCatalogInfo(
             hash=catalog.catalog_hash,
-            drift_detected=False,
+            drift_detected=bool(state.catalog_drift or state.upstream_drift_tools),
         )
 
         # Build call summary from chain entries.