diff --git a/docs/spec/execution-correlation.md b/docs/spec/execution-correlation.md index a38653e..da11178 100644 --- a/docs/spec/execution-correlation.md +++ b/docs/spec/execution-correlation.md @@ -177,3 +177,33 @@ coverage. This design does not invent version negotiation behavior. **TRACE records remain joinable offline.** The audit entry carries the join key. The Claim binds the chain and does not duplicate its identifiers. + +## Non-operational foundation (#606) + +The standalone `ExecutionRegistry` provides reservation, classification and +recovery primitives for future integration. Its opaque binding inputs are test +inputs, not an adopted action-identity contract. The gateway does not construct, +recover, admit through, or finalize this registry. There is no runtime opt-in. + +The typed audit field and ingress validation are present. A valid supplied +`execution_id` is refused with `execution_correlation_unavailable` before +health/catalog checks, upstream discovery, or tool invocation, and the refusal retains +the request hash, call identity, workflow, tool name, and asserted identifier. +Malformed values, including explicit null, are refused with +`execution_invalid_execution_id` and are not copied into audit identity. Only +omission preserves the legacy call path. +No same-operation, replay-enforcement or exactly-once guarantee is provided by +this foundation in the running gateway. + +Activation requires both: + +1. An adopted action preimage and canonicalization contract (#588), integrated + with admission and its interoperability and semantic-admissibility vectors. +2. Terminal execution state and terminal audit evidence committed consistently + at one durable transaction boundary, with crash/recovery evidence meeting the + agreed requirement. Independent audit and registry commits are not sufficient. + +The standalone registry tests exercise local storage primitives only. They do +not establish integrated audit consistency. #565 remains open until integration +and its acceptance evidence are complete; completing #588 alone cannot activate +this feature. diff --git a/src/cmcp_runtime/audit/chain.py b/src/cmcp_runtime/audit/chain.py index 1e0ca28..1be15c3 100644 --- a/src/cmcp_runtime/audit/chain.py +++ b/src/cmcp_runtime/audit/chain.py @@ -81,6 +81,13 @@ class AuditEntry: # _max_sensitivity can only return the higher of the two. None when no # declaration was made, the transcript falls back to the catalog value. effective_data_class: str | None = None + # #565: validated session-independent correlation key for one executable unit. + # A typed field, always serialized (null when the caller supplied none), never + # a `detail` key. A present value passed ingress validation; operational + # admission is a separate integration step. Null means no valid execution + # identifier was retained; inspect the refusal rule to distinguish omission + # from invalid input. The TRACE Claim does not enumerate these values. + execution_id: str | None = None entry_hash: str = field(default="") # computed after construction def _canonical_body(self) -> bytes: @@ -219,6 +226,7 @@ def append( workflow_id: str | None = None, external_execution_evidence: dict[str, str] | None = None, effective_data_class: str | None = None, + execution_id: str | None = None, ) -> AuditEntry: prev_hash = self._entries[-1].entry_hash if self._entries else "genesis" now = datetime.now(tz=UTC) @@ -248,6 +256,7 @@ def append( workflow_id=workflow_id, external_execution_evidence=external_execution_evidence, effective_data_class=effective_data_class, + execution_id=execution_id, prev_entry_hash=prev_hash, ) entry.entry_hash = entry.compute_hash() diff --git a/src/cmcp_runtime/execution/__init__.py b/src/cmcp_runtime/execution/__init__.py new file mode 100644 index 0000000..ee23706 --- /dev/null +++ b/src/cmcp_runtime/execution/__init__.py @@ -0,0 +1,19 @@ +"""Non-operational execution-state foundation; no gateway integration (#565).""" + +from cmcp_runtime.execution.registry import ( + Admission, + AdmissionStatus, + Disposition, + ExecutionRegistry, + ExecutionStateError, + valid_execution_id, +) + +__all__ = [ + "Admission", + "AdmissionStatus", + "Disposition", + "ExecutionRegistry", + "ExecutionStateError", + "valid_execution_id", +] diff --git a/src/cmcp_runtime/execution/registry.py b/src/cmcp_runtime/execution/registry.py new file mode 100644 index 0000000..ef3c79b --- /dev/null +++ b/src/cmcp_runtime/execution/registry.py @@ -0,0 +1,259 @@ +"""Standalone execution-state foundation for #565; not wired into the gateway. + +Reservations are keyed by (agent_identity, execution_id). Admission atomically +stores or compares an opaque binding. Completed and outcome_unknown rows are +terminal, while recovery marks abandoned in_flight rows outcome_unknown. +Identifiers never expire and repeat reservations never authorize invocation. + +These are storage primitives, not production execution enforcement. The binding +preimage is unresolved in #588. finalize() commits only the registry row: it +cannot atomically publish a terminal audit entry. Integrating this module requires +an adopted binding contract and a shared durable terminal/audit boundary, with +crash and recovery tests. The gateway refuses supplied execution IDs until both +requirements are met; there is no runtime activation option. +""" + +from __future__ import annotations + +import logging +import re +import sqlite3 +import threading +from dataclasses import dataclass +from datetime import UTC, datetime +from enum import StrEnum +from pathlib import Path + +logger = logging.getLogger(__name__) + +_IN_FLIGHT = "in_flight" + +# #574 says the gateway validates execution_id, not just scopes it. This is the +# durable primary key: an empty string or a multi-megabyte value must never +# reach it. 1 to 200 printable, non-space ASCII characters covers UUIDs, ULIDs, +# and URN-style identifiers without opening the key to control characters or +# unbounded length. +_EXECUTION_ID_RE = re.compile(r"[\x21-\x7e]{1,200}\Z") + + +def valid_execution_id(value: str) -> bool: + """True when `value` is well-formed enough to be a durable correlation key.""" + return _EXECUTION_ID_RE.match(value) is not None + +_CREATE_TABLE = """ +CREATE TABLE IF NOT EXISTS executions ( + agent_identity TEXT NOT NULL, + execution_id TEXT NOT NULL, + action_binding TEXT NOT NULL, + state TEXT NOT NULL, + call_id TEXT, + terminal_audit_entry_hash TEXT, + created_utc TEXT NOT NULL, + updated_utc TEXT NOT NULL, + PRIMARY KEY (agent_identity, execution_id) +); +""" + + +class AdmissionStatus(StrEnum): + """Classification of an `admit()` request. Only ADMITTED may invoke upstream.""" + + ADMITTED = "admitted" + REPLAY_IN_FLIGHT = "replay_in_flight" + REPLAY_TERMINAL = "replay_terminal" + REPLAY_OUTCOME_UNKNOWN = "replay_outcome_unknown" + COLLISION_CHANGED_BINDING = "collision_changed_binding" + + +class Disposition(StrEnum): + """Terminal disposition supplied to `finalize()`. Both are non-replayable.""" + + COMPLETED = "completed" + OUTCOME_UNKNOWN = "outcome_unknown" + + +class ExecutionStateError(RuntimeError): + """A caller finalized an execution that was never admitted.""" + + +@dataclass(frozen=True) +class Admission: + """ + Outcome of `admit()`. + + `action_binding` is the opaque digest the caller supplied, echoed back so it + can be recorded on the audit entry whether the request was admitted or + refused. + """ + + status: AdmissionStatus + action_binding: str + + @property + def admitted(self) -> bool: + return self.status is AdmissionStatus.ADMITTED + + @property + def audit_rule(self) -> str: + """Stable `policy_rule_matched` value for a refused admission.""" + return f"execution:{self.status.value}" + + +class ExecutionRegistry: + """Durable owner of execution-correlation state. One instance per process.""" + + def __init__(self, db_path: Path) -> None: + # check_same_thread=False plus one lock: the proxy calls this from async + # handlers and worker threads, and every write is a short serialized + # transaction. busy_timeout covers a second gateway process sharing the + # file behind BEGIN IMMEDIATE's reserved lock. + self._conn = sqlite3.connect(str(db_path), check_same_thread=False) + self._lock = threading.Lock() + self._conn.execute("PRAGMA journal_mode=WAL") + self._conn.execute("PRAGMA synchronous=FULL") + self._conn.execute("PRAGMA busy_timeout=5000") + self._conn.executescript(_CREATE_TABLE) + self._conn.commit() + logger.info("Execution registry opened: path=%s", db_path) + + def recover(self) -> int: + """ + Fail closed over crash ambiguity. Every execution still `in_flight` is + moved to `outcome_unknown`: the gateway stopped between reserving the + identifier and recording a terminal, so the external effect is unknown + and the identifier must never admit another invocation. Returns the + number of executions sealed this way. Call once at startup. + """ + now = _now() + with self._lock: + self._conn.execute("BEGIN IMMEDIATE") + try: + cur = self._conn.execute( + "UPDATE executions SET state='outcome_unknown', updated_utc=? WHERE state=?", + (now, _IN_FLIGHT), + ) + self._conn.commit() + except BaseException: + self._conn.rollback() + raise + if cur.rowcount: + logger.warning( + "Execution registry recovery sealed %d in-flight execution(s) as " + "outcome_unknown", + cur.rowcount, + ) + return cur.rowcount + + def admit( + self, + *, + agent_identity: str, + execution_id: str, + action_binding: str, + call_id: str, + ) -> Admission: + """ + Atomically reserve (agent_identity, execution_id, action_binding) before + upstream invocation, or classify why the request cannot proceed. + + `action_binding` is an opaque digest string produced by the caller. A key + that already exists is never rewritten: an identical binding is a replay + classified by the stored state, a different binding is a collision. Both + are refused here, before any upstream effect. + """ + now = _now() + with self._lock: + self._conn.execute("BEGIN IMMEDIATE") + try: + row = self._conn.execute( + "SELECT state, action_binding FROM executions " + "WHERE agent_identity=? AND execution_id=?", + (agent_identity, execution_id), + ).fetchone() + if row is None: + self._conn.execute( + "INSERT INTO executions (agent_identity, execution_id, " + "action_binding, state, call_id, created_utc, updated_utc) " + "VALUES (?, ?, ?, ?, ?, ?, ?)", + (agent_identity, execution_id, action_binding, _IN_FLIGHT, call_id, now, now), + ) + self._conn.commit() + return Admission(AdmissionStatus.ADMITTED, action_binding) + self._conn.rollback() + except BaseException: + self._conn.rollback() + raise + + return Admission(_classify_existing(row[0], row[1], action_binding), action_binding) + + def finalize( + self, + *, + agent_identity: str, + execution_id: str, + disposition: Disposition, + terminal_audit_entry_hash: str, + ) -> None: + """ + Record the terminal outcome for an admitted execution and pin it to the + audit entry that carries the same fact. Legal only from `in_flight`; a + second call is a no-op so a retry of the caller's terminal path cannot + raise after the effect already happened. Raises ExecutionStateError if + the key was never admitted. + """ + now = _now() + with self._lock: + self._conn.execute("BEGIN IMMEDIATE") + try: + row = self._conn.execute( + "SELECT state FROM executions WHERE agent_identity=? AND execution_id=?", + (agent_identity, execution_id), + ).fetchone() + if row is None: + self._conn.rollback() + raise ExecutionStateError( + f"finalize before admit: execution_id={execution_id!r}" + ) + if row[0] != _IN_FLIGHT: + self._conn.rollback() + logger.warning( + "Execution %r already terminal (%s); finalize(%s) ignored", + execution_id, + row[0], + disposition.value, + ) + return + self._conn.execute( + "UPDATE executions SET state=?, terminal_audit_entry_hash=?, " + "updated_utc=? WHERE agent_identity=? AND execution_id=?", + ( + disposition.value, + terminal_audit_entry_hash, + now, + agent_identity, + execution_id, + ), + ) + self._conn.commit() + except BaseException: + self._conn.rollback() + raise + + def close(self) -> None: + with self._lock: + self._conn.close() + + +def _classify_existing(state: str, stored_binding: str, presented_binding: str) -> AdmissionStatus: + """Map an existing row to a refusal status. Binding mismatch outranks state.""" + if stored_binding != presented_binding: + return AdmissionStatus.COLLISION_CHANGED_BINDING + if state == _IN_FLIGHT: + return AdmissionStatus.REPLAY_IN_FLIGHT + if state == "outcome_unknown": + return AdmissionStatus.REPLAY_OUTCOME_UNKNOWN + return AdmissionStatus.REPLAY_TERMINAL + + +def _now() -> str: + return datetime.now(tz=UTC).isoformat() diff --git a/src/cmcp_runtime/mcp/proxy.py b/src/cmcp_runtime/mcp/proxy.py index 0c6bccd..9cb58b1 100644 --- a/src/cmcp_runtime/mcp/proxy.py +++ b/src/cmcp_runtime/mcp/proxy.py @@ -32,6 +32,7 @@ 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.execution import valid_execution_id from cmcp_runtime.mcp import tls_pinning from cmcp_runtime.mcp.stdio import StdioServer from cmcp_runtime.mcp.streamable_http import ( @@ -97,6 +98,8 @@ class _CallFinalizationState: server_identity: str | None = None external_execution_evidence: dict[str, str] | None = None terminal_entry_id: str | None = None + # Validated caller identity retained on the unavailable-feature refusal. + execution_id: str | None = None @property def terminal_disposition(self) -> str: @@ -202,11 +205,12 @@ def _extract_external_execution_evidence(response_text: str) -> dict[str, str] | class CMCPProxy: """ Enforces every tool call through the cMCP runtime gateway: - 1. Checked against the attested catalog - 2. Evaluated by the Cedar PolicyEvaluator - 3. Checked for rate limits, dangerous parameters, and unsafe responses - 4. Logged to the TEE-sealed AuditChain - 5. Session state updated via inspection handoff + 1. Execution-correlation requests validated or refused before discovery + 2. Checked against the attested catalog + 3. Evaluated by the Cedar PolicyEvaluator + 4. Checked for rate limits, dangerous parameters, and unsafe responses + 5. Logged to the TEE-sealed AuditChain + 6. Session state updated via inspection handoff One CMCPProxy instance per gateway session. """ @@ -765,10 +769,103 @@ def _append_call_terminal( """Persist one terminal for this invocation, independent of call_id reuse.""" if finalization.terminal_entry_id is not None: raise RuntimeError("terminal audit entry already persisted for this invocation") + # #565: every terminal for a correlated call carries its execution_id. + # Set from one place so no per-branch call site has to remember it. + fields.setdefault("execution_id", finalization.execution_id) entry = self._audit.append(entry_type, **fields) # type: ignore[arg-type] finalization.terminal_entry_id = entry.entry_id finalization.effect_boundary_state = _EffectBoundaryState.TERMINAL_DURABLE + def _check_execution_available( + self, + finalization: _CallFinalizationState, + *, + execution_id: str | None, + call_id: str, + tool_name: str, + entry: CatalogEntry | None, + request_payload_hash: str, + sensitivity_before: str, + workflow_id: str | None, + t0: float, + called_at: datetime, + ) -> CallResult | None: + """Refuse requested execution correlation until its contract is implemented. + + Omission preserves legacy calls. There is no runtime opt-in: action + binding and atomic terminal/audit persistence must both land first. + """ + if execution_id is None: + return None + if not valid_execution_id(execution_id): + return self._refuse_execution( + finalization, entry, call_id, tool_name, request_payload_hash, + sensitivity_before, workflow_id, t0, called_at, + rule="execution:invalid_execution_id", + deny_reason="execution_invalid_execution_id", + ) + finalization.execution_id = execution_id + return self._refuse_execution( + finalization, entry, call_id, tool_name, request_payload_hash, + sensitivity_before, workflow_id, t0, called_at, + rule="execution:unavailable", + deny_reason="execution_correlation_unavailable", + ) + + def _refuse_execution( + self, + finalization: _CallFinalizationState, + entry: CatalogEntry | None, + call_id: str, + tool_name: str, + request_payload_hash: str, + sensitivity_before: str, + workflow_id: str | None, + t0: float, + called_at: datetime, + *, + rule: str, + deny_reason: str, + ) -> CallResult: + """Audit and return the deny for an execution that must not reach upstream.""" + import time + + self._append_call_terminal( + finalization, + "tool_call", + call_id=call_id, + tool_name=tool_name, + server_identity=entry.server.url if entry is not None else None, + policy_decision="deny", + policy_rule_matched=rule, + 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={"execution": "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=deny_reason, + latency_us=int(elapsed_ms * 1000), + audit_entry_hash=self._audit.chain_tip, + ) + def _finalize_unexpected_call_failure( self, finalization: _CallFinalizationState, @@ -825,6 +922,7 @@ async def call_tool( arguments: dict[str, Any], workflow_id: str | None = None, declared_data_class: str | None = None, + execution_id: str | None = None, ) -> CallResult: """Run one call and guarantee one terminal on failure or cancellation.""" finalization = _CallFinalizationState() @@ -835,6 +933,7 @@ async def call_tool( arguments, workflow_id, declared_data_class, + execution_id=execution_id, _finalization=finalization, ) except BaseException as exc: @@ -857,19 +956,22 @@ async def _call_tool_impl( workflow_id: str | None = None, declared_data_class: str | None = None, *, + execution_id: str | None = None, _finalization: _CallFinalizationState, ) -> CallResult: """ Execute one MCP tool call through the full enforcement pipeline. Pipeline: - 1. Catalog lookup (fast-path deny if not in catalog) - 2. Cedar policy evaluation - 3. cMCP runtime enforcement (sanitization, rate limit, scan) - 4. Forward to upstream - 5. Audit chain write - 6. Session state update - 7. Call log record + suspicious-sequence check + 1. Request serialization and execution-correlation validation/refusal + 2. Health check + 3. Catalog lookup (fast-path deny if not in catalog) + 4. Cedar policy evaluation + 5. cMCP runtime enforcement (sanitization, rate limit, scan) + 6. Forward to upstream + 7. Audit chain write + 8. Session state update + 9. Call log record + suspicious-sequence check declared_data_class (#479 piece 2): an optional class the caller declares for this specific call via _cmcp.data_class, raising this call's effective @@ -888,7 +990,38 @@ class above the tool's catalogued sensitivity_level. It can never lower sensitivity_before = self._session.max_sensitivity would_have_denied = False - # Step 0: health check (attestation staleness, catalog drift) + # Step 0: serialize the request before any early refusal so the audit + # entry can retain a stable request hash even when no catalog entry is + # available. JSON-RPC ingress already guarantees JSON-compatible args; + # direct callers still get the normal fault-finalization path if this + # serialization fails. + _finalization.failure_stage = "request_serialization" + _payload_bytes = json.dumps(arguments, sort_keys=True, separators=(",", ":")).encode() + request_payload_hash = f"sha256:{hashlib.sha256(_payload_bytes).hexdigest()}" + _finalization.request_payload_hash = request_payload_hash + + # Step 1: execution correlation is unavailable until binding and + # persistence contracts are complete; supplied IDs must be refused + # before health/catalog checks or any upstream discovery. The entry is + # not known yet, so the refusal carries request context and a null + # server. Omission returns immediately and preserves the normal path. + _finalization.failure_stage = "execution_admission" + execution_refusal = self._check_execution_available( + _finalization, + execution_id=execution_id, + call_id=call_id, + tool_name=tool_name, + entry=None, + request_payload_hash=request_payload_hash, + sensitivity_before=sensitivity_before, + workflow_id=workflow_id, + t0=t0, + called_at=called_at, + ) + if execution_refusal is not None: + return execution_refusal + + # Step 2: health check (attestation staleness, catalog drift) _finalization.failure_stage = "health_check" unhealthy_reason = self._check_health() if unhealthy_reason is not None: @@ -903,12 +1036,7 @@ class above the tool's catalogued sensitivity_level. It can never lower audit_entry_hash=self._audit.chain_tip, ) - _finalization.failure_stage = "request_serialization" - _payload_bytes = json.dumps(arguments, sort_keys=True, separators=(",", ":")).encode() - request_payload_hash = f"sha256:{hashlib.sha256(_payload_bytes).hexdigest()}" - _finalization.request_payload_hash = request_payload_hash - - # Step 1: catalog lookup + # Step 3: catalog lookup _finalization.failure_stage = "catalog_lookup" entry = self._catalog.lookup(tool_name) if entry is None: @@ -951,7 +1079,7 @@ class above the tool's catalogued sensitivity_level. It can never lower _finalization.server_identity = entry.server.url - # Step 1a (#521): does this server still offer what we approved? First + # 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 # to know which server to ask, and before the policy decision because a @@ -993,7 +1121,7 @@ class above the tool's catalogued sensitivity_level. It can never lower else None ) - # Step 1b: break-glass warning - log and audit every call via an exception entry + # Step 3b: 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", @@ -1012,7 +1140,7 @@ class above the tool's catalogued sensitivity_level. It can never lower workflow_id=workflow_id, ) - # Step 2: Cedar policy evaluation + # Step 4: Cedar policy evaluation _finalization.failure_stage = "policy_evaluation" cedar_context = self._build_cedar_context( tool_name, arguments, workflow_id, effective_data_class @@ -1082,7 +1210,7 @@ class above the tool's catalogued sensitivity_level. It can never lower ) raise - # Step 3a: native pre-call interception: per-agent rate limiting, + # Step 5a: native pre-call interception: per-agent rate limiting, # parameter sanitization, and allow/deny. Fail closed on internal errors. _finalization.failure_stage = "ingress_gateway" agt_allowed, agt_reason = self._mcp_gateway.intercept_tool_call( @@ -1128,7 +1256,7 @@ class above the tool's catalogued sensitivity_level. It can never lower audit_entry_hash=self._audit.chain_tip, ) - # Step 3b: forward to the attested upstream MCP server. + # Step 5b: forward to the attested upstream MCP server. _finalization.failure_stage = "upstream_invocation" try: response_text = await self._forward_to_upstream( diff --git a/src/cmcp_runtime/mcp/server.py b/src/cmcp_runtime/mcp/server.py index 16df4cf..6a18806 100644 --- a/src/cmcp_runtime/mcp/server.py +++ b/src/cmcp_runtime/mcp/server.py @@ -609,6 +609,15 @@ async def _handle_tool_call(self, rpc_id: Any, params: dict[str, Any]) -> Respon cmcp_params = {} raw_workflow = cmcp_params.get("workflow_id") workflow_id: str | None = raw_workflow if isinstance(raw_workflow, str) else None + # #565: validated session-independent execution identity, supplied beside + # workflow_id and independent of it. Only an omitted ID is absent. + # Map present non-strings to an invalid empty ID so the proxy uses its + # audited refusal path instead of silently bypassing correlation. + raw_execution = cmcp_params.get("execution_id") + execution_id: str | None = ( + raw_execution if isinstance(raw_execution, str) + else "" if "execution_id" in cmcp_params else None + ) # #479 piece 2: the caller may declare a class for this specific call. raw_data_class = cmcp_params.get("data_class") declared_data_class: str | None = ( @@ -622,6 +631,7 @@ async def _handle_tool_call(self, rpc_id: Any, params: dict[str, Any]) -> Respon arguments, workflow_id=workflow_id, declared_data_class=declared_data_class, + execution_id=execution_id, ) except Exception as exc: logger.error("TEE_FAULT during call_tool: call_id=%s error=%s", call_id, exc) diff --git a/tests/unit/test_cli_wiring.py b/tests/unit/test_cli_wiring.py index 095ab91..2609321 100644 --- a/tests/unit/test_cli_wiring.py +++ b/tests/unit/test_cli_wiring.py @@ -82,6 +82,16 @@ def test_health_exempt_from_auth(ctx): assert client.get("/health").status_code != 401 +def test_production_build_does_not_open_execution_registry(ctx, monkeypatch): + """The unfinished state store must stay disconnected from production startup.""" + def unexpected_registry(*args, **kwargs): + pytest.fail("production constructed the non-operational execution registry") + + monkeypatch.setattr("cmcp_runtime.execution.ExecutionRegistry", unexpected_registry) + monkeypatch.setattr("cmcp_runtime.execution.registry.ExecutionRegistry", unexpected_registry) + build_server(ctx) + + def test_audit_chain_persists_to_store(ctx, tmp_path): """AUDIT-001: the session_start entry must land in the SQLite DB.""" build_server(ctx) diff --git a/tests/unit/test_execution_correlation.py b/tests/unit/test_execution_correlation.py new file mode 100644 index 0000000..289d26b --- /dev/null +++ b/tests/unit/test_execution_correlation.py @@ -0,0 +1,308 @@ +"""Registry-level gates for durable execution correlation (issue #565).""" + +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +import pytest + +from cmcp_runtime.execution import ( + AdmissionStatus, + Disposition, + ExecutionStateError, + valid_execution_id, +) +from cmcp_runtime.execution.registry import ExecutionRegistry + + +@pytest.mark.parametrize( + "value,ok", + [ + ("exec-1", True), + ("exec-café", False), # non-ASCII + ("a" * 200, True), + ("a" * 201, False), + ("", False), + ("has space", False), + ("line\nbreak", False), + ("tab\tsep", False), + ("urn:example:execution:42", True), + ], +) +def test_valid_execution_id_bounds(value, ok): + assert valid_execution_id(value) is ok + +AGENT = "spiffe://example.org/agent-a" +OTHER_AGENT = "spiffe://example.org/agent-b" +BINDING_A = "sha256:aaaa" +BINDING_B = "sha256:bbbb" + + +def _registry(tmp_path: Path, name: str = "executions.db") -> ExecutionRegistry: + return ExecutionRegistry(tmp_path / name) + + +def _admit(reg: ExecutionRegistry, *, agent: str = AGENT, execution_id: str = "exec-1", + binding: str = BINDING_A, call_id: str = "call-1"): + return reg.admit( + agent_identity=agent, + execution_id=execution_id, + action_binding=binding, + call_id=call_id, + ) + + +def test_first_admit_reserves_and_returns_binding(tmp_path): + reg = _registry(tmp_path) + result = _admit(reg) + assert result.status is AdmissionStatus.ADMITTED + assert result.admitted + assert result.action_binding == BINDING_A + + +def test_same_identity_same_binding_is_replay_not_a_second_reservation(tmp_path): + reg = _registry(tmp_path) + first = _admit(reg, call_id="call-1") + second = _admit(reg, call_id="call-2") + assert first.admitted + assert not second.admitted + assert second.status is AdmissionStatus.REPLAY_IN_FLIGHT + # The binding still resolves so the refusal can be audited under it. + assert second.action_binding == first.action_binding + + +def test_changed_binding_same_execution_id_is_a_collision(tmp_path): + reg = _registry(tmp_path) + _admit(reg, binding=BINDING_A) + collided = _admit(reg, binding=BINDING_B) + assert collided.status is AdmissionStatus.COLLISION_CHANGED_BINDING + assert not collided.admitted + + +def test_different_identities_do_not_collide(tmp_path): + reg = _registry(tmp_path) + a = _admit(reg, agent=AGENT) + b = _admit(reg, agent=OTHER_AGENT) + assert a.admitted and b.admitted + + +def test_terminal_completed_is_not_replayable(tmp_path): + reg = _registry(tmp_path) + first = _admit(reg) + reg.finalize( + agent_identity=AGENT, execution_id="exec-1", + disposition=Disposition.COMPLETED, terminal_audit_entry_hash="hash-1", + ) + replay = _admit(reg, call_id="call-2") + assert replay.status is AdmissionStatus.REPLAY_TERMINAL + assert not replay.admitted + assert replay.action_binding == first.action_binding + + +def test_outcome_unknown_is_not_replayable(tmp_path): + reg = _registry(tmp_path) + _admit(reg) + reg.finalize( + agent_identity=AGENT, execution_id="exec-1", + disposition=Disposition.OUTCOME_UNKNOWN, terminal_audit_entry_hash="hash-1", + ) + replay = _admit(reg, call_id="call-2") + assert replay.status is AdmissionStatus.REPLAY_OUTCOME_UNKNOWN + assert not replay.admitted + + +def test_finalize_before_admit_raises(tmp_path): + reg = _registry(tmp_path) + with pytest.raises(ExecutionStateError): + reg.finalize( + agent_identity=AGENT, execution_id="ghost", + disposition=Disposition.COMPLETED, terminal_audit_entry_hash="h", + ) + + +def test_double_finalize_is_ignored(tmp_path): + reg = _registry(tmp_path) + _admit(reg) + reg.finalize( + agent_identity=AGENT, execution_id="exec-1", + disposition=Disposition.COMPLETED, terminal_audit_entry_hash="hash-1", + ) + # A retry of the caller's terminal path must not raise after the effect. + reg.finalize( + agent_identity=AGENT, execution_id="exec-1", + disposition=Disposition.OUTCOME_UNKNOWN, terminal_audit_entry_hash="hash-2", + ) + row = reg._conn.execute( + "SELECT state, terminal_audit_entry_hash FROM executions " + "WHERE agent_identity=? AND execution_id=?", + (AGENT, "exec-1"), + ).fetchone() + assert row == ("completed", "hash-1") + + +def test_restart_recovery_seals_in_flight_as_outcome_unknown(tmp_path): + reg = _registry(tmp_path) + _admit(reg, execution_id="exec-live") + _admit(reg, execution_id="exec-done", call_id="c2") + reg.finalize( + agent_identity=AGENT, execution_id="exec-done", + disposition=Disposition.COMPLETED, terminal_audit_entry_hash="h", + ) + reg.close() + + restarted = _registry(tmp_path) + sealed = restarted.recover() + assert sealed == 1 + replay_live = restarted.admit( + agent_identity=AGENT, execution_id="exec-live", + action_binding=BINDING_A, call_id="c3", + ) + assert replay_live.status is AdmissionStatus.REPLAY_OUTCOME_UNKNOWN + assert not replay_live.admitted + + +class _InterceptConn: + """Forwards to a real sqlite3.Connection, letting a test intervene per statement.""" + + def __init__(self, real: sqlite3.Connection, on_execute) -> None: + self._real = real + self._on_execute = on_execute + + def execute(self, sql: str, *args): + self._on_execute(self._real, sql, args) + return self._real.execute(sql, *args) + + def __getattr__(self, name): + return getattr(self._real, name) + + +def test_injected_persistence_failure_during_finalize_fails_closed(tmp_path): + reg = _registry(tmp_path) + _admit(reg, execution_id="exec-x") + + def fail_the_update(_real, sql, _args): + if sql.startswith("UPDATE executions SET state="): + raise sqlite3.OperationalError("disk I/O error (injected)") + + reg._conn = _InterceptConn(reg._conn, fail_the_update) + with pytest.raises(sqlite3.OperationalError): + reg.finalize( + agent_identity=AGENT, execution_id="exec-x", + disposition=Disposition.COMPLETED, terminal_audit_entry_hash="h", + ) + reg._conn = reg._conn._real + reg.close() + + # The row is still in_flight on disk; recovery seals it and it never + # admits another invocation. + restarted = _registry(tmp_path) + assert restarted.recover() == 1 + replay = restarted.admit( + agent_identity=AGENT, execution_id="exec-x", + action_binding=BINDING_A, call_id="c9", + ) + assert replay.status is AdmissionStatus.REPLAY_OUTCOME_UNKNOWN + + +def test_injected_persistence_failure_during_recovery_rolls_back(tmp_path): + reg = _registry(tmp_path) + _admit(reg, execution_id="exec-recover") + + def fail_the_recovery_update(_real, sql, _args): + if sql.startswith("UPDATE executions SET state='outcome_unknown'"): + raise sqlite3.OperationalError("disk I/O error (injected)") + + reg._conn = _InterceptConn(reg._conn, fail_the_recovery_update) + with pytest.raises(sqlite3.OperationalError): + reg.recover() + reg._conn = reg._conn._real + + row = reg._conn.execute( + "SELECT state FROM executions WHERE execution_id='exec-recover'" + ).fetchone() + assert row == ("in_flight",) + reg.close() + + +def test_injected_persistence_failure_during_admission_rolls_back(tmp_path): + reg = _registry(tmp_path) + + def fail_the_insert(_real, sql, _args): + if sql.startswith("INSERT INTO executions"): + raise sqlite3.OperationalError("disk I/O error (injected)") + + reg._conn = _InterceptConn(reg._conn, fail_the_insert) + with pytest.raises(sqlite3.OperationalError): + _admit(reg, execution_id="exec-admit") + reg._conn = reg._conn._real + + assert reg._conn.execute("SELECT COUNT(*) FROM executions").fetchone() == (0,) + assert _admit(reg, execution_id="exec-admit").status is AdmissionStatus.ADMITTED + reg.close() + + +def test_concurrent_admits_of_one_key_reserve_exactly_once(tmp_path): + """Two threads racing to admit the same key: exactly one is ADMITTED, the + other is classified as a replay. No double reservation, no exception.""" + import threading + + reg = _registry(tmp_path) + barrier = threading.Barrier(8) + results: list = [] + lock = threading.Lock() + + def worker(i: int) -> None: + barrier.wait() + r = reg.admit( + agent_identity=AGENT, execution_id="exec-hot", + action_binding=BINDING_A, call_id=f"c{i}", + ) + with lock: + results.append(r.status) + + threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert results.count(AdmissionStatus.ADMITTED) == 1 + assert all( + s in (AdmissionStatus.ADMITTED, AdmissionStatus.REPLAY_IN_FLIGHT) for s in results + ) + rows = reg._conn.execute( + "SELECT COUNT(*) FROM executions WHERE execution_id='exec-hot'" + ).fetchone() + assert rows == (1,) + + +def test_separate_registry_connections_serialize_same_key(tmp_path): + """BEGIN IMMEDIATE serializes two registry instances on one database file.""" + import threading + + first = _registry(tmp_path) + second = _registry(tmp_path) + barrier = threading.Barrier(2) + results: list = [] + lock = threading.Lock() + + def worker(reg: ExecutionRegistry, call_id: str) -> None: + barrier.wait() + result = _admit(reg, execution_id="exec-shared", call_id=call_id) + with lock: + results.append(result.status) + + threads = [ + threading.Thread(target=worker, args=(first, "call-a")), + threading.Thread(target=worker, args=(second, "call-b")), + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + assert results.count(AdmissionStatus.ADMITTED) == 1 + assert results.count(AdmissionStatus.REPLAY_IN_FLIGHT) == 1 + first.close() + second.close() diff --git a/tests/unit/test_execution_correlation_call_path.py b/tests/unit/test_execution_correlation_call_path.py new file mode 100644 index 0000000..63baf99 --- /dev/null +++ b/tests/unit/test_execution_correlation_call_path.py @@ -0,0 +1,240 @@ +"""Unavailable execution correlation fails closed at ingress and in the proxy.""" + +from __future__ import annotations + +import hashlib +from unittest.mock import AsyncMock, MagicMock, patch + +import httpx +import pytest + +from cmcp_runtime.audit.chain import AuditChain +from cmcp_runtime.catalog.loader import ( + ApprovedDefinition, + CatalogEntry, + ServerIdentity, + ToolCatalog, +) +from cmcp_runtime.config import AttestationConfig, Config, EnforcementMode +from cmcp_runtime.policy.evaluator import PolicyDecision, PolicyEvaluator +from cmcp_runtime.session.state import SessionState + +AGENT = "spiffe://example.org/agent-a" + + +def _decision() -> PolicyDecision: + return PolicyDecision( + allowed=True, + enforcement_mode=EnforcementMode.ENFORCING, + rule_matched=None, + advice={}, + evaluation_ms=0.1, + would_have_denied=False, + ) + + +def _evaluator() -> PolicyEvaluator: + evaluator = MagicMock(spec=PolicyEvaluator) + evaluator.evaluate.return_value = _decision() + evaluator.authorize_egress.return_value = _decision() + evaluator.bundle_hash = "sha256:" + "0" * 64 + evaluator.enforcement_mode = EnforcementMode.ENFORCING + return evaluator + + +def _make_proxy(chain: AuditChain, mode=EnforcementMode.ENFORCING): + from cmcp_runtime.mcp.proxy import CMCPProxy + + entry = CatalogEntry( + tool_name="billing.charge", + server=ServerIdentity( + display_name="Local", + url="https://local.invalid/mcp", + tls_fingerprint="SHA256:" + "A" * 43 + "=", + spiffe_id=None, + transport="http-sse", + rotation_mode="key-pinned", + ), + approved_definition=ApprovedDefinition( + description="charge", input_schema={}, output_schema=None + ), + definition_hash="sha256:" + "0" * 64, + compliance_domain="public", + requires_baa=False, + sensitivity_level="public", + added_at="2026-08-25T00:00:00Z", + approved_by="issue-565", + ) + catalog = ToolCatalog(entries={"billing.charge": entry}, catalog_hash="sha256:" + "1" * 64) + config = Config(attestation=AttestationConfig(enforcement_mode=mode)) + with ( + patch("cmcp_runtime.mcp.proxy.MCPGateway") as gateway, + patch("cmcp_runtime.mcp.proxy.MCPResponseScanner"), + ): + scan = MagicMock() + scan.allowed = True + scan.threats = [] + scan.content = None + gateway.return_value.intercept_tool_call.return_value = (True, None) + gateway.return_value.intercept_tool_response.return_value = scan + proxy = CMCPProxy( + catalog, + _evaluator(), + SessionState(session_id="s-565"), + chain, + config, + ) + proxy._check_upstream_drift = AsyncMock(return_value=False) + proxy._forward_to_upstream = AsyncMock(return_value='{"ok": true}') + return proxy + + +def _tool_entries(chain: AuditChain): + return [e for e in chain.entries if e.entry_type in ("tool_call", "fault", "egress_denied")] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("metadata,allowed,reason,audit_id", [ + ({"execution_id": 17}, False, "execution_invalid_execution_id", None), + ({"execution_id": True}, False, "execution_invalid_execution_id", None), + ({"execution_id": []}, False, "execution_invalid_execution_id", None), + ({"execution_id": {}}, False, "execution_invalid_execution_id", None), + ({"execution_id": None}, False, "execution_invalid_execution_id", None), + ({"execution_id": ""}, False, "execution_invalid_execution_id", None), + ({"execution_id": "has space"}, False, "execution_invalid_execution_id", None), + ({"execution_id": "x" * 201}, False, "execution_invalid_execution_id", None), + ({"execution_id": "valid-id"}, False, "execution_correlation_unavailable", "valid-id"), + ({}, True, None, None), +]) +async def test_http_execution_identity_validation(metadata, allowed, reason, audit_id): + from cmcp_runtime.mcp.server import MCPServer + + chain = AuditChain(session_id="s-565") + proxy = _make_proxy(chain) + transport = httpx.ASGITransport(app=MCPServer(proxy).app) + async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client: + response = await client.post("/mcp", json={ + "jsonrpc": "2.0", "id": 1, "method": "tools/call", + "params": {"name": "billing.charge", "arguments": {"amount": 100}, + "_cmcp": metadata}, + }) + assert response.status_code == (200 if allowed else 403) + assert proxy._forward_to_upstream.await_count == int(allowed) + assert len(_tool_entries(chain)) == 1 + assert _tool_entries(chain)[0].execution_id == audit_id + if reason: + expected_rule = ( + "execution:unavailable" if audit_id else "execution:invalid_execution_id" + ) + assert _tool_entries(chain)[0].policy_rule_matched == expected_rule + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mode", list(EnforcementMode)) +@pytest.mark.parametrize("execution_id,reason", [ + ("valid-id", "execution_correlation_unavailable"), + ("", "execution_invalid_execution_id"), + ("x\ny", "execution_invalid_execution_id"), +]) +async def test_direct_proxy_calls_cannot_enable_execution(execution_id, reason, mode): + chain = AuditChain(session_id="s-565") + proxy = _make_proxy(chain, mode) + for amount in (100, 999): + result = await proxy.call_tool("c1", "billing.charge", {"amount": amount}, + execution_id=execution_id) + assert not result.allowed + assert result.deny_reason == reason + proxy._forward_to_upstream.assert_not_awaited() + assert len(_tool_entries(chain)) == 2 + + +@pytest.mark.asyncio +async def test_execution_refusal_precedes_unknown_tool_lookup_and_discovery(): + """A supplied ID is refused and audited before an unknown tool is resolved.""" + chain = AuditChain(session_id="s-565") + proxy = _make_proxy(chain) + lookup = MagicMock(wraps=proxy._catalog.lookup) + proxy._catalog.lookup = lookup + + result = await proxy.call_tool( + "unknown-call", + "unknown.tool", + {"amount": 100}, + workflow_id="wf-565", + execution_id="valid-id", + ) + + assert not result.allowed + assert result.deny_reason == "execution_correlation_unavailable" + lookup.assert_not_called() + proxy._check_upstream_drift.assert_not_awaited() + proxy._forward_to_upstream.assert_not_awaited() + + [entry] = _tool_entries(chain) + assert entry.call_id == "unknown-call" + assert entry.tool_name == "unknown.tool" + assert entry.server_identity is None + assert entry.workflow_id == "wf-565" + assert entry.execution_id == "valid-id" + expected_hash = "sha256:" + hashlib.sha256(b'{"amount":100}').hexdigest() + assert entry.request_payload_hash == expected_hash + assert entry.policy_rule_matched == "execution:unavailable" + + +@pytest.mark.asyncio +async def test_execution_refusal_precedes_upstream_drift_discovery(): + """A supplied ID is refused before catalog or upstream drift discovery.""" + chain = AuditChain(session_id="s-565") + proxy = _make_proxy(chain) + lookup = MagicMock(wraps=proxy._catalog.lookup) + proxy._catalog.lookup = lookup + proxy._check_upstream_drift = AsyncMock(return_value=True) + + result = await proxy.call_tool( + "drift-call", + "billing.charge", + {"amount": 100}, + execution_id="valid-id", + ) + + assert not result.allowed + assert result.deny_reason == "execution_correlation_unavailable" + lookup.assert_not_called() + proxy._check_upstream_drift.assert_not_awaited() + proxy._forward_to_upstream.assert_not_awaited() + + [entry] = _tool_entries(chain) + assert entry.tool_name == "billing.charge" + assert entry.server_identity is None + assert entry.execution_id == "valid-id" + assert entry.policy_rule_matched == "execution:unavailable" + + +@pytest.mark.asyncio +async def test_execution_refusal_precedes_existing_catalog_drift_health_failure(): + """A supplied ID is audited even when the session is already unhealthy.""" + chain = AuditChain(session_id="s-565") + proxy = _make_proxy(chain) + lookup = MagicMock(wraps=proxy._catalog.lookup) + proxy._catalog.lookup = lookup + proxy._session.catalog_drift = True + + result = await proxy.call_tool( + "drifted-call", + "billing.charge", + {"amount": 100}, + execution_id="valid-id", + ) + + assert not result.allowed + assert result.deny_reason == "execution_correlation_unavailable" + lookup.assert_not_called() + proxy._check_upstream_drift.assert_not_awaited() + proxy._forward_to_upstream.assert_not_awaited() + + [entry] = _tool_entries(chain) + assert entry.call_id == "drifted-call" + assert entry.tool_name == "billing.charge" + assert entry.server_identity is None + assert entry.execution_id == "valid-id" + assert entry.policy_rule_matched == "execution:unavailable" diff --git a/tests/unit/test_workflow_scope.py b/tests/unit/test_workflow_scope.py index 635caa7..2566018 100644 --- a/tests/unit/test_workflow_scope.py +++ b/tests/unit/test_workflow_scope.py @@ -154,7 +154,10 @@ async def test_server_extracts_workflow_id_from_cmcp_params(): original = proxy.call_tool captured: dict = {} - async def _spy(call_id, tool_name, arguments, *, workflow_id=None, declared_data_class=None): + async def _spy( + call_id, tool_name, arguments, *, workflow_id=None, declared_data_class=None, + execution_id=None, + ): captured["workflow_id"] = workflow_id return await original( call_id, @@ -195,7 +198,10 @@ async def test_server_extracts_data_class_from_cmcp_params(): original = proxy.call_tool captured: dict = {} - async def _spy(call_id, tool_name, arguments, *, workflow_id=None, declared_data_class=None): + async def _spy( + call_id, tool_name, arguments, *, workflow_id=None, declared_data_class=None, + execution_id=None, + ): captured["declared_data_class"] = declared_data_class return await original( call_id, @@ -236,7 +242,10 @@ async def test_server_malformed_data_class_does_not_fail_call(): original = proxy.call_tool captured: dict = {} - async def _spy(call_id, tool_name, arguments, *, workflow_id=None, declared_data_class=None): + async def _spy( + call_id, tool_name, arguments, *, workflow_id=None, declared_data_class=None, + execution_id=None, + ): captured["declared_data_class"] = declared_data_class return await original( call_id,