fix(cli): wire bearer token, audit store, and attestation staleness into the live server

src/cmcp_runtime/audit/store.py

@@ -5,6 +5,7 @@
 import json
 import logging
 import sqlite3
+import threading
 from dataclasses import asdict
 from pathlib import Path
 
@@ -40,7 +41,10 @@ class SqliteAuditStore:
 
     def __init__(self, db_path: Path) -> None:
         self._db_path = db_path
+        # check_same_thread=False allows use from async handlers and worker
+        # threads; all access is serialised through self._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.executescript(_CREATE_TABLE)
@@ -49,21 +53,22 @@ def __init__(self, db_path: Path) -> None:
 
     def append(self, entry: AuditEntry) -> None:
         payload = json.dumps(asdict(entry), sort_keys=True, separators=(",", ":"))
-        self._conn.execute(
-            "INSERT INTO audit_entries "
-            "(sequence_number, session_id, entry_id, entry_type, entry_hash, prev_entry_hash, payload) "
-            "VALUES (?, ?, ?, ?, ?, ?, ?)",
-            (
-                entry.sequence_number,
-                entry.session_id,
-                entry.entry_id,
-                entry.entry_type,
-                entry.entry_hash,
-                entry.prev_entry_hash,
-                payload,
-            ),
-        )
-        self._conn.commit()
+        with self._lock:
+            self._conn.execute(
+                "INSERT INTO audit_entries "
+                "(sequence_number, session_id, entry_id, entry_type, entry_hash, prev_entry_hash, payload) "
+                "VALUES (?, ?, ?, ?, ?, ?, ?)",
+                (
+                    entry.sequence_number,
+                    entry.session_id,
+                    entry.entry_id,
+                    entry.entry_type,
+                    entry.entry_hash,
+                    entry.prev_entry_hash,
+                    payload,
+                ),
+            )
+            self._conn.commit()
 
     def find_orphaned_sessions(self) -> list[str]:
         """
@@ -72,16 +77,18 @@ def find_orphaned_sessions(self) -> list[str]:
         These represent sessions that were open when the gateway last stopped,
         either due to a crash or an unclean shutdown.
         """
-        cur = self._conn.execute(
-            """
-            SELECT DISTINCT session_id FROM audit_entries
-            WHERE entry_type = 'session_start'
-              AND session_id NOT IN (
-                  SELECT session_id FROM audit_entries WHERE entry_type = 'session_end'
-              )
-            """
-        )
-        return [row[0] for row in cur.fetchall()]
+        with self._lock:
+            cur = self._conn.execute(
+                """
+                SELECT DISTINCT session_id FROM audit_entries
+                WHERE entry_type = 'session_start'
+                  AND session_id NOT IN (
+                      SELECT session_id FROM audit_entries WHERE entry_type = 'session_end'
+                  )
+                """
+            )
+            return [row[0] for row in cur.fetchall()]
 
     def close(self) -> None:
-        self._conn.close()
+        with self._lock:
+            self._conn.close()

src/cmcp_runtime/benchmarks.py

@@ -194,12 +194,6 @@ def _make_proxy(bundle: Any, catalog: Any) -> tuple[Any, Any]:
     session = SessionState(session_id=str(uuid.uuid4()))
     chain = AuditChain(session_id=session.session_id)
 
-    agt_result = MagicMock(
-        sensitivity_tags=[],
-        injection_detected=False,
-        modified_response=b'{"result": "benchmark-ok"}',
-    )
-
     with patch("cmcp_runtime.mcp.proxy.MCPGateway"), \
          patch("cmcp_runtime.mcp.proxy.MCPResponseScanner"):
         proxy = CMCPProxy(
@@ -209,8 +203,21 @@ def _make_proxy(bundle: Any, catalog: Any) -> tuple[Any, Any]:
             audit_chain=chain,
             config=config,
         )
+        # Mock the gateway seam (pre-call check, upstream forward, response
+        # scan) so benchmarks measure cmcp overhead, not network latency.
         proxy._mcp_gateway = MagicMock()
-        proxy._mcp_gateway.call_tool = AsyncMock(return_value=agt_result)
+        proxy._mcp_gateway.intercept_tool_call = MagicMock(return_value=(True, "ok"))
+        proxy._forward_to_upstream = AsyncMock(
+            return_value='{"result": "benchmark-ok"}'
+        )
+        proxy._mcp_gateway.intercept_tool_response = MagicMock(
+            return_value=MagicMock(
+                allowed=True,
+                content='{"result": "benchmark-ok"}',
+                threats=[],
+                action="allowed",
+            )
+        )
 
     return proxy, evaluator
 

src/cmcp_runtime/cli.py

@@ -2,10 +2,62 @@
 
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
+
 import click
 
 from cmcp_runtime import __version__
 
+if TYPE_CHECKING:
+    from cmcp_runtime.mcp.server import MCPServer
+    from cmcp_runtime.startup import RuntimeContext
+
+
+def build_server(ctx: RuntimeContext) -> MCPServer:
+    """
+    Compose the running gateway from a validated RuntimeContext.
+
+    All components validated by run_startup() MUST be wired here — a component
+    that is validated but not passed through is silently inert in production
+    (the AUTH-001 bearer token and AUDIT-001 store were both lost this way).
+    """
+    from cmcp_runtime.audit.trace_claim import _PROVIDER_MAP
+    from cmcp_runtime.mcp.proxy import CMCPProxy
+    from cmcp_runtime.mcp.server import MCPServer
+    from cmcp_runtime.policy.evaluator import PolicyEvaluator
+    from cmcp_runtime.session.manager import SessionManager
+
+    # Resolve provider string to canonical platform name for Cedar context.
+    # Falls back to the raw provider string if not in the map (e.g. future providers).
+    attestation_platform = _PROVIDER_MAP.get(
+        ctx.attestation_report.provider, ctx.attestation_report.provider
+    )
+
+    # AUDIT-001/AUDIT-002: sessions MUST be created through SessionManager so the
+    # chain is backed by the durable SQLite store and TEE-anchored at creation.
+    session_manager = SessionManager(ctx)
+    session, audit_chain = session_manager.create_session()
+    policy_evaluator = PolicyEvaluator(bundle=ctx.policy_bundle, config=ctx.config)
+    proxy = CMCPProxy(
+        catalog=ctx.catalog,
+        policy_evaluator=policy_evaluator,
+        session=session,
+        audit_chain=audit_chain,
+        config=ctx.config,
+        attestation_generated_at=ctx.attestation_report.attestation_generated_at,
+        attestation_validity_seconds=ctx.attestation_report.attestation_validity_seconds,
+        attestation_platform=attestation_platform,
+    )
+    # AUTH-001: the token validated in run_startup must reach the server, otherwise
+    # every protected endpoint is reachable unauthenticated.
+    return MCPServer(
+        proxy=proxy,
+        session_manager=session_manager,
+        audit_chain=audit_chain,
+        session=session,
+        bearer_token=ctx.config.bearer_token,
+    )
+
 
 @click.group()
 @click.version_option(__version__, prog_name="cmcp")
@@ -19,16 +71,8 @@ def main() -> None:
               help="Override attestation.enforcement_mode from config")
 def start(config: str, enforcement: str | None) -> None:
     """Start the cMCP Runtime."""
-    from uuid import uuid4
-
     import uvicorn
 
-    from cmcp_runtime.audit.chain import AuditChain
-    from cmcp_runtime.audit.trace_claim import _PROVIDER_MAP
-    from cmcp_runtime.mcp.proxy import CMCPProxy
-    from cmcp_runtime.mcp.server import MCPServer
-    from cmcp_runtime.policy.evaluator import PolicyEvaluator
-    from cmcp_runtime.session.state import SessionState
     from cmcp_runtime.startup import run_startup
 
     ctx = run_startup(config)
@@ -38,24 +82,7 @@ def start(config: str, enforcement: str | None) -> None:
         from cmcp_runtime.config import EnforcementMode
         ctx.config.attestation.enforcement_mode = EnforcementMode(enforcement)
 
-    # Resolve provider string to canonical platform name for Cedar context.
-    # Falls back to the raw provider string if not in the map (e.g. future providers).
-    attestation_platform = _PROVIDER_MAP.get(
-        ctx.attestation_report.provider, ctx.attestation_report.provider
-    )
-
-    session = SessionState(session_id=str(uuid4()))
-    audit_chain = AuditChain(session_id=session.session_id)
-    policy_evaluator = PolicyEvaluator(bundle=ctx.policy_bundle, config=ctx.config)
-    proxy = CMCPProxy(
-        catalog=ctx.catalog,
-        policy_evaluator=policy_evaluator,
-        session=session,
-        audit_chain=audit_chain,
-        config=ctx.config,
-        attestation_platform=attestation_platform,
-    )
-    server = MCPServer(proxy=proxy)
+    server = build_server(ctx)
 
     host, _, port_str = ctx.config.listen_addr.rpartition(":")
     port = int(port_str)

src/cmcp_runtime/errors.py

@@ -38,6 +38,18 @@ class PolicyDeny(CMCPError):
     code = "POLICY_DENY"
     http_status = 403
 
+    def __init__(
+        self,
+        message: str,
+        *,
+        detail: str | None = None,
+        advice: dict[str, str] | None = None,
+    ) -> None:
+        super().__init__(message, detail=detail)
+        # Annotations of the forbid policies that caused this deny — sourced
+        # from the hash-pinned policy bundle, safe to reflect to the caller.
+        self.advice: dict[str, str] = advice or {}
+
 
 class CatalogToolNameCollision(CMCPError):
     code = "CATALOG_TOOL_NAME_COLLISION"
@@ -84,6 +96,16 @@ class TeeFault(CMCPError):
     http_status = 500
 
 
+class UpstreamUnavailable(CMCPError):
+    code = "UPSTREAM_UNAVAILABLE"
+    http_status = 502
+
+
+class UpstreamToolError(CMCPError):
+    code = "UPSTREAM_TOOL_ERROR"
+    http_status = 502
+
+
 class AttestationStale(CMCPError):
     code = "ATTESTATION_STALE"
     http_status = 412