diff --git a/examples/bfsi-demo/cmcp-config.yaml b/examples/bfsi-demo/cmcp-config.yaml index ba433236..11a9ecbf 100644 --- a/examples/bfsi-demo/cmcp-config.yaml +++ b/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 diff --git a/examples/minimal/cmcp-config.yaml b/examples/minimal/cmcp-config.yaml index 4ff9e5ed..8ec9a4ba 100644 --- a/examples/minimal/cmcp-config.yaml +++ b/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 diff --git a/src/cmcp_runtime/audit/chain.py b/src/cmcp_runtime/audit/chain.py index f6b58464..f6a7910c 100644 --- a/src/cmcp_runtime/audit/chain.py +++ b/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 diff --git a/src/cmcp_runtime/audit/store.py b/src/cmcp_runtime/audit/store.py new file mode 100644 index 00000000..79e744d8 --- /dev/null +++ b/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() diff --git a/src/cmcp_runtime/config.py b/src/cmcp_runtime/config.py index 2710f94b..9dc38cb2 100644 --- a/src/cmcp_runtime/config.py +++ b/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, ) diff --git a/src/cmcp_runtime/session/manager.py b/src/cmcp_runtime/session/manager.py index 1a41dbb3..35e5afe2 100644 --- a/src/cmcp_runtime/session/manager.py +++ b/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. diff --git a/src/cmcp_runtime/startup.py b/src/cmcp_runtime/startup.py index 332df611..adc1946e 100644 --- a/src/cmcp_runtime/startup.py +++ b/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,6 +254,26 @@ 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, @@ -259,6 +281,7 @@ def run_startup(config_path: str) -> RuntimeContext: signing_key=signing_key, policy_bundle=policy_store, catalog=catalog, + audit_store=audit_store, spiffe=spiffe_result, nras_appraisal=nras_appraisal, ) diff --git a/tests/unit/test_audit_sqlite.py b/tests/unit/test_audit_sqlite.py new file mode 100644 index 00000000..9ed506f0 --- /dev/null +++ b/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