diff --git a/.clusterfuzzlite/Dockerfile b/.clusterfuzzlite/Dockerfile new file mode 100644 index 0000000..a654076 --- /dev/null +++ b/.clusterfuzzlite/Dockerfile @@ -0,0 +1,5 @@ +FROM gcr.io/oss-fuzz-base/base-builder-python + +COPY . $SRC/cmcp +COPY .clusterfuzzlite/build.sh $SRC/build.sh +WORKDIR $SRC/cmcp diff --git a/.clusterfuzzlite/README.md b/.clusterfuzzlite/README.md new file mode 100644 index 0000000..12918a3 --- /dev/null +++ b/.clusterfuzzlite/README.md @@ -0,0 +1,68 @@ +# Fuzzing + +Coverage-guided fuzzing via [ClusterFuzzLite](https://google.github.io/clusterfuzzlite/), +running Atheris against the untrusted input surface. Mirrors the setup in +agentrust-io/agent-manifest, where the same targets found a real bug. + +## What is fuzzed + +| Target | Surface | +| --- | --- | +| `fuzz_attestation_parsers.py` | `parse_event_log` walks a TCG event log: a Spec ID header declaring which digest algorithms are present and how long each is, then events each carrying their own declared digest count and data length. `parse_nv_certify` reads a bare or size-prefixed TPM NV certification. Every one of those lengths comes from the blob, which arrives from the platform before anything about it is verified. | +| `fuzz_canonical_json.py` | The bytes a catalog approval signature covers. | + +## The properties + +The parser target asserts each function fails closed: it returns, or raises the +`ValueError` its module documents (`EventLogError` is a `ValueError`). A +`struct.error`, `IndexError`, `MemoryError` or `OverflowError` reaching the +caller means a declared length was believed. + +`fuzz_canonical_json.py` asserts a round trip: parsing the canonical output must +reproduce the input. That is stronger than checking for a crash, deliberately. +The three RFC 8785 bugs found in the sibling agent-manifest canonicalizer in +September 2026 were all silent; the sharpest normalized two distinct object keys +into one, so the output carried that key twice and a field disappeared from a +document whose signature claimed to cover it. Nothing raised. cmcp does not have +that bug, and this is what keeps it that way. + +Equality is asserted up to JSON's number model, since JSON has one number type +and a large float legitimately re-parses as an int. + +## Standing when added + +Both parsers already fail closed: a local probe of 12,000 mutated inputs found +no undeclared exception escaping either. The canonicalizer is clean too: 17,918 +generated documents round-tripped and 12,080 were refused as declared, with no +invariant violation. These are regression guards. + +## Bundling gotcha + +`compile_python_fuzzer` bundles each target with PyInstaller, which follows +static imports only. The cryptography and pydantic stacks reach `email.mime` +lazily, so without help the bundled target dies at runtime with +`ModuleNotFoundError: No module named 'email.mime'`, and libFuzzer reports that +as a crash in the target rather than a build problem. `build.sh` passes +`--collect-submodules=email`. A new dependency with a lazy import can need the +same treatment. + +PyInstaller also bundles code and not package data. `cmcp_verify`'s import +chain reaches `agentrust_trace`, which loads its JSON schema from inside the +package at import time, so `build.sh` passes `--collect-data` for the three +packages involved. Without it the build check reports the target as broken +with `FileNotFoundError` on `agentrust_trace/schema/trace-v0.2.json`. + +## Running locally + +``` +git clone https://github.com/google/clusterfuzzlite --depth 1 /tmp/clusterfuzzlite +python /tmp/clusterfuzzlite/infra/helper.py build_image --external $PWD +python /tmp/clusterfuzzlite/infra/helper.py build_fuzzers --external --sanitizer address $PWD +python /tmp/clusterfuzzlite/infra/helper.py run_fuzzer --external $PWD fuzz_canonical_json +``` + +## In CI + +`cflite_pr.yml` fuzzes only code the pull request touched, for five minutes. +`cflite_batch.yml` runs every target for an hour, nightly. Both are read-only. +Budget 45 minutes for the PR job: the oss-fuzz base image build dominates. diff --git a/.clusterfuzzlite/build.sh b/.clusterfuzzlite/build.sh new file mode 100755 index 0000000..dab5eaf --- /dev/null +++ b/.clusterfuzzlite/build.sh @@ -0,0 +1,30 @@ +#!/bin/bash -eu +# Build the fuzz targets for ClusterFuzzLite. +# +# Installed rather than put on the path so the targets exercise the same import +# surface a consumer gets. + +cd "$SRC/cmcp" +pip3 install --no-cache-dir . + +# compile_python_fuzzer bundles each target with PyInstaller, which follows +# static imports only. The cryptography and pydantic stacks reach email.mime +# lazily, so without this the bundled target dies at runtime with +# "ModuleNotFoundError: No module named 'email.mime'" and libFuzzer reports it +# as a crash in the target. +PYI_ARGS=( + # Lazy stdlib import from the cryptography and pydantic stacks. + --collect-submodules=email + # PyInstaller bundles code, not package data. cmcp_verify's import chain + # reaches agentrust_trace, which loads its JSON schema from inside the + # package at import time, so without this the bundled target dies with + # FileNotFoundError on agentrust_trace/schema/trace-v0.2.json and the build + # check reports the target as broken. + --collect-data=agentrust_trace + --collect-data=cmcp_runtime + --collect-data=cmcp_verify +) + +for target in "$SRC"/cmcp/.clusterfuzzlite/fuzz_*.py; do + compile_python_fuzzer "$target" "${PYI_ARGS[@]}" +done diff --git a/.clusterfuzzlite/fuzz_attestation_parsers.py b/.clusterfuzzlite/fuzz_attestation_parsers.py new file mode 100644 index 0000000..611e970 --- /dev/null +++ b/.clusterfuzzlite/fuzz_attestation_parsers.py @@ -0,0 +1,48 @@ +#!/usr/bin/python3 +"""Fuzz the attestation blob parsers in cmcp_verify. + +parse_event_log walks a TCG event log: a Spec ID header that declares which +digest algorithms are present and how long each one is, then a run of events +each carrying its own declared digest count and data length. Every one of those +numbers comes from the blob being parsed, and the log arrives from the platform +before anything about it has been verified. + +parse_nv_certify reads a bare or size-prefixed TPM NV certification, where a +length prefix decides how much of the rest is structure. + +The property is that each parser fails closed: it returns, or raises the +ValueError its module documents (EventLogError is a ValueError). A struct.error, +IndexError, MemoryError or OverflowError reaching the caller means a declared +length was believed, and callers written against the documented exception will +not catch it. +""" +import sys + +import atheris + +with atheris.instrument_imports(): + from cmcp_verify.nv_certify import parse_nv_certify + from cmcp_verify.tcg_event_log import parse_event_log + +_TARGETS = [parse_event_log, parse_nv_certify] + + +def TestOneInput(data: bytes) -> None: + if not data: + return + fdp = atheris.FuzzedDataProvider(data) + parser = _TARGETS[fdp.ConsumeIntInRange(0, len(_TARGETS) - 1)] + blob = fdp.ConsumeBytes(fdp.remaining_bytes()) + try: + parser(blob) + except ValueError: + pass + + +def main() -> None: + atheris.Setup(sys.argv, TestOneInput) + atheris.Fuzz() + + +if __name__ == "__main__": + main() diff --git a/.clusterfuzzlite/fuzz_canonical_json.py b/.clusterfuzzlite/fuzz_canonical_json.py new file mode 100644 index 0000000..fe0dba8 --- /dev/null +++ b/.clusterfuzzlite/fuzz_canonical_json.py @@ -0,0 +1,110 @@ +#!/usr/bin/python3 +"""Fuzz the canonicalizer that catalog approvals are signed over. + +canonical_json() produces the bytes an approval signature covers, so those bytes +have to be a faithful, lossless encoding of the input. The property asserted +here is a round trip: parsing the canonical output must reproduce the input. + +That is stronger than checking for a crash, deliberately. The three RFC 8785 +bugs found in the sibling agent-manifest canonicalizer in September 2026 were +all silent. The sharpest was NFC normalization of object keys: two distinct keys +normalized to one, the output carried that key twice, json.loads kept one of the +pair, and a field disappeared from a document whose signature claimed to cover +it. Nothing raised. cmcp's implementation does not have that bug, and this is +what keeps it that way. + +Structure is built from the fuzz data rather than by mutating JSON text, so the +budget goes on key and value shapes (combining marks, surrogate pairs, control +characters, integer boundaries) instead of on producing syntactically valid JSON. +""" +import json +import math +import sys + +import atheris + +with atheris.instrument_imports(): + from cmcp_runtime.catalog.approval import canonical_json + +_MAX_DEPTH = 4 +_MAX_ITEMS = 6 + + +def _build(fdp: atheris.FuzzedDataProvider, depth: int = 0): + if depth >= _MAX_DEPTH or fdp.remaining_bytes() == 0: + return fdp.ConsumeUnicodeNoSurrogates(16) + kind = fdp.ConsumeIntInRange(0, 7) + if kind == 0: + return None + if kind == 1: + return fdp.ConsumeBool() + if kind == 2: + # Straddle the safe-integer boundary on purpose: past it, RFC 8785 maps + # two distinct integers to the same digits, so the canonicalizer has to + # refuse rather than emit them. + return fdp.ConsumeIntInRange(-(2**54), 2**54) + if kind == 3: + return fdp.ConsumeFloat() + if kind == 4: + return fdp.ConsumeUnicodeNoSurrogates(64) + if kind == 5: + return [_build(fdp, depth + 1) for _ in range(fdp.ConsumeIntInRange(0, _MAX_ITEMS))] + return { + fdp.ConsumeUnicodeNoSurrogates(24): _build(fdp, depth + 1) + for _ in range(fdp.ConsumeIntInRange(0, _MAX_ITEMS)) + } + + +def _json_equal(a, b) -> bool: + """Equality up to JSON's number model. + + JSON has one number type, so a float whose shortest form has no fractional + part re-parses as a Python int and will not compare equal to the float it + came from. That is correct output, not a defect, so the round trip is + asserted up to numeric type. + """ + if isinstance(a, bool) or isinstance(b, bool): + return a is b + if isinstance(a, (int, float)) and isinstance(b, (int, float)): + return float(a) == float(b) + if isinstance(a, dict) and isinstance(b, dict): + return a.keys() == b.keys() and all(_json_equal(a[k], b[k]) for k in a) + if isinstance(a, list) and isinstance(b, list): + return len(a) == len(b) and all(_json_equal(x, y) for x, y in zip(a, b)) + return type(a) is type(b) and a == b + + +def _has_nonfinite(value) -> bool: + """NaN and Infinity have no JSON form; the canonicalizer rejects them.""" + if isinstance(value, float): + return not math.isfinite(value) + if isinstance(value, dict): + return any(_has_nonfinite(v) for v in value.values()) + if isinstance(value, list): + return any(_has_nonfinite(v) for v in value) + return False + + +def TestOneInput(data: bytes) -> None: + fdp = atheris.FuzzedDataProvider(data) + value = _build(fdp) + if _has_nonfinite(value): + return + try: + out = canonical_json(value) + except ValueError: + # Declared: CatalogApprovalError is a ValueError, and covers floats, + # integers outside the RFC 8785 safe domain, and unsupported types. + return + + assert _json_equal(json.loads(out), value), f"canonical bytes did not round-trip: {out!r}" + assert canonical_json(value) == out, "canonicalization is not deterministic" + + +def main() -> None: + atheris.Setup(sys.argv, TestOneInput) + atheris.Fuzz() + + +if __name__ == "__main__": + main() diff --git a/.github/workflows/cflite_batch.yml b/.github/workflows/cflite_batch.yml new file mode 100644 index 0000000..a8ff89b --- /dev/null +++ b/.github/workflows/cflite_batch.yml @@ -0,0 +1,37 @@ +name: ClusterFuzzLite batch + +on: + schedule: + # 04:20 UTC daily, off the hour so it does not queue behind everything else + # that runs at midnight. + - cron: '20 4 * * *' + workflow_dispatch: + +permissions: read-all + +concurrency: + group: cflite-batch + cancel-in-progress: false + +jobs: + fuzz: + name: Batch fuzz + runs-on: ubuntu-latest + timeout-minutes: 90 + steps: + - name: Build fuzzers + uses: google/clusterfuzzlite/actions/build_fuzzers@884713a6c30a92e5e8544c39945cd7cb630abcd1 # v1 + with: + language: python + github-token: ${{ secrets.GITHUB_TOKEN }} + sanitizer: address + + - name: Run fuzzers + uses: google/clusterfuzzlite/actions/run_fuzzers@884713a6c30a92e5e8544c39945cd7cb630abcd1 # v1 + with: + github-token: ${{ secrets.GITHUB_TOKEN }} + # Every target, not just what changed, and long enough for the + # attestation parsers to get past their length and tag fields. + fuzz-seconds: 3600 + mode: batch + sanitizer: address diff --git a/.github/workflows/cflite_pr.yml b/.github/workflows/cflite_pr.yml new file mode 100644 index 0000000..0619bbf --- /dev/null +++ b/.github/workflows/cflite_pr.yml @@ -0,0 +1,41 @@ +name: ClusterFuzzLite PR + +on: + pull_request: + paths: + - 'src/**' + - '.clusterfuzzlite/**' + - '.github/workflows/cflite_pr.yml' + +# Read-only. Findings surface in the job log and the uploaded crash artifact +# rather than as code-scanning alerts, so no security-events: write is needed. +permissions: read-all + +concurrency: + group: cflite-pr-${{ github.ref }} + cancel-in-progress: true + +jobs: + fuzz: + name: Fuzz changed code + runs-on: ubuntu-latest + timeout-minutes: 45 + steps: + # Only the address sanitizer. The targets are pure Python under Atheris, + # where undefined-behaviour instrumentation has nothing to instrument. + - name: Build fuzzers + uses: google/clusterfuzzlite/actions/build_fuzzers@884713a6c30a92e5e8544c39945cd7cb630abcd1 # v1 + with: + language: python + github-token: ${{ secrets.GITHUB_TOKEN }} + sanitizer: address + + - name: Run fuzzers + uses: google/clusterfuzzlite/actions/run_fuzzers@884713a6c30a92e5e8544c39945cd7cb630abcd1 # v1 + with: + github-token: ${{ secrets.GITHUB_TOKEN }} + fuzz-seconds: 300 + # code-change fuzzes only what the PR touched, which is what keeps + # this inside a PR's time budget. The nightly batch covers the rest. + mode: code-change + sanitizer: address diff --git a/CHANGELOG.md b/CHANGELOG.md index 4c59bf8..95449f1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,54 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Security + +- **A response arriving during an operator reset raised the successor session.** + The per-session mutation lock serialised a reset and a response elevation but + did not order them, so whichever coroutine acquired it second won. A response + in flight when the reset landed was applied to the successor, which had just + been initialised to `public`, and recorded the pre-reset `call_id` as the call + that raised it. The successor exists to start at the minimum level, so this + carried the closed session's sensitivity across the boundary the reset drew. + `update_from_inspection()` now takes the `reset_count` observed at call entry + and drops a response whose generation no longer matches, logging + `SESSION_RESET_RACE`. The discriminator is `reset_count` rather than + `session_id` because `upgrade_attestation()` rotates the identifier while + deliberately continuing the same session, so a call in flight across an + attestation upgrade must still apply. The previous concurrency test asserted + only that `max_sensitivity` remained a member of `SENSITIVITY_ORDER`, which + every value satisfies. + +- **The reset route accepted the tool-invocation token.** `POST + /sessions/{id}/reset` is not reachable as an MCP tool, but it sat behind the + same single `CMCP_BEARER_TOKEN` as `POST /mcp`, so an agent host holding its + own tool-invocation credential could clear accumulated session sensitivity. + The operator interface (session reset and catalog exception) now takes + `CMCP_OPERATOR_TOKEN`, which must differ from `CMCP_BEARER_TOKEN` and is + required outside `CMCP_DEV_MODE=1` (`OPERATOR_TOKEN_REQUIRED`). Where it is + unset those routes still fall back to the bearer token, so an existing + single-token deployment keeps working until it sets the new variable. + +### Changed + +- The reset audit entry now identifies the session boundary rather than only the + sensitivity transition: `detail` carries the closed session identifier, the + successor identifier, the resulting reset counter, and which credential was + verified. `detail` is inside the canonical body, so those fields are covered by + the entry hash. + +- **The audit chain no longer attributes post-reset entries to the closed + session.** `AuditChain.rotate_session_id()` moves attribution to the successor + after the boundary entry is written, so the reset entry belongs to the session + that reached the recorded value and later entries belong to the successor. + Previously every entry after a reset carried the closed session's identifier + and the successor's identifier appeared nowhere in the chain. + +- A reset now preserves the closed session's final state as a distinct + `ClosedSessionRecord` instead of overwriting it, and + `POST /sessions/{id}/reset` returns `closed_session_max_sensitivity` and + `reset_count`. + ## [0.5.0] - 2026-09-05 ### Security diff --git a/docs/configuration.md b/docs/configuration.md index f07f495..a83d6e4 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -135,6 +135,7 @@ Environment variables control secrets and mode flags that must not appear in con |----------|-------------|-----------| | `CMCP_DEV_MODE=1` | Enables software-only attestation. No hardware TEE required. TRACE Claims will show `partially_verified` status. Required when `provider` is `software-only`. | `attestation.provider` (forces software-only) | | `CMCP_BEARER_TOKEN` | Optional bearer token for runtime HTTP auth. If set, all requests to the runtime must include `Authorization: Bearer `. If unset, no bearer auth is enforced. This token is required for non-loopback binds. | none | +| `CMCP_OPERATOR_TOKEN` | Credential for the operator interface: `POST /sessions/{id}/reset` and `POST /catalog/exception`. Required outside `CMCP_DEV_MODE=1` (`OPERATOR_TOKEN_REQUIRED`), and must differ from `CMCP_BEARER_TOKEN`. When set, those two routes accept only this token and reject the tool-invocation token; when unset they fall back to `CMCP_BEARER_TOKEN`. A reset lowers accumulated session sensitivity, so an agent host holding only the tool-invocation token cannot clear the state that monotonicity exists to keep. | none | | `OPAQUE_ATTESTATION_URL` | Enables the OPAQUE Managed Runtime provider. Must be set to the OPAQUE attestation service URL. Required when `provider` is `opaque` or `auto` on OPAQUE infrastructure. | enables `opaque` provider detection | | `CMCP_POLICY_HASH` | SHA-256 hash of the approved policy bundle. Required in non-dev mode and checked by startup before Agent Manifest binding. The gateway fails closed at startup if this is unset and `CMCP_DEV_MODE` is not `1`. Format: `sha256:`. | none (startup policy integrity check) | | `CMCP_CATALOG_HASH` | SHA-256 hash of the approved `catalog.json`. Required in non-dev mode. The gateway fails closed at startup if this is unset and `CMCP_DEV_MODE` is not `1`. Format: `sha256:`. | none (additional startup check) | diff --git a/src/cmcp_runtime/audit/chain.py b/src/cmcp_runtime/audit/chain.py index 1e0ca28..a66cdd2 100644 --- a/src/cmcp_runtime/audit/chain.py +++ b/src/cmcp_runtime/audit/chain.py @@ -257,6 +257,18 @@ def append( self._notify_sinks(entry) return entry + def rotate_session_id(self, new_session_id: str) -> None: + """Attribute subsequent entries to ``new_session_id``. + + A credentialed reset closes one session and opens a successor on the same + hash-linked chain. Without this, every entry after a reset carries the + closed session's identifier and the successor's identifier appears + nowhere, so the record cannot say which session an entry belongs to. + Entries already appended are unchanged: they are hashed and remain + attributed to the session that produced them. + """ + self._session_id = new_session_id + def add_sink(self, sink: Callable[[AuditEntry], None]) -> None: """Register a read-only observer of appended entries. See __init__.""" self._sinks.append(sink) diff --git a/src/cmcp_runtime/cli.py b/src/cmcp_runtime/cli.py index e1f91f6..db99c15 100644 --- a/src/cmcp_runtime/cli.py +++ b/src/cmcp_runtime/cli.py @@ -95,6 +95,7 @@ def build_server(ctx: RuntimeContext) -> MCPServer: audit_chain=audit_chain, session=session, bearer_token=ctx.config.bearer_token, + operator_token=ctx.config.operator_token, ) diff --git a/src/cmcp_runtime/config.py b/src/cmcp_runtime/config.py index dae935e..7fd4597 100644 --- a/src/cmcp_runtime/config.py +++ b/src/cmcp_runtime/config.py @@ -134,6 +134,13 @@ class Config: audit_db_path: str = "audit.db" # AUDIT-001: durable audit chain storage dev_mode: bool = False bearer_token: str | None = None + #: Credential for the operator interface (session reset, catalog exception). + #: Held separately from ``bearer_token`` so that the credential authorizing a + #: reset is not the credential an agent host already holds to invoke tools. + #: A reset lowers accumulated session sensitivity, so an agent able to + #: present its own tool-invocation token to the reset route could clear the + #: state that monotonicity exists to keep. + operator_token: str | None = None #: AARM R6. A named conformance profile tightens defaults that stay #: permissive for developers. None is the default, and nothing changes. #: "aarm" requires an Agent Manifest binding, because R6 says every receipt @@ -446,6 +453,14 @@ def load_config(path: str) -> Config: dev_mode = DEV_MODE # TEE-002: use the frozen constant, never re-read from env bearer_token = os.environ.get("CMCP_BEARER_TOKEN") or None + operator_token = os.environ.get("CMCP_OPERATOR_TOKEN") or None + + if operator_token is not None and operator_token == bearer_token: + raise ConfigError( + "CMCP_OPERATOR_TOKEN must differ from CMCP_BEARER_TOKEN. The operator " + "credential authorizes a session-sensitivity reset and must not be " + "reachable by a holder of the tool-invocation credential." + ) default_listen_addr = ( "127.0.0.1:8443" @@ -533,4 +548,5 @@ def load_config(path: str) -> Config: conformance_profile=profile, dev_mode=dev_mode, bearer_token=bearer_token, + operator_token=operator_token, ) diff --git a/src/cmcp_runtime/mcp/proxy.py b/src/cmcp_runtime/mcp/proxy.py index 0c6bccd..99d3db9 100644 --- a/src/cmcp_runtime/mcp/proxy.py +++ b/src/cmcp_runtime/mcp/proxy.py @@ -91,6 +91,9 @@ class _CallFinalizationState: """Per-invocation facts needed for honest terminal finalization.""" failure_stage: str = "call_entry" + # Session generation observed at call entry, so a response landing after an + # operator reset is not applied to the successor session. + reset_count: int | None = None effect_boundary_state: _EffectBoundaryState = _EffectBoundaryState.PRE_TRANSPORT request_payload_hash: str | None = None response_payload_hash: str | None = None @@ -828,6 +831,10 @@ async def call_tool( ) -> CallResult: """Run one call and guarantee one terminal on failure or cancellation.""" finalization = _CallFinalizationState() + # The session generation this call was issued under. A reset arriving + # mid-call closes that session, and this response must not raise the + # successor. + finalization.reset_count = self._session.reset_count try: return await self._call_tool_impl( call_id, @@ -1242,6 +1249,7 @@ class above the tool's catalogued sensitivity_level. It can never lower else [entry.sensitivity_level] ), injection_detected=injection_detected, + for_reset_count=_finalization.reset_count, response_allowed=False, ) threat_categories = ",".join( @@ -1314,11 +1322,18 @@ class above the tool's catalogued sensitivity_level. It can never lower injection_threshold = None _finalization.failure_stage = "session_update" async with self._session.mutation_lock: - self._session.update_from_inspection( + applied = self._session.update_from_inspection( call_id=call_id, sensitivity_tags=response_sensitivity, injection_detected=injection_detected, response_allowed=True, + for_reset_count=_finalization.reset_count, + ) + if not applied: + logger.warning( + "SESSION_RESET_RACE: response for call_id=%s dropped from session " + "state; the session it was issued under was closed by a reset", + call_id, ) # Step 5: egress Cedar policy check diff --git a/src/cmcp_runtime/mcp/server.py b/src/cmcp_runtime/mcp/server.py index 16df4cf..3b0c25e 100644 --- a/src/cmcp_runtime/mcp/server.py +++ b/src/cmcp_runtime/mcp/server.py @@ -14,6 +14,7 @@ import json import logging import os +import re import time import uuid from collections import defaultdict @@ -32,7 +33,7 @@ if TYPE_CHECKING: from cmcp_runtime.audit.chain import AuditChain from cmcp_runtime.session.manager import SessionManager - from cmcp_runtime.session.state import SessionState + from cmcp_runtime.session.state import ClosedSessionRecord, SessionState logger = logging.getLogger(__name__) @@ -48,6 +49,12 @@ class StatelessKernel: # Endpoints exempt from bearer-token auth (Kubernetes liveness / readiness probes) _AUTH_EXEMPT_PATHS = {"/health", "/readyz"} +# The operator interface. These routes are not reachable as MCP tools and, when an +# operator token is configured, they do not accept the tool-invocation token: a +# reset lowers accumulated session sensitivity, so the credential that authorizes +# one must not be the credential an agent host already holds. +_OPERATOR_PATH_RE = re.compile(r"^/(?:sessions/[^/]+/reset|catalog/exception)$") + # DOS-001: default ceiling on a single request body. Overridable per # deployment via MCPServer(max_request_bytes=...). Named here rather than # left inline on the constructor so the argument-shape caps below can be @@ -264,15 +271,27 @@ async def dispatch(self, request: Request, call_next: Any) -> Response: class _BearerAuthMiddleware(BaseHTTPMiddleware): - """AUTH-001 (CRITICAL): validate Authorization: Bearer on all protected endpoints.""" + """AUTH-001 (CRITICAL): validate Authorization: Bearer on all protected endpoints. + + Operator routes are matched against ``_OPERATOR_PATH_RE`` and, when an + operator token is configured, accept only that token. Where none is + configured they fall back to the bearer token, which keeps existing + single-token deployments working; startup refuses that outside dev mode. + """ - def __init__(self, app: Any, *, bearer_token: str) -> None: + def __init__( + self, app: Any, *, bearer_token: str, operator_token: str | None = None + ) -> None: super().__init__(app) self._token = bearer_token + self._operator_token = operator_token async def dispatch(self, request: Request, call_next: Any) -> Response: if request.url.path in _AUTH_EXEMPT_PATHS: return await call_next(request) + expected = self._token + if self._operator_token is not None and _OPERATOR_PATH_RE.match(request.url.path): + expected = self._operator_token auth = request.headers.get("Authorization", "") prefix = "Bearer " if not auth.startswith(prefix): @@ -283,7 +302,7 @@ async def dispatch(self, request: Request, call_next: Any) -> Response: ) provided = auth[len(prefix):] # Constant-time compare to prevent timing oracle on the token - if not hmac.compare_digest(provided, self._token): + if not hmac.compare_digest(provided, expected): logger.warning("AUTH_FAILURE: invalid bearer token from %s", request.client) return JSONResponse( {"error": "unauthorized", "error_code": "INVALID_BEARER_TOKEN"}, @@ -308,6 +327,7 @@ def __init__( session_manager: SessionManager | None = None, audit_chain: AuditChain | None = None, bearer_token: str | None = None, + operator_token: str | None = None, session: SessionState | None = None, max_request_bytes: int = _DEFAULT_MAX_REQUEST_BYTES, ) -> None: @@ -316,6 +336,7 @@ def __init__( self._audit_chain = audit_chain self._session = session self._max_request_bytes = max_request_bytes + self._operator_token = operator_token self._audit = audit_chain # Chains of closed sessions, kept so /audit/export still serves them # after the live session rotates. @@ -329,10 +350,20 @@ def __init__( requests_per_minute=60, ) middleware = [rate_limit] + ( - [Middleware(_BearerAuthMiddleware, bearer_token=bearer_token)] + [ + Middleware( + _BearerAuthMiddleware, + bearer_token=bearer_token, + operator_token=operator_token, + ) + ] if bearer_token is not None else [] ) + # Final state of sessions closed by a credentialed reset, kept so the + # value a closed session reached survives the successor starting at the + # minimum level. + self._closed_sessions: dict[str, ClosedSessionRecord] = {} # AUTH-004: session cleanup interval configurable via env var (default 60s) self._cleanup_interval_s: int = int( os.environ.get("CMCP_SESSION_CLEANUP_INTERVAL_SECONDS", "60") @@ -925,16 +956,29 @@ async def _session_reset(self, request: Request) -> Response: return JSONResponse( {"error": f"session_id={session_id} not found"}, status_code=404 ) + # The middleware has already authenticated the operator credential on this + # route; record which credential was verified so the entry says so. + credential = ( + "operator_token" if self._operator_token is not None else "bearer_token" + ) # AUTH-002: lock guards against a concurrent tool-call coroutine modifying sensitivity. async with self._session.mutation_lock: # Capture the pre-reset sensitivity: reset() drops it back to # "public", and the elevated value the session held at reset time # is exactly the forensic detail the audit entry must preserve. sensitivity_before = self._session.max_sensitivity + closed = self._session.snapshot_for_close( + reason="operator reset via API", + authorized_by=credential, + ) old_id, new_id = self._session.reset( reason="operator reset via API", - authorized_by="api", + authorized_by=credential, ) + reset_count = self._session.reset_count + self._closed_sessions[closed.session_id] = closed + # Written while the chain still names the closed session, so the entry + # recording the boundary belongs to the session that reached that value. self._audit_chain.append( "session_reset", call_id=None, @@ -942,10 +986,21 @@ async def _session_reset(self, request: Request) -> Response: policy_decision="n/a", session_sensitivity_before=sensitivity_before, session_sensitivity_after=self._session.max_sensitivity, + detail={ + "closed_session_id": old_id, + "successor_session_id": new_id, + "reset_count": reset_count, + "credential_verified": credential, + "reason": "operator reset via API", + }, ) + # Entries after the boundary belong to the successor. + self._audit_chain.rotate_session_id(new_id) return JSONResponse({ "old_session_id": old_id, "new_session_id": new_id, + "closed_session_max_sensitivity": closed.max_sensitivity, + "reset_count": reset_count, "status": "reset", "attestation_stale": False, }) diff --git a/src/cmcp_runtime/session/state.py b/src/cmcp_runtime/session/state.py index b1e7006..44aaa51 100644 --- a/src/cmcp_runtime/session/state.py +++ b/src/cmcp_runtime/session/state.py @@ -78,6 +78,26 @@ class InjectionEvent: timestamp: str +@dataclass(frozen=True) +class ClosedSessionRecord: + """The final state of a session closed by a credentialed reset. + + Held apart from the successor's live state so that the accumulated value the + closed session reached is preserved rather than overwritten. The successor + starts at the minimum level, and this is the only place its predecessor's + final value survives outside the audit chain. + """ + + session_id: str + max_sensitivity: str + sensitivity_raised_at: str | None + sensitivity_raised_by_call: str | None + reset_count: int + closed_at: str + reason: str + authorized_by: str + + @dataclass class SessionState: """ @@ -88,9 +108,17 @@ class SessionState: only way to lower sensitivity. update_from_inspection() is the ONLY place where session sensitivity state - is updated. It is called by InspectionPipeline after all inspection stages + is updated. It is called by the proxy response path after all inspection stages complete, including for denied responses (a denied high-sensitivity response still raises session sensitivity because the agent knows the call was attempted). + + A response is only allowed to raise the session it was issued under. Callers + pass the ``reset_count`` observed when the call started and a response that + lands after a reset is dropped rather than applied to the successor. The + discriminator is ``reset_count`` and not ``session_id`` because + upgrade_attestation() rotates ``session_id`` while deliberately continuing + the same session at its current sensitivity, so a call in flight across an + attestation upgrade must still be applied. """ session_id: str @@ -125,12 +153,23 @@ def update_from_inspection( sensitivity_tags: list[str], injection_detected: bool, response_allowed: bool, # noqa: ARG002 (logged for future use) - ) -> None: + *, + for_reset_count: int | None = None, + ) -> bool: """ Update session state from an inspection result. - Called by InspectionPipeline after all stages complete. + Called by the proxy response path after all stages complete. Returns True + if the state was updated, False if the response belonged to a session that + has since been closed by a reset and was therefore dropped. + + ``for_reset_count`` is the reset counter observed when the call started. + When it does not match the current counter the response is evidence about + a closed session and must not raise the successor, whose whole purpose is + to start at the minimum level. """ + if for_reset_count is not None and for_reset_count != self.reset_count: + return False for tag in sensitivity_tags: new_max = _max_sensitivity(self.max_sensitivity, tag, self.sensitivity_order) if new_max != self.max_sensitivity: @@ -145,6 +184,24 @@ def update_from_inspection( timestamp=datetime.now(tz=UTC).isoformat(), ) ) + return True + + def snapshot_for_close(self, *, reason: str, authorized_by: str) -> ClosedSessionRecord: + """Capture this session's final state before a reset opens a successor. + + Call inside the mutation lock, immediately before reset(), so the value + recorded is the one the session held at the ordered session boundary. + """ + return ClosedSessionRecord( + session_id=self.session_id, + max_sensitivity=self.max_sensitivity, + sensitivity_raised_at=self.sensitivity_raised_at, + sensitivity_raised_by_call=self.sensitivity_raised_by_call, + reset_count=self.reset_count, + closed_at=datetime.now(tz=UTC).isoformat(), + reason=reason, + authorized_by=authorized_by, + ) def reset(self, *, reason: str, authorized_by: str) -> tuple[str, str]: """ diff --git a/src/cmcp_runtime/startup.py b/src/cmcp_runtime/startup.py index 7ca8ec7..5989bb1 100644 --- a/src/cmcp_runtime/startup.py +++ b/src/cmcp_runtime/startup.py @@ -441,6 +441,19 @@ def run_startup(config_path: str) -> RuntimeContext: ) sys.exit(1) + # A session reset lowers accumulated session sensitivity. Requiring a + # separate credential for it keeps the reset out of reach of a holder of the + # tool-invocation token, which is the whole point of the monotonic state. + if config.operator_token is None and not config.dev_mode: + _fatal( + "OPERATOR_TOKEN_REQUIRED", + "CMCP_OPERATOR_TOKEN env var is not set. " + "Set it to a secret token, distinct from CMCP_BEARER_TOKEN, that " + "operators must present to the session reset and catalog exception " + "routes. Set CMCP_DEV_MODE=1 only in development.", + ) + sys.exit(1) + # Step 4: policy bundle policy_expected_hash = os.environ.get("CMCP_POLICY_HASH") if policy_expected_hash is None and not config.dev_mode: diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index a99794b..a5b283a 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -450,3 +450,31 @@ def test_compliance_domains_non_mapping_raises(config_file): path = config_file("sensitivity:\n compliance_domains: not_a_mapping\n") with pytest.raises(ConfigError, match="mapping"): load_config(path) + + +# ── OPQ_P0006: the operator credential must be distinct ─────────────────────── + + +def test_operator_token_is_loaded(config_file, monkeypatch): + import cmcp_runtime.config as config_module + + monkeypatch.setattr(config_module, "DEV_MODE", False) + monkeypatch.setenv("CMCP_BEARER_TOKEN", "tool-token") + monkeypatch.setenv("CMCP_OPERATOR_TOKEN", "operator-token") + + cfg = load_config(config_file("")) + + assert cfg.bearer_token == "tool-token" + assert cfg.operator_token == "operator-token" + + +def test_operator_token_equal_to_bearer_token_is_refused(config_file, monkeypatch): + """Reusing the tool-invocation token as the operator token defeats the separation.""" + import cmcp_runtime.config as config_module + + monkeypatch.setattr(config_module, "DEV_MODE", False) + monkeypatch.setenv("CMCP_BEARER_TOKEN", "same-token") + monkeypatch.setenv("CMCP_OPERATOR_TOKEN", "same-token") + + with pytest.raises(ConfigError, match="must differ from CMCP_BEARER_TOKEN"): + load_config(config_file("")) diff --git a/tests/unit/test_session.py b/tests/unit/test_session.py index 38654d2..ac8c399 100644 --- a/tests/unit/test_session.py +++ b/tests/unit/test_session.py @@ -226,3 +226,53 @@ def test_upgrade_attestation_does_not_increment_reset_count(): state = SessionState(session_id="s1") state.upgrade_attestation() assert state.reset_count == 0 + + +@pytest.mark.asyncio +async def test_response_from_closed_session_does_not_raise_successor(): + """A response issued before an operator reset must not elevate the successor. + + The successor session exists to start at the minimum level. A response that + was in flight when the reset closed the previous session is evidence about + that closed session, so it is dropped rather than applied. + """ + state = SessionState(session_id="s-pre") + state.update_from_inspection("call-A", ["pii"], False, True) + generation = state.reset_count + + async def _responder() -> bool: + async with state.mutation_lock: + return state.update_from_inspection( + "call-B", ["confidential"], False, True, for_reset_count=generation + ) + + async def _reset() -> None: + async with state.mutation_lock: + state.reset(reason="operator reset via API", authorized_by="test") + + # Hold the lock so both queue behind us, then let the reset win it. + await state.mutation_lock.acquire() + resetter = asyncio.create_task(_reset()) + await asyncio.sleep(0) + responder = asyncio.create_task(_responder()) + await asyncio.sleep(0) + state.mutation_lock.release() + await resetter + applied = await responder + + assert applied is False + assert state.max_sensitivity == "public" + assert state.sensitivity_raised_by_call is None + + +@pytest.mark.asyncio +async def test_response_across_attestation_upgrade_still_raises(): + """upgrade_attestation() continues the session, so an in-flight response applies.""" + state = SessionState(session_id="s1") + generation = state.reset_count + state.upgrade_attestation() + applied = state.update_from_inspection( + "call-C", ["pii"], False, True, for_reset_count=generation + ) + assert applied is True + assert state.max_sensitivity == "pii" diff --git a/tests/unit/test_session_reset.py b/tests/unit/test_session_reset.py index 6c8faa9..5519418 100644 --- a/tests/unit/test_session_reset.py +++ b/tests/unit/test_session_reset.py @@ -224,3 +224,141 @@ def test_reset_without_session_configured_returns_501(): client = TestClient(server.app, raise_server_exceptions=True) resp = client.post("/sessions/bare-sess/reset") assert resp.status_code == 501 + + +# ── OPQ_P0006: the reset credential is not the tool-invocation credential ───── + + +def _make_token_server(session_id: str = "sess-tok-001", *, operator_token: str | None): + """Server with a tool-invocation bearer token and an optional operator token.""" + from cmcp_runtime.mcp.proxy import CMCPProxy + + cfg = Config() + cfg.attestation = AttestationConfig(enforcement_mode=EnforcementMode.ENFORCING) + session = SessionState(session_id=session_id) + chain = AuditChain(session_id) + with patch("cmcp_runtime.mcp.proxy.MCPGateway"), \ + patch("cmcp_runtime.mcp.proxy.MCPResponseScanner"): + proxy = CMCPProxy(_make_catalog(), _make_evaluator(), session, chain, cfg) + wire_mock_gateway(proxy) + server = MCPServer( + proxy, + session=session, + audit_chain=chain, + bearer_token="tool-token", + operator_token=operator_token, + ) + return server, session, chain + + +def test_reset_rejects_the_tool_invocation_token(): + """The credential that invokes tools must not authorize a sensitivity reset.""" + server, session, _ = _make_token_server(operator_token="operator-token") + client = TestClient(server.app, raise_server_exceptions=True) + + resp = client.post( + f"/sessions/{session.session_id}/reset", + headers={"Authorization": "Bearer tool-token"}, + ) + assert resp.status_code == 401 + assert resp.json()["error_code"] == "INVALID_BEARER_TOKEN" + + +def test_reset_accepts_the_operator_token(): + server, session, _ = _make_token_server(operator_token="operator-token") + original_id = session.session_id + client = TestClient(server.app, raise_server_exceptions=True) + + resp = client.post( + f"/sessions/{original_id}/reset", + headers={"Authorization": "Bearer operator-token"}, + ) + assert resp.status_code == 200 + assert resp.json()["old_session_id"] == original_id + + +def test_tool_endpoint_rejects_the_operator_token(): + """The separation runs both ways: the operator credential is not a tool credential.""" + server, _, _ = _make_token_server(operator_token="operator-token") + client = TestClient(server.app, raise_server_exceptions=True) + + resp = client.get("/tools/list", headers={"Authorization": "Bearer operator-token"}) + assert resp.status_code == 401 + + +def test_reset_falls_back_to_bearer_token_when_no_operator_token(): + """Single-token deployments keep working; startup refuses them outside dev mode.""" + server, session, _ = _make_token_server(operator_token=None) + client = TestClient(server.app, raise_server_exceptions=True) + + resp = client.post( + f"/sessions/{session.session_id}/reset", + headers={"Authorization": "Bearer tool-token"}, + ) + assert resp.status_code == 200 + + +# ── OPQ_P0006: the reset record carries the session boundary ────────────────── + + +def test_reset_audit_entry_identifies_both_sessions_and_the_credential(): + server, session, chain = _make_token_server(operator_token="operator-token") + session.update_from_inspection("call-A", ["pii"], False, True) + original_id = session.session_id + + client = TestClient(server.app, raise_server_exceptions=True) + resp = client.post( + f"/sessions/{original_id}/reset", + headers={"Authorization": "Bearer operator-token"}, + ) + new_id = resp.json()["new_session_id"] + + entry = next(e for e in chain.entries if e.entry_type == "session_reset") + assert entry.session_id == original_id + assert entry.session_sensitivity_before == "pii" + assert entry.session_sensitivity_after == "public" + assert entry.detail["closed_session_id"] == original_id + assert entry.detail["successor_session_id"] == new_id + assert entry.detail["reset_count"] == 1 + assert entry.detail["credential_verified"] == "operator_token" + assert entry.prev_entry_hash + # detail is inside the canonical body, so these fields are hash-covered + assert entry.entry_hash == entry.compute_hash() + + +def test_entries_after_a_reset_are_attributed_to_the_successor(): + """Before this, every later entry carried the closed session's identifier.""" + server, session, chain = _make_token_server(operator_token="operator-token") + original_id = session.session_id + + client = TestClient(server.app, raise_server_exceptions=True) + new_id = client.post( + f"/sessions/{original_id}/reset", + headers={"Authorization": "Bearer operator-token"}, + ).json()["new_session_id"] + + later = chain.append("session_start", policy_decision="n/a") + assert later.session_id == new_id + reset_entry = next(e for e in chain.entries if e.entry_type == "session_reset") + assert reset_entry.session_id == original_id + assert chain.verify_chain() + + +def test_closed_session_final_value_is_preserved_apart_from_the_successor(): + server, session, _ = _make_token_server(operator_token="operator-token") + session.update_from_inspection("call-A", ["pii"], False, True) + original_id = session.session_id + + client = TestClient(server.app, raise_server_exceptions=True) + resp = client.post( + f"/sessions/{original_id}/reset", + headers={"Authorization": "Bearer operator-token"}, + ) + + assert resp.json()["closed_session_max_sensitivity"] == "pii" + assert session.max_sensitivity == "public" + closed = server._closed_sessions[original_id] + assert closed.session_id == original_id + assert closed.max_sensitivity == "pii" + assert closed.sensitivity_raised_by_call == "call-A" + assert closed.authorized_by == "operator_token"