fix(audit): persist audit chain to SQLite WAL store on every append (AUDIT-001)

examples/bfsi-demo/cmcp-config.yaml

@@ -7,3 +7,4 @@ policy_bundle_path: ./policies/
 catalog_path: ./catalog.json
 listen_addr: "0.0.0.0:8443"
 max_response_size_bytes: 2097152  # 2MB
+audit_db_path: ./audit.db

examples/minimal/cmcp-config.yaml

@@ -3,3 +3,4 @@ attestation:
   enforcement_mode: advisory
 policy_bundle_path: ./policies/
 catalog_path: ./catalog.json
+audit_db_path: ./audit.db

src/cmcp_runtime/audit/chain.py

@@ -7,9 +7,12 @@
 import logging
 from dataclasses import asdict, dataclass, field
 from datetime import UTC, datetime
-from typing import Literal
+from typing import TYPE_CHECKING, Literal
 from uuid import uuid4
 
+if TYPE_CHECKING:
+    from cmcp_runtime.audit.store import SqliteAuditStore
+
 logger = logging.getLogger(__name__)
 
 EntryType = Literal[
@@ -91,9 +94,10 @@ class AuditChain:
     internal hash-chain check still runs.
     """
 
-    def __init__(self, session_id: str) -> None:
+    def __init__(self, session_id: str, store: SqliteAuditStore | None = None) -> None:
         self._session_id = session_id
         self._entries: list[AuditEntry] = []
+        self._store = store
         # AUDIT-002: TEE-anchored chain root.  None until set_tee_anchor() is called.
         self._tee_anchor: str | None = None
         self._append_session_start()
@@ -186,6 +190,8 @@ def append(
             prev_entry_hash=prev_hash,
         )
         entry.entry_hash = entry.compute_hash()
+        if self._store is not None:
+            self._store.append(entry)
         self._entries.append(entry)
         return entry
 

src/cmcp_runtime/audit/store.py

@@ -0,0 +1,87 @@
+"""SQLite-backed audit store — durable persistence for AuditChain entries (AUDIT-001)."""
+
+from __future__ import annotations
+
+import json
+import logging
+import sqlite3
+from dataclasses import asdict
+from pathlib import Path
+
+from cmcp_runtime.audit.chain import AuditEntry
+
+logger = logging.getLogger(__name__)
+
+_CREATE_TABLE = """
+CREATE TABLE IF NOT EXISTS audit_entries (
+    sequence_number   INTEGER NOT NULL,
+    session_id        TEXT    NOT NULL,
+    entry_id          TEXT    NOT NULL PRIMARY KEY,
+    entry_type        TEXT    NOT NULL,
+    entry_hash        TEXT    NOT NULL,
+    prev_entry_hash   TEXT    NOT NULL,
+    payload           TEXT    NOT NULL
+);
+CREATE INDEX IF NOT EXISTS idx_session ON audit_entries (session_id, sequence_number);
+"""
+
+
+class SqliteAuditStore:
+    """
+    Append-only SQLite store for audit chain entries.
+
+    One row per AuditEntry. Entries are written synchronously (WAL mode) before
+    AuditChain.append() returns, so a crash after acknowledgement still has the
+    entry on disk.
+
+    The full entry is serialised as JSON in the `payload` column so the schema
+    is forward-compatible with new AuditEntry fields without a migration.
+    """
+
+    def __init__(self, db_path: Path) -> None:
+        self._db_path = db_path
+        self._conn = sqlite3.connect(str(db_path), check_same_thread=False)
+        self._conn.execute("PRAGMA journal_mode=WAL")
+        self._conn.execute("PRAGMA synchronous=FULL")
+        self._conn.executescript(_CREATE_TABLE)
+        self._conn.commit()
+        logger.info("Audit store opened: path=%s", db_path)
+
+    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()
+
+    def find_orphaned_sessions(self) -> list[str]:
+        """
+        Return session IDs that have a session_start entry but no session_end entry.
+
+        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()]
+
+    def close(self) -> None:
+        self._conn.close()

src/cmcp_runtime/config.py

@@ -55,11 +55,12 @@ class Config:
     listen_addr: str = "0.0.0.0:8443"
     max_response_size_bytes: int = 2 * 1024 * 1024  # 2MB
     policy_reload_interval_seconds: int = 0  # 0 = disabled (POLICY-001)
+    audit_db_path: str = "audit.db"  # AUDIT-001: durable audit chain storage
     dev_mode: bool = False
     bearer_token: str | None = None
 
 
-_KNOWN_TOP_KEYS = {"attestation", "policy_bundle_path", "catalog_path", "listen_addr", "max_response_size_bytes", "policy_reload_interval_seconds"}
+_KNOWN_TOP_KEYS = {"attestation", "policy_bundle_path", "catalog_path", "listen_addr", "max_response_size_bytes", "policy_reload_interval_seconds", "audit_db_path"}
 _KNOWN_ATTEST_KEYS = {"provider", "enforcement_mode", "validity_seconds", "staleness_policy", "expected_measurement"}
 
 
@@ -146,8 +147,10 @@ def load_config(path: str) -> Config:
 
     policy_bundle_path = raw.get("policy_bundle_path", "policy/")
     catalog_path = raw.get("catalog_path", "catalog.json")
+    audit_db_path = raw.get("audit_db_path", "audit.db")
     _check_no_traversal("policy_bundle_path", policy_bundle_path)
     _check_no_traversal("catalog_path", catalog_path)
+    _check_no_traversal("audit_db_path", audit_db_path)
 
     return Config(
         attestation=AttestationConfig(
@@ -162,6 +165,7 @@ def load_config(path: str) -> Config:
         listen_addr=raw.get("listen_addr", "0.0.0.0:8443"),
         max_response_size_bytes=max_bytes,
         policy_reload_interval_seconds=policy_reload_interval,
+        audit_db_path=audit_db_path,
         dev_mode=dev_mode,
         bearer_token=bearer_token,
     )

src/cmcp_runtime/session/manager.py

@@ -61,7 +61,7 @@ def create_session(self) -> tuple[SessionState, AuditChain]:
         """
         session_id = str(uuid4())
         state = SessionState(session_id=session_id)
-        chain = AuditChain(session_id=session_id)
+        chain = AuditChain(session_id=session_id, store=self._ctx.audit_store)
 
         # AUDIT-002: derive a per-session nonce that encodes the chain root so
         # the TEE report binds this specific chain to the attestation evidence.

src/cmcp_runtime/startup.py

@@ -9,6 +9,7 @@
 from typing import Any
 
 from cmcp_runtime.audit.keys import SigningKey
+from cmcp_runtime.audit.store import SqliteAuditStore
 from cmcp_runtime.catalog.loader import ToolCatalog, load_catalog
 from cmcp_runtime.config import Config, load_config
 from cmcp_runtime.errors import (
@@ -48,6 +49,7 @@ class RuntimeContext:
     signing_key: SigningKey
     policy_bundle: PolicyStore
     catalog: ToolCatalog
+    audit_store: SqliteAuditStore | None = None
     spiffe: SpiffeClientResult | None = None
     nras_appraisal: AppraisalResult | None = None
 
@@ -252,13 +254,34 @@ def run_startup(config_path: str) -> RuntimeContext:
     # CMCP_NRAS_API_KEY missing -> skip with warning; any NRAS error -> skip with warning.
     nras_appraisal = try_appraise(attestation_report)
 
+    # Step 5d: open durable audit store and warn on orphaned sessions (AUDIT-001).
+    try:
+        from pathlib import Path as _Path
+        audit_store = SqliteAuditStore(_Path(config.audit_db_path))
+        orphaned = audit_store.find_orphaned_sessions()
+        if orphaned:
+            logger.warning(
+                "AUDIT-001: %d session(s) have no session_end entry in the audit DB — "
+                "gateway may have restarted mid-session. Orphaned session IDs: %s",
+                len(orphaned),
+                orphaned,
+            )
+    except Exception as exc:
+        _fatal(
+            "AUDIT_STORE_UNAVAILABLE",
+            f"Cannot open audit store at '{config.audit_db_path}': {exc}",
+            action="startup_aborted",
+        )
+        sys.exit(1)
+
     return RuntimeContext(
         config=config,
         tee_provider=tee_provider,
         attestation_report=attestation_report,
         signing_key=signing_key,
         policy_bundle=policy_store,
         catalog=catalog,
+        audit_store=audit_store,
         spiffe=spiffe_result,
         nras_appraisal=nras_appraisal,
     )

tests/unit/test_audit_sqlite.py

@@ -0,0 +1,75 @@
+"""Tests for SQLite-backed audit store and durable AuditChain (AUDIT-001)."""
+
+from __future__ import annotations
+
+import pytest
+
+from cmcp_runtime.audit.chain import AuditChain
+from cmcp_runtime.audit.store import SqliteAuditStore
+
+
+@pytest.fixture()
+def store(tmp_path):
+    s = SqliteAuditStore(tmp_path / "audit.db")
+    yield s
+    s.close()
+
+
+def test_store_creates_db_file(tmp_path):
+    db = tmp_path / "audit.db"
+    SqliteAuditStore(db).close()
+    assert db.exists()
+
+
+def test_store_persists_entries(store):
+    chain = AuditChain(session_id="s1", store=store)
+    chain.append("tool_call", tool_name="read_file", policy_decision="allow")
+    # Re-open same DB and verify rows are there
+    store.close()
+    store2 = SqliteAuditStore(store._db_path)
+    cur = store2._conn.execute(
+        "SELECT entry_type FROM audit_entries WHERE session_id = ? ORDER BY sequence_number",
+        ("s1",),
+    )
+    types = [row[0] for row in cur.fetchall()]
+    assert types == ["session_start", "tool_call"]
+    store2.close()
+
+
+def test_no_orphans_after_clean_close(store, tmp_path):
+    chain = AuditChain(session_id="clean", store=store)
+    chain.append("session_end")
+    assert store.find_orphaned_sessions() == []
+
+
+def test_orphan_detected_after_crash(store):
+    AuditChain(session_id="crash-session", store=store)
+    # No session_end appended — simulates crash
+    orphans = store.find_orphaned_sessions()
+    assert "crash-session" in orphans
+
+
+def test_chain_without_store_still_works():
+    chain = AuditChain(session_id="no-store")
+    chain.append("tool_call", tool_name="x", policy_decision="allow")
+    assert chain.verify_chain()
+
+
+def test_chain_with_store_verifies(store):
+    chain = AuditChain(session_id="with-store", store=store)
+    chain.append("tool_call", tool_name="read", policy_decision="allow")
+    chain.append("tool_call", tool_name="write", policy_decision="deny")
+    assert chain.verify_chain()
+    assert chain.length == 3  # session_start + 2 tool_calls
+
+
+def test_multiple_sessions_stored_independently(store):
+    c1 = AuditChain(session_id="sess-a", store=store)
+    c2 = AuditChain(session_id="sess-b", store=store)
+    c1.append("tool_call", tool_name="x", policy_decision="allow")
+    c2.append("tool_call", tool_name="y", policy_decision="deny")
+    c1.append("session_end")
+
+    orphans = store.find_orphaned_sessions()
+    assert "sess-a" not in orphans
+    assert "sess-b" in orphans