Enforce sensitivity ceilings at tool and response boundaries

CHANGELOG.md

@@ -9,6 +9,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Security
 
+- Add optional operator-owned tool and caller-response sensitivity ceilings.
+  They enforce accumulated/catalog/declared classification independently of
+  Cedar mode, before discovery and dispatch and before response release. Strict
+  sink policy suppresses captured stdio stderr content; upstream error logs no
+  longer echo tool messages. This is conservative label enforcement, not semantic
+  information-flow verification or automatic declassification.
+
 - Add verifier-owned SNP platform policy to native and Azure verification and
   the public TRACE verifier. Explicit requirements fail closed without signed
   SNP evidence. The optional policy covers PLATFORM_INFO, not guest DEBUG,

LIMITATIONS.md

@@ -2,6 +2,22 @@
 
 This document describes what cMCP does not prevent, where its guarantees end, and what operators and verifiers must address through separate controls.
 
+## Optional sink sensitivity ceilings
+
+The optional `sink_policy` enforces operator-configured ceilings
+for tools and caller responses, using the accumulated session classification,
+catalog floor, and caller-declared class. It cannot discover unlabelled secrets,
+track exact data dependencies, or establish safe declassification of a summary.
+Equal-ranked labels are equivalent for this gate, not isolated compartments.
+Permitted remote tools still need their own confidentiality protections.
+
+With this option, gateway-created stdio servers suppress captured stderr content.
+Other application/third-party logs, direct agent egress, and audit metadata remain
+outside that control. Audit payload hashes can reveal predictable values through
+guessing and do not make an audit bundle safe to publish. The startup sink policy
+is immutable per proxy but is not separately bound into a remote attestation
+claim. Protect and review deployment configuration and operator reset authority.
+
 ## What cMCP does not prevent
 
 **Prompt injection into Cedar policy**

docs/configuration.md

@@ -180,6 +180,11 @@ catalog_path: ./catalog.json
 
 ## Production hardening checklist
 
+An optional `sink_policy` sets hard per-tool and caller-response sensitivity
+ceilings, including when Cedar uses advisory mode. It also suppresses captured
+stdio stderr content. See [sink sensitivity ceilings](spec/sink-policy.md) for
+configuration, classification assumptions, and the remaining audit/log limits.
+
 - Set `attestation.enforcement_mode` to `enforcing`. Advisory mode provides no blocking protection against policy violations.
 - Set `CMCP_CATALOG_HASH` to the SHA-256 of the approved `catalog.json`. The gateway fails closed at startup if this is unset in non-dev mode, but setting it explicitly pins the approved catalog hash and prevents silent substitution.
 - Configure `agent_manifest.path`, `agent_manifest.trust_anchor_path`, and `agent_manifest.authenticated_subject` for agents with signed manifests. The runtime will refuse to start if the signed manifest does not bind the authenticated agent subject to the loaded policy bundle and catalog hashes.

docs/spec/sink-policy.md

@@ -0,0 +1,70 @@
+# Optional sensitivity ceilings for tool and response sinks
+
+`sink_policy` adds hard admission ceilings to the actual gateway call path.
+It is supplied by the operator at startup, independent of tool arguments and
+Cedar advisory/silent modes. Omit the block to preserve existing behavior.
+
+```yaml
+sink_policy:
+  tool_max_sensitivity:
+    records.read: confidential
+    internal.summarize: confidential
+    public.search: public
+  response_max_sensitivity: confidential
+```
+
+Both keys are required. An empty tool map denies all tools. An absent tool,
+unknown sensitivity label, malformed block, or misspelled ceiling fails closed.
+Custom ceilings use the configured additive sensitivity vocabulary. Ceilings
+compare ranks: labels at the same rank are equivalent for this check. They do
+not provide compartment, purpose, recipient identity, or jurisdiction rules.
+
+Before dispatch, every applicable label must fit the tool's ceiling: the
+accumulated session class, the approved catalog's class, and any caller-declared
+class. A caller can increase classification but cannot lower an existing floor.
+After inspection and session-state update, the same rule controls release to
+the caller using `response_max_sensitivity`. The class observed at call entry
+is retained for that call even if an operator resets the session in flight.
+Refusals use `sink_policy:tool_denied` or `sink_policy:response_denied` in the
+ordinary terminal audit path. A response refusal does not undo a tool operation
+that has already executed.
+
+For example, once a session has read a catalogued confidential record, a
+subsequent public-search call is denied even if it contains a clean-looking
+summary and declares itself public. There is no automatic declassification or
+model-generated approval. A session reset is an operator action; it must not be
+used to carry the same secret into a fresh public session.
+
+## Logging boundary
+
+When this policy is configured, gateway-created stdio servers suppress captured
+stderr content and log its byte count only. Ordinary upstream error logs retain
+the error code rather than the upstream's potentially private message. Proxy
+Cedar exception logs retain the exception class without a traceback or message.
+These changes constrain these specific log sites, not every diagnostic sink.
+
+The audit chain still includes payload hashes and metadata. Hashes of predictable
+inputs can support guessing attacks, and tool names, identifiers, timing, and
+external evidence can be sensitive. Protect audit exports and application logs
+with deployment access policy; this feature does not encrypt them or make them
+safe for publication. Application, runtime, third-party, and agent-host logs
+remain outside the stderr control.
+
+## Assurance limits
+
+- This is a conservative session-label gate, not semantic information-flow
+  tracking. It relies on accurate catalog/input classification. It does not
+  discover unlabelled secrets or prove a derived output contains none.
+- The gateway cannot mediate an agent's direct sockets, files, remote model
+  calls, or other paths that bypass it. A permitted remote tool still needs its
+  own confidentiality protections. A catalog entry is not remote attestation.
+- The response ceiling is an operator authorization for this gateway's callers,
+  not per-user clearance negotiation. Run separate deployments where callers
+  require different ceilings.
+- The immutable policy is captured when a proxy is created. Treat configuration
+  and catalog changes as deployment changes; this feature does not add their
+  digest to an attestation claim or implement a remotely verified policy update.
+- Cross-session classification, durable session state, trusted operator resets,
+  and protection of the process/configuration remain deployment obligations.
+  This gate does not establish a distributed total order over simultaneous
+  calls or independently operated gateways.

mkdocs.yml

@@ -176,6 +176,7 @@ nav:
           - Component model: spec/component-model.md
           - Cedar policy: spec/cedar-policy.md
           - Session policy: spec/session-policy.md
+          - Sink sensitivity ceilings: spec/sink-policy.md
           - Policy hot-reload: spec/policy-hot-reload.md
           - Tool identity: spec/tool-identity.md
           - Catalog lifecycle: spec/catalog-lifecycle.md

src/cmcp_runtime/config.py

@@ -13,6 +13,7 @@
 
 from cmcp_runtime.errors import ConfigError
 from cmcp_runtime.session.state import COMPLIANCE_DOMAINS, SENSITIVITY_ORDER
+from cmcp_runtime.sink_policy import SinkPolicy
 
 # TEE-002: read exactly once at import time so the value is immutable for the
 # lifetime of the process. No code may call os.environ.get("CMCP_DEV_MODE")
@@ -153,9 +154,11 @@ class Config:
     #: MUST be bound to an agent identity while the developer default leaves
     #: binding optional. Naming the profile is what lets both be true.
     conformance_profile: str | None = None
+    sink_policy: SinkPolicy | None = None
 
 
 _KNOWN_TOP_KEYS = {
+    "sink_policy",
     "attestation",
     "agent_manifest",
     "catalog",
@@ -522,7 +525,18 @@ def load_config(path: str) -> Config:
             f"{sorted(_KNOWN_CONFORMANCE_PROFILES)}, got {profile!r}"
         )
 
+    sink_policy = None
+    if "sink_policy" in raw:
+        sink_raw = raw["sink_policy"]
+        if not isinstance(sink_raw, dict) or set(sink_raw) != {
+            "tool_max_sensitivity", "response_max_sensitivity",
+        }:
+            raise ConfigError("sink_policy requires exactly tool_max_sensitivity and response_max_sensitivity")
+        sink_policy = SinkPolicy(**sink_raw)
+        sink_policy.validate({**sensitivity_vocabulary, **SENSITIVITY_ORDER})
+
     return Config(
+        sink_policy=sink_policy,
         attestation=AttestationConfig(
             provider=provider,
             enforcement_mode=enforcement_mode,

src/cmcp_runtime/mcp/proxy.py

@@ -54,7 +54,7 @@
 from cmcp_runtime.provenance import ProvenanceResult, check_server_provenance
 from cmcp_runtime.runtime_gateway import GovernancePolicy, MCPGateway, MCPResponseScanner
 from cmcp_runtime.session.call_log import CallLog, CallRecord, SessionCallLog
-from cmcp_runtime.session.state import SessionState, _max_sensitivity
+from cmcp_runtime.session.state import SessionState, _max_sensitivity, effective_sensitivity_order
 
 logger = logging.getLogger(__name__)
 
@@ -252,6 +252,12 @@ def __init__(
         catalog_scanner: CatalogScanner | None = None,
     ) -> None:
         self._catalog = catalog
+        # Capture operator policy once. Per-call arguments cannot replace it,
+        # and changing the Config object later cannot disable this gate.
+        self._sink_policy = config.sink_policy
+        self._sink_order = effective_sensitivity_order(config.sensitivity.vocabulary)
+        if self._sink_policy is not None:
+            self._sink_policy.validate(self._sink_order)
         self._policy = policy_evaluator
         self._session = session
         self._audit = audit_chain
@@ -699,6 +705,7 @@ async def _stdio_for(self, entry: CatalogEntry) -> StdioServer:
             server = StdioServer(
                 entry.server.spawn,
                 allow_unmeasured=self._config.attestation.allow_unmeasured_spawn,
+                log_stderr=self._sink_policy is None,
             )
             await server.start()
             self._stdio_servers[key] = server
@@ -1486,6 +1493,41 @@ class above the tool's catalogued sensitivity_level. It can never lower
 
         _finalization.server_identity = entry.server.url
 
+        # Sink admission precedes even discovery: stdio discovery starts a
+        # child inside the gateway isolation domain. Recheck at Cedar admission
+        # below after discovery awaits in case session classification increased.
+        if self._sink_policy is not None:
+            _finalization.failure_stage = "sink_admission"
+            try:
+                self._sink_policy.require(
+                    tool=tool_name,
+                    labels=(sensitivity_before, self._session.max_sensitivity,
+                            entry.sensitivity_level, *(() if declared_data_class is None
+                                                       else (declared_data_class,))),
+                    order=self._sink_order,
+                )
+            except PolicyDeny as exc:
+                self._append_call_terminal(
+                    _finalization, "tool_call", call_id=call_id, tool_name=tool_name,
+                    server_identity=entry.server.url, policy_decision="deny",
+                    policy_rule_matched=str(exc), request_payload_hash=request_payload_hash,
+                    session_sensitivity_before=sensitivity_before,
+                    session_sensitivity_after=self._session.max_sensitivity,
+                    workflow_id=workflow_id,
+                )
+                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={"sink_policy": "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=str(exc),
+                    latency_us=int(elapsed_ms * 1000), audit_entry_hash=self._audit.chain_tip,
+                )
+
         # Step 3a (#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
@@ -1555,6 +1597,14 @@ class above the tool's catalogued sensitivity_level. It can never lower
         policy_rule: str | None = None
         ingress_advice: dict[str, str] = {}
         try:
+            if self._sink_policy is not None:
+                self._sink_policy.require(
+                    tool=tool_name,
+                    labels=(sensitivity_before, self._session.max_sensitivity,
+                            entry.sensitivity_level, *(() if declared_data_class is None
+                                                       else (declared_data_class,))),
+                    order=self._sink_order,
+                )
             decision = self._policy.evaluate(cedar_context)
             policy_rule = decision.rule_matched
             would_have_denied = decision.would_have_denied
@@ -1607,7 +1657,7 @@ class above the tool's catalogued sensitivity_level. It can never lower
             # POLICY-003: Cedar backend raised an unexpected exception (e.g. malformed
             # policy). Write a fault audit entry so the incident is traceable, then
             # re-raise so server.py can return a generic 500.
-            logger.error("CEDAR_FAULT: tool=%s error=%s", tool_name, exc, exc_info=True)
+            logger.error("CEDAR_FAULT: tool=%s exception_type=%s", tool_name, type(exc).__name__)
             self._finalize_unexpected_call_failure(
                 _finalization,
                 exc,
@@ -1675,7 +1725,8 @@ class above the tool's catalogued sensitivity_level. It can never lower
             )
             _finalization.effect_boundary_state = _EffectBoundaryState.TRANSPORT_RESPONSE_RECEIVED
         except (UpstreamUnavailable, UpstreamToolError) as exc:
-            logger.warning("Upstream call failed: tool=%s error=%s", tool_name, exc)
+            # An upstream error message can contain the complete private input.
+            logger.warning("Upstream call failed: tool=%s code=%s", tool_name, exc.code)
             self._append_call_terminal(
                 _finalization,
                 "fault",
@@ -1865,6 +1916,14 @@ class above the tool's catalogued sensitivity_level. It can never lower
         # Step 5: egress Cedar policy check
         _finalization.failure_stage = "egress_policy"
         try:
+            if self._sink_policy is not None:
+                self._sink_policy.require(
+                    tool=None,
+                    labels=(sensitivity_before, self._session.max_sensitivity,
+                            entry.sensitivity_level, *(() if declared_data_class is None
+                                                       else (declared_data_class,))),
+                    order=self._sink_order,
+                )
             egress_decision = self._policy.authorize_egress(
                 tool_name, response_bytes, self._session, workflow_id=workflow_id
             )

src/cmcp_runtime/mcp/stdio.py

@@ -154,10 +154,12 @@ def __init__(
         *,
         allow_unmeasured: bool = False,
         env: dict[str, str] | None = None,
+        log_stderr: bool = True,
     ) -> None:
         self._spawn = spawn
         self._allow_unmeasured = allow_unmeasured
         self._env = env
+        self._log_stderr = log_stderr
         self._proc: asyncio.subprocess.Process | None = None
         self._lock = asyncio.Lock()
         self._stderr_bytes = 0
@@ -365,6 +367,9 @@ async def _collect_stderr(self) -> None:
             return
         if data:
             self._stderr_bytes += len(data)
+            if not self._log_stderr:
+                logger.warning("stdio server stderr suppressed (%d bytes)", len(data))
+                return
             # Logged, never recorded: diagnostics carry payloads and the audit
             # chain is meant to be shareable.
             logger.warning(

src/cmcp_runtime/sink_policy.py

@@ -0,0 +1,45 @@
+"""Operator-owned sensitivity ceilings; no content-based declassification."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from dataclasses import dataclass
+from types import MappingProxyType
+
+from cmcp_runtime.errors import ConfigError, PolicyDeny
+
+
+@dataclass(frozen=True)
+class SinkPolicy:
+    """An absent tool is denied. The response ceiling applies to the caller.
+
+    Names refer to the deployment's sensitivity ordering, not compartment or
+    purpose permissions. Transport identity remains the catalog's responsibility.
+    """
+
+    tool_max_sensitivity: Mapping[str, str]
+    response_max_sensitivity: str
+
+    def __post_init__(self) -> None:
+        if not isinstance(self.tool_max_sensitivity, Mapping):
+            raise ConfigError("sink_policy.tool_max_sensitivity must be a mapping")
+        if any(not isinstance(k, str) or not k or not isinstance(v, str) or not v
+               for k, v in self.tool_max_sensitivity.items()):
+            raise ConfigError("sink_policy tool names and ceilings must be nonempty strings")
+        if not isinstance(self.response_max_sensitivity, str) or not self.response_max_sensitivity:
+            raise ConfigError("sink_policy.response_max_sensitivity must be a nonempty string")
+        object.__setattr__(self, "tool_max_sensitivity", MappingProxyType(dict(self.tool_max_sensitivity)))
+
+    def validate(self, order: Mapping[str, int]) -> None:
+        if any(label not in order for label in (
+            *self.tool_max_sensitivity.values(), self.response_max_sensitivity,
+        )):
+            raise ConfigError("sink_policy contains an unknown sensitivity ceiling")
+
+    def require(self, *, tool: str | None, labels: tuple[str, ...], order: Mapping[str, int]) -> None:
+        """No unknown-label fallback, advisory bypass, or lowering by callers."""
+        ceiling = self.response_max_sensitivity if tool is None else self.tool_max_sensitivity.get(tool)
+        sink = "response" if tool is None else "tool"
+        if (ceiling is None or ceiling not in order or not labels or any(label not in order for label in labels)
+                or any(order[label] > order[ceiling] for label in labels)):
+            raise PolicyDeny(f"sink_policy:{sink}_denied")