diff --git a/CHANGELOG.md b/CHANGELOG.md index dc9a94cc..094171f9 100644 --- a/CHANGELOG.md +++ b/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, diff --git a/LIMITATIONS.md b/LIMITATIONS.md index 2c4759cd..834cbdf9 100644 --- a/LIMITATIONS.md +++ b/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** diff --git a/docs/configuration.md b/docs/configuration.md index a7be0b69..1a355081 100644 --- a/docs/configuration.md +++ b/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. diff --git a/docs/spec/sink-policy.md b/docs/spec/sink-policy.md new file mode 100644 index 00000000..3f51348b --- /dev/null +++ b/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. diff --git a/mkdocs.yml b/mkdocs.yml index 98a99dfc..40610021 100644 --- a/mkdocs.yml +++ b/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 diff --git a/src/cmcp_runtime/config.py b/src/cmcp_runtime/config.py index 1bc9a345..8bd92cb6 100644 --- a/src/cmcp_runtime/config.py +++ b/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, diff --git a/src/cmcp_runtime/mcp/proxy.py b/src/cmcp_runtime/mcp/proxy.py index 11a4b324..4775949b 100644 --- a/src/cmcp_runtime/mcp/proxy.py +++ b/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 ) diff --git a/src/cmcp_runtime/mcp/stdio.py b/src/cmcp_runtime/mcp/stdio.py index 379d5822..344e156e 100644 --- a/src/cmcp_runtime/mcp/stdio.py +++ b/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( diff --git a/src/cmcp_runtime/sink_policy.py b/src/cmcp_runtime/sink_policy.py new file mode 100644 index 00000000..a36d73b6 --- /dev/null +++ b/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") diff --git a/tests/unit/test_sink_policy.py b/tests/unit/test_sink_policy.py new file mode 100644 index 00000000..551aba12 --- /dev/null +++ b/tests/unit/test_sink_policy.py @@ -0,0 +1,187 @@ +"""Sink decisions exercise real proxy dispatch and response paths, not scanners.""" + +from dataclasses import replace +from unittest.mock import AsyncMock, MagicMock + +import pytest +import yaml + +from cmcp_runtime.audit.chain import AuditChain +from cmcp_runtime.config import Config, EnforcementMode, load_config +from cmcp_runtime.errors import ConfigError, UpstreamToolError +from cmcp_runtime.mcp.proxy import CMCPProxy +from cmcp_runtime.mcp.stdio import StdioServer, StdioSpawn +from cmcp_runtime.policy.evaluator import PolicyEvaluator +from cmcp_runtime.session.state import SessionState +from cmcp_runtime.sink_policy import SinkPolicy +from tests.unit.test_egress_policy import _make_bundle, _make_catalog + + +def proxy_for(*, tool_cap="public", response_cap="public", session_label="public", + catalog_label="public", mode=EnforcementMode.ENFORCING, enabled=True, caps=None): + config = Config(sink_policy=SinkPolicy(caps if caps is not None else {"test.tool": tool_cap}, response_cap) + if enabled else None) + config.attestation.enforcement_mode = mode + catalog = _make_catalog("test.tool", "public.tool") + catalog.entries["test.tool"] = replace(catalog.entries["test.tool"], sensitivity_level=catalog_label) + session = SessionState("sink-test", max_sensitivity=session_label) + chain = AuditChain(session.session_id) + # Actual Cedar evaluator with a permissive bundle and actual response scanner. + proxy = CMCPProxy(catalog, PolicyEvaluator(_make_bundle(), config), session, chain, config) + proxy._advertised_tools = AsyncMock(return_value=None) + proxy._forward_to_upstream = AsyncMock(return_value="A harmless looking derived summary") + return proxy, session, chain, config + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mode", list(EnforcementMode)) +@pytest.mark.parametrize("source", ["session", "catalog", "declared"]) +async def test_high_label_denies_before_tool_dispatch_in_all_modes(mode, source): + proxy, _, chain, _ = proxy_for(mode=mode, session_label="confidential" if source == "session" else "public", + catalog_label="confidential" if source == "catalog" else "public") + result = await proxy.call_tool("c1", "test.tool", {"summary": "Looks safe"}, + declared_data_class="confidential" if source == "declared" else "public") + assert not result.allowed + assert result.deny_reason == "sink_policy:tool_denied" + proxy._advertised_tools.assert_not_awaited() + proxy._forward_to_upstream.assert_not_awaited() + assert any(e.policy_rule_matched == "sink_policy:tool_denied" for e in chain.entries) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mode", list(EnforcementMode)) +async def test_clean_derived_response_cannot_leave_lower_clearance_sink(mode): + proxy, session, chain, _ = proxy_for(tool_cap="confidential", catalog_label="confidential", mode=mode) + result = await proxy.call_tool("c1", "test.tool", {}) + proxy._forward_to_upstream.assert_awaited_once() + assert not result.allowed and result.response is None + assert result.deny_reason == "sink_policy:response_denied" + assert session.max_sensitivity == "confidential" + assert any(e.entry_type == "egress_denied" for e in chain.entries) + + +@pytest.mark.asyncio +async def test_positive_protected_tool_and_response(): + proxy, _, _, _ = proxy_for(tool_cap="confidential", response_cap="confidential", session_label="confidential") + result = await proxy.call_tool("c1", "test.tool", {}) + assert result.allowed and result.response == "A harmless looking derived summary" + + +@pytest.mark.asyncio +async def test_disabled_policy_preserves_legacy_behavior(): + proxy, _, _, _ = proxy_for(session_label="confidential", enabled=False) + assert (await proxy.call_tool("c1", "test.tool", {})).allowed + + +@pytest.mark.asyncio +async def test_unknown_declared_label_fails_closed(): + proxy, _, _, _ = proxy_for() + result = await proxy.call_tool("c1", "test.tool", {}, declared_data_class="future-secret") + assert not result.allowed + proxy._forward_to_upstream.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_unlisted_tool_fails_closed_and_cannot_disable_captured_policy(): + proxy, _, _, config = proxy_for(caps={}) + config.sink_policy = None + assert not (await proxy.call_tool("c1", "test.tool", {})).allowed + proxy._forward_to_upstream.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_private_read_then_derived_public_write_is_denied(): + proxy, session, _, _ = proxy_for(catalog_label="confidential", response_cap="confidential", + caps={"test.tool": "confidential", "public.tool": "public"}) + assert (await proxy.call_tool("read", "test.tool", {})).allowed + assert session.max_sensitivity == "confidential" + proxy._forward_to_upstream.reset_mock() + result = await proxy.call_tool("write", "public.tool", {"summary": "Apparently harmless", "sink_policy": None}, + declared_data_class="public") + assert not result.allowed and result.deny_reason == "sink_policy:tool_denied" + proxy._forward_to_upstream.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_sensitivity_raised_during_discovery_is_rechecked_before_dispatch(): + proxy, session, _, _ = proxy_for() + async def discover(*args, **kwargs): + session.update_from_inspection("parallel-read", ["confidential"], + injection_detected=False, response_allowed=True) + proxy._advertised_tools = AsyncMock(side_effect=discover) + result = await proxy.call_tool("c1", "test.tool", {}) + assert not result.allowed and result.deny_reason == "sink_policy:tool_denied" + proxy._forward_to_upstream.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_reset_during_response_does_not_lower_inflight_floor(): + proxy, session, _, _ = proxy_for(tool_cap="confidential", session_label="confidential") + async def return_after_reset(*args, **kwargs): + session.reset(reason="operator reset", authorized_by="operator") + return "A harmless looking derived summary" + proxy._forward_to_upstream = AsyncMock(side_effect=return_after_reset) + result = await proxy.call_tool("c1", "test.tool", {}) + assert not result.allowed and result.response is None + + +@pytest.mark.asyncio +async def test_upstream_error_plaintext_never_reaches_gateway_log(caplog): + proxy, _, chain, _ = proxy_for() + secret = "private-contract-canary-471" + proxy._forward_to_upstream.side_effect = UpstreamToolError(secret) + result = await proxy.call_tool("c1", "test.tool", {}) + assert not result.allowed + assert secret not in caplog.text + assert secret not in str(result) + assert all(secret not in str(entry) for entry in chain.entries) + + +@pytest.mark.asyncio +async def test_child_stderr_suppressed_at_log_sink(caplog): + server = StdioServer(StdioSpawn("unused"), log_stderr=False) + server._proc = MagicMock() + server._proc.stderr.read = AsyncMock(return_value=b"private-stderr-canary") + await server._collect_stderr() + assert server.stderr_bytes == len(b"private-stderr-canary") + assert "private-stderr-canary" not in caplog.text + assert "stderr suppressed" in caplog.text + + +@pytest.mark.asyncio +async def test_proxy_wires_stderr_control_into_real_spawn_seam(monkeypatch): + proxy, _, _, _ = proxy_for() + entry = proxy._catalog.entries["test.tool"] + entry.server = replace(entry.server, transport="stdio", spawn=StdioSpawn("unused")) + monkeypatch.setattr(StdioServer, "start", AsyncMock()) + server = await proxy._stdio_for(entry) + assert server._log_stderr is False + + +def test_policy_copies_caller_mapping(): + caps = {"test.tool": "public"} + policy = SinkPolicy(caps, "public") + caps["test.tool"] = "confidential" + assert policy.tool_max_sensitivity["test.tool"] == "public" + with pytest.raises(TypeError): + policy.tool_max_sensitivity["test.tool"] = "confidential" + + +@pytest.mark.parametrize("value", [None, {}, {"tool_max_sensitivity": {}}, + {"tool_max_sensitivity": {}, "response_max_sensitivity": "typo"}, + {"tool_max_sensitivity": {"x": "typo"}, "response_max_sensitivity": "public"}, + {"tool_max_sensitivity": [], "response_max_sensitivity": "public"}, + {"tool_max_sensitivity": {}, "response_max_sensitivity": "public", "advisory": True}, +]) +def test_malformed_sink_config_is_rejected(tmp_path, value): + path = tmp_path / "config.yaml" + path.write_text(yaml.safe_dump({"sink_policy": value})) + with pytest.raises(ConfigError): + load_config(str(path)) + + +def test_custom_label_is_validated_and_ranked(tmp_path): + path = tmp_path / "config.yaml" + path.write_text(yaml.safe_dump({"sensitivity": {"vocabulary": {"top_secret": 4}}, + "sink_policy": {"tool_max_sensitivity": {"x": "top_secret"}, "response_max_sensitivity": "top_secret"}})) + assert load_config(str(path)).sink_policy.response_max_sensitivity == "top_secret"