From 20e4133f4eb3f83404e2e6f9e76b50d14b0009dd Mon Sep 17 00:00:00 2001 From: Harshith Date: Mon, 19 Oct 2026 01:30:16 +0530 Subject: [PATCH 1/7] fix(store,authz,replay): atomic Cosmos writes, explicit invariants, honest replay, indexed timestamps Findings-catalogue items, each with a test that fails before the change: - COSMOS-2: CosmosForensicStore read, then upserted, so two writers of one span id could both see it absent and the second replaced the first. It now uses create_item (atomic) and compares only on a conflict; a retried create meets its own identical copy and stays idempotent. - AUTHZ-16: the engine asserted rule.condition outside its fail-closed try, so a broken rule crashed the call instead of denying it, and -O dropped the check. The authorization and gateway asserts are explicit checks. - REPLAY-1: with one parent per span, the walk can only re-enter a span through a duplicate, yet that was reported as a cycle. Duplicates are now reported as duplicates; real cycles still show as unreachable spans. - STORE-2: listing and retention ran json_extract over every span under the store lock. Spans carry an indexed ts column; old databases migrate. - TCD-22: the fake Cosmos container ignored the SQL; it now refuses any query but the one it answers, and has create_item. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 17 ++++++++ src/sentinel/authorization/engine.py | 5 ++- src/sentinel/authorization/parser.py | 3 +- src/sentinel/authorization/policy.py | 3 +- src/sentinel/control/mcp_gateway.py | 3 +- src/sentinel/forensics/replay.py | 23 +++++++--- src/sentinel/forensics/store.py | 44 ++++++++++--------- tests/conftest.py | 23 +++++++++- tests/test_authorization_engine.py | 14 ++++++ tests/test_forensic_history.py | 5 ++- tests/test_forensics_replay.py | 12 ++++++ tests/test_forensics_store.py | 64 ++++++++++++++++++++++++++++ 12 files changed, 184 insertions(+), 32 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 13cc7d2..ae85fff 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,23 @@ All notable changes to this project are documented here. Format follows ## [Unreleased] +### Security +- The Cosmos store writes spans with `create_item`, which fails atomically if + the id exists. It read first and then upserted, so two writers of one span + id could both see it absent and the second silently replaced the first. +- The authorization engine denies a rule with no condition instead of failing + an `assert` outside its fail-closed path (and the asserts on the policy and + gateway paths are explicit checks, which `python -O` does not strip). + +### Fixed +- Replay reports a duplicated span id as a duplicate; it was reported as a + cycle, which with one parent per span it cannot be. +- SQLite stores each span's timestamp in an indexed column (existing databases + are migrated in place), so listing traces and retention purges no longer + parse every span's JSON while holding the store lock. +- The test suite's fake Cosmos container refuses queries other than the one it + can answer; it ignored the SQL entirely. + ## [0.2.2] — 2026-10-19 **Upgrade from 0.2.1 or earlier.** Read *Changed (breaking)* first: catalogue diff --git a/src/sentinel/authorization/engine.py b/src/sentinel/authorization/engine.py index 62099db..c1ef147 100644 --- a/src/sentinel/authorization/engine.py +++ b/src/sentinel/authorization/engine.py @@ -235,7 +235,10 @@ def _evaluate( ) -> tuple[bool, str | None]: if rule.deny_always: return True, "deny_always" - assert rule.condition is not None # guaranteed by the compiler + if rule.condition is None: + # The compiler never produces this. An assert here sat outside the + # fail-closed try below (a crash, not a deny) and vanished under -O. + return True, "rule has no condition (fail-closed deny)" try: return rule.condition.evaluate(namespace), None except Exception as exc: # noqa: BLE001 - any failure to evaluate is a deny diff --git a/src/sentinel/authorization/parser.py b/src/sentinel/authorization/parser.py index 08c7c58..efcc115 100644 --- a/src/sentinel/authorization/parser.py +++ b/src/sentinel/authorization/parser.py @@ -103,7 +103,8 @@ def _tokenize(text: str) -> list[_Token]: ) pos = match.end() kind = match.lastgroup - assert kind is not None + if kind is None: # every alternative in _TOKEN_RE is named + raise PolicyParseError(f"unnamed token at position {pos} in {text!r}") if kind == "WS": continue value = match.group() diff --git a/src/sentinel/authorization/policy.py b/src/sentinel/authorization/policy.py index 2ed1f29..cb5feaf 100644 --- a/src/sentinel/authorization/policy.py +++ b/src/sentinel/authorization/policy.py @@ -196,7 +196,8 @@ def _compile(doc: _PolicyDoc) -> CompiledPolicy: if rule.deny_always: compiled.append(CompiledRule(id=rule.id, deny_always=True, condition=None)) else: - assert rule.deny_if is not None # guaranteed by _RuleDoc validator + if rule.deny_if is None: # _RuleDoc's validator guarantees this + raise PolicyLoadError(f"tool {tool_name!r} rule {rule.id!r}: no effect") try: condition = parse_condition(rule.deny_if) except PolicyParseError as exc: diff --git a/src/sentinel/control/mcp_gateway.py b/src/sentinel/control/mcp_gateway.py index 01359a6..ea73cb3 100644 --- a/src/sentinel/control/mcp_gateway.py +++ b/src/sentinel/control/mcp_gateway.py @@ -548,7 +548,8 @@ async def _enter(self) -> SentinelGateway: await self._connect_remote_downstream(urls) else: await self._connect_inmemory_downstream() - assert self._router is not None # every connect path sets it + if self._router is None: # every connect path sets it; never serve without one + raise RuntimeError("no downstream router after connect") # Preflight (health-check + cache schemas) so list_tools can't flake, AND # vet the catalogue itself: a poisoned tool DESCRIPTION subverts the agent # before anything is retrieved, so provenance alone cannot catch it. By diff --git a/src/sentinel/forensics/replay.py b/src/sentinel/forensics/replay.py index 51fc7d3..340fd86 100644 --- a/src/sentinel/forensics/replay.py +++ b/src/sentinel/forensics/replay.py @@ -58,11 +58,23 @@ def get(self, span_id: str) -> Span | None: def build_replay(trace_id: str, spans: list[Span]) -> TraceReplay: """Pure (store-free) replay builder — the testable core of :func:`replay`.""" - ordered = sorted(spans, key=_seq_key) - spans_by_id: dict[str, Span] = {s.span_id: s for s in ordered} - reasons: list[str] = [] + # One span per id. A duplicate used to reach the DFS twice and be reported + # as a cycle, which it is not: with one parent per span, re-entry is only + # possible through a duplicate, and real cycles show up as unreachable. + first: dict[str, Span] = {} + copies: dict[str, int] = {} + for span in sorted(spans, key=_seq_key): + if span.span_id in first: + copies[span.span_id] = copies.get(span.span_id, 1) + 1 + continue + first[span.span_id] = span + for span_id, count in sorted(copies.items()): + reasons.append(f"span id {span_id} appears {count} times (duplicate)") + ordered = list(first.values()) + spans_by_id: dict[str, Span] = dict(first) + # Group real parent->child edges (parent present in this trace), seq-ordered. children: dict[str, list[Span]] = {} roots: list[Span] = [] @@ -89,9 +101,8 @@ def build_replay(trace_id: str, spans: list[Span]) -> TraceReplay: stack: list[Span] = list(reversed(roots)) # reversed → pop in seq order while stack: span = stack.pop() - if span.span_id in visited: - # Re-entry can only happen via a malformed (cyclic) edge; stop branch. - reasons.append(f"cycle/re-entry detected at span {span.span_id}") + if span.span_id in visited: # defensive: duplicates were removed above + reasons.append(f"re-entry at span {span.span_id}") continue visited.add(span.span_id) depth_first.append(span) diff --git a/src/sentinel/forensics/store.py b/src/sentinel/forensics/store.py index e0edacf..26c258e 100644 --- a/src/sentinel/forensics/store.py +++ b/src/sentinel/forensics/store.py @@ -187,26 +187,22 @@ def __init__(self, container: ContainerProxy) -> None: self._container = container async def put(self, span: Span) -> None: - # Immutable-record guard. A point read (single partition, by id) tells us - # whether this span_id already exists; if so we must NOT blindly overwrite: - # identical content → idempotent no-op; different content → raise. - # Only genuinely-new spans are written. upsert (not create) keeps the - # write idempotent under the Cosmos SDK's default retry policy — we add - # no retry of our own. - from azure.cosmos.exceptions import CosmosResourceNotFoundError + # Immutable-record guard, atomic: create_item fails if the id exists, and + # only then is the stored copy compared (identical → no-op, different → + # raise). This used to read first and then upsert, so two writers of the + # same span id could both see "absent" and the second silently replaced + # the first. A create the SDK retried after it had in fact succeeded now + # meets its own identical copy, so the write stays idempotent. + from azure.cosmos.exceptions import CosmosResourceExistsError try: + await self._container.create_item(body=_to_item(span)) + return + except CosmosResourceExistsError: existing_doc = await self._container.read_item( item=span.span_id, partition_key=span.trace_id ) - except CosmosResourceNotFoundError: - existing_doc = None - - if existing_doc is not None: - # Raises on conflict; returns True (duplicate) when identical → no-op. - _ensure_no_conflict(_from_item(existing_doc), span) - return - await self._container.upsert_item(body=_to_item(span)) + _ensure_no_conflict(_from_item(existing_doc), span) async def get_spans(self, trace_id: str) -> list[Span]: query = "SELECT * FROM c WHERE c.trace_id = @trace_id" @@ -290,6 +286,16 @@ def __init__(self, path: str | Path) -> None: self._conn.execute( "CREATE INDEX IF NOT EXISTS spans_by_trace ON spans(trace_id, seq)" ) + # Each span's timestamp in its own column, so listing and retention read + # an index instead of parsing every span's JSON under the store lock. + # Databases created before the column existed are migrated in place. + columns = {row[1] for row in self._conn.execute("PRAGMA table_info(spans)")} + if "ts" not in columns: + self._conn.execute("ALTER TABLE spans ADD COLUMN ts TEXT") + self._conn.execute( + "UPDATE spans SET ts = json_extract(body, '$.timestamp') WHERE ts IS NULL" + ) + self._conn.execute("CREATE INDEX IF NOT EXISTS spans_by_trace_ts ON spans(trace_id, ts)") # Who each trace belongs to. Not forensic evidence (spans are), but # without it a restart cannot tell whose history is whose: every # restored run used to be attributed to the default tenant, which hid @@ -342,8 +348,8 @@ def _put_blocking(self, span: Span, body: str) -> None: # Plain INSERT: the check above already refused a conflicting span, and # OR REPLACE would let anything that slipped past it overwrite evidence. self._conn.execute( - "INSERT INTO spans (trace_id, span_id, seq, body) VALUES (?, ?, ?, ?)", - (span.trace_id, span.span_id, span.seq, body), + "INSERT INTO spans (trace_id, span_id, seq, body, ts) VALUES (?, ?, ?, ?, ?)", + (span.trace_id, span.span_id, span.seq, body, json.loads(body)["timestamp"]), ) async def get_spans(self, trace_id: str) -> list[Span]: @@ -372,7 +378,7 @@ def _list_traces_blocking(self) -> list[str]: # first, not the most recent. cursor = self._conn.execute( "SELECT trace_id FROM spans GROUP BY trace_id " - "ORDER BY MAX(julianday(json_extract(body, '$.timestamp'))) DESC, trace_id" + "ORDER BY MAX(julianday(ts)) DESC, trace_id" ) return [row[0] for row in cursor.fetchall()] @@ -411,7 +417,7 @@ def _purge_blocking(self, cutoff: datetime, keep: frozenset[str]) -> list[str]: stamp = cutoff.astimezone(UTC).strftime("%Y-%m-%dT%H:%M:%S.%fZ") rows = self._conn.execute( "SELECT trace_id FROM spans GROUP BY trace_id " - "HAVING MAX(julianday(json_extract(body, '$.timestamp'))) < julianday(?)", + "HAVING MAX(julianday(ts)) < julianday(?)", (stamp,), ).fetchall() doomed = [row[0] for row in rows if row[0] not in keep] diff --git a/tests/conftest.py b/tests/conftest.py index e680412..ae757e9 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -93,6 +93,16 @@ def __init__(self) -> None: # partition_key (trace_id) -> { id (span_id) -> document } self._partitions: dict[str, dict[str, dict[str, Any]]] = {} + async def create_item(self, *, body: dict[str, Any]) -> dict[str, Any]: + """Atomic: fails if the id already exists in the partition, as Cosmos does.""" + partition = self._partitions.setdefault(body["trace_id"], {}) + if body["id"] in partition: + from azure.cosmos.exceptions import CosmosResourceExistsError + + raise CosmosResourceExistsError(message=f"{body['id']} exists") + partition[body["id"]] = copy.deepcopy(body) + return copy.deepcopy(body) + async def upsert_item(self, *, body: dict[str, Any]) -> dict[str, Any]: partition_key = body["trace_id"] item_id = body["id"] @@ -118,7 +128,12 @@ def query_items( ) -> AsyncIterator[dict[str, Any]]: # Single-partition query: return exactly the requested partition's docs. # That is equivalent to the store's `WHERE c.trace_id = @trace_id`, since - # the partition key IS trace_id. + # the partition key IS trace_id. Any other query is refused rather than + # answered as if it were that one (the fake used to ignore the SQL). + expected = "SELECT * FROM c WHERE c.trace_id = @trace_id" + bound = {p["name"]: p["value"] for p in parameters or []} + if query != expected or bound != {"@trace_id": partition_key}: + raise NotImplementedError(f"fake Cosmos container cannot answer {query!r} {bound!r}") docs = list(self._partitions.get(partition_key or "", {}).values()) async def _aiter() -> AsyncIterator[dict[str, Any]]: @@ -128,6 +143,12 @@ async def _aiter() -> AsyncIterator[dict[str, Any]]: return _aiter() +@pytest.fixture() +def fake_cosmos_container() -> type[FakeAsyncCosmosContainer]: + """The fake container class, for tests that need to subclass it.""" + return FakeAsyncCosmosContainer + + @pytest.fixture(params=["memory", "cosmos", "sqlite"]) def store( request: pytest.FixtureRequest, tmp_path_factory: pytest.TempPathFactory diff --git a/tests/test_authorization_engine.py b/tests/test_authorization_engine.py index 7aeff7b..a2c79ef 100644 --- a/tests/test_authorization_engine.py +++ b/tests/test_authorization_engine.py @@ -172,3 +172,17 @@ def test_compound_deny_if_allows_clean_lineage() -> None: ) ) assert result.allowed is True + + +def test_a_rule_without_a_condition_denies_instead_of_crashing() -> None: + """The engine asserted this outside its fail-closed path (review AUTHZ-16).""" + from sentinel.authorization.policy import CompiledPolicy, CompiledRule + + broken = CompiledPolicy( + policy_version=1, + tools={"t": (CompiledRule(id="r", deny_always=False, condition=None),)}, + ) + decision = AuthorizationEngine(broken).authorize( + ToolCall(name="t", provenance=frozenset({"USER"})) + ) + assert decision.decision == "DENY" and decision.matched_rule_id == "r" diff --git a/tests/test_forensic_history.py b/tests/test_forensic_history.py index 2d79758..1bf2d9a 100644 --- a/tests/test_forensic_history.py +++ b/tests/test_forensic_history.py @@ -111,8 +111,9 @@ def _age(db: Path, trace_id: str, days: float) -> None: stamp = (datetime.now(UTC) - timedelta(days=days)).strftime("%Y-%m-%dT%H:%M:%S.%fZ") with sqlite3.connect(db) as conn: conn.execute( - "UPDATE spans SET body = json_set(body, '$.timestamp', ?) WHERE trace_id = ?", - (stamp, trace_id), + # Both copies: the store indexes the timestamp in its own column. + "UPDATE spans SET body = json_set(body, '$.timestamp', ?), ts = ? WHERE trace_id = ?", + (stamp, stamp, trace_id), ) diff --git a/tests/test_forensics_replay.py b/tests/test_forensics_replay.py index 7f29162..99af4e7 100644 --- a/tests/test_forensics_replay.py +++ b/tests/test_forensics_replay.py @@ -121,3 +121,15 @@ def test_replay_is_cycle_safe(make_span: Callable[..., Span]) -> None: assert rep.malformed is True assert rep.roots == [] assert any("unreachable" in r for r in rep.malformed_reasons) + + +def test_a_duplicate_span_is_reported_as_a_duplicate_not_a_cycle( + make_span: Callable[..., Span], +) -> None: + """With one parent per span, only a duplicate can re-enter the walk (REPLAY-1).""" + root = make_span(seq=0) + rep = build_replay(TRACE_A, [root, root]) + assert [s.span_id for s in rep.ordered] == [root.span_id] + assert rep.malformed + assert any("duplicate" in r for r in rep.malformed_reasons) + assert not any("cycle" in r for r in rep.malformed_reasons) diff --git a/tests/test_forensics_store.py b/tests/test_forensics_store.py index 62516e3..bc6c516 100644 --- a/tests/test_forensics_store.py +++ b/tests/test_forensics_store.py @@ -285,3 +285,67 @@ async def test_in_memory_spans_cannot_be_rewritten_through_a_reference( got.payload.arguments["after_get"] = "y" # type: ignore[attr-defined] stored = (await store.get_spans(span.trace_id))[0] assert list(stored.payload.arguments) == ["a"] # type: ignore[attr-defined] + + +async def test_two_cosmos_writers_of_one_span_id_cannot_both_win( + make_span: Callable[..., Span], fake_cosmos_container: type +) -> None: + """Read-then-upsert let both see "absent"; the second silently replaced the first.""" + from sentinel.forensics.events import ToolCallProposed + from sentinel.forensics.store import CosmosForensicStore + + class SlowReads(fake_cosmos_container): # type: ignore[misc, valid-type] + async def read_item(self, *, item: str, partition_key: str) -> dict[str, object]: + # Answer first, then yield, as a network round trip does: the other + # writer runs between this check and the write that follows it. + try: + found = await super().read_item(item=item, partition_key=partition_key) + except Exception: + await asyncio.sleep(0) + raise + await asyncio.sleep(0) + return found + + container = SlowReads() + store = CosmosForensicStore(container) # type: ignore[arg-type] + first = make_span(seq=0, payload=ToolCallProposed(tool_name="a", arguments={})) + second = first.model_copy(update={"payload": ToolCallProposed(tool_name="b", arguments={})}) + results = await asyncio.gather(store.put(first), store.put(second), return_exceptions=True) + assert any(isinstance(r, ForensicIntegrityError) for r in results), results + stored = await store.get_spans(first.trace_id) + assert [s.payload.tool_name for s in stored] == ["a"] # type: ignore[attr-defined] + + +async def test_a_database_from_before_the_timestamp_column_is_migrated( + tmp_path: Path, make_span: Callable[..., Span] +) -> None: + """Listing and retention parsed every span's JSON under the lock (STORE-2).""" + import json as _json + import sqlite3 + + db = tmp_path / "old.db" + span = make_span(seq=0) + with sqlite3.connect(db) as conn: # the schema as it was + conn.execute( + "CREATE TABLE spans (trace_id TEXT NOT NULL, span_id TEXT NOT NULL PRIMARY KEY," + " seq INTEGER NOT NULL, body TEXT NOT NULL)" + ) + conn.execute( + "INSERT INTO spans VALUES (?, ?, ?, ?)", + (span.trace_id, span.span_id, span.seq, _json.dumps(span.model_dump(mode="json"))), + ) + store = SqliteForensicStore(db) + assert await store.list_trace_ids() == [span.trace_id] + later = make_span(trace_id="b" * 32, seq=0, span_id="2" * 16) + await store.put(later) + assert await store.list_trace_ids() == ["b" * 32, span.trace_id] + store.close() + with sqlite3.connect(db) as conn: + assert conn.execute("SELECT COUNT(*) FROM spans WHERE ts IS NULL").fetchone()[0] == 0 + plan = " ".join( + str(r) for r in conn.execute( + "EXPLAIN QUERY PLAN SELECT trace_id FROM spans GROUP BY trace_id" + " ORDER BY MAX(julianday(ts))" + ) + ) + assert "spans_by_trace_ts" in plan From 2629fd4693b7fce9a53d91418c1fd43072097d38 Mon Sep 17 00:00:00 2001 From: Harshith Date: Mon, 19 Oct 2026 01:36:07 +0530 Subject: [PATCH 2/7] fix(cosmos,gateway,manager): Cosmos can start, loud SDK dependency, no wasted restore Findings-catalogue items, each with a test that fails before the change: - DOC-1 (and a real defect behind it): both Cosmos builders used the async Azure clients, which need aiohttp, and SENTINEL does not install it. So setting AZURE_COSMOS_ENDPOINT failed at startup with a bare import error; Cosmos persistence could not run. New extra sentinel-prox[cosmos]; without it the error names the extra. The unused duplicate create_cosmos_store (whose docstring promised wiring "in a later phase") is removed. - GW-18: releasing ended sessions reads the SDK's private _server_instances. Losing it meant no releases at all, so every tenant would hit its session cap for good, silently. Now one accessor, an error logged once if it is missing, and a test pinning the SDK attribute. - CM-10: restore guessed each trace's scenario from its user input, which is redacted before storage, so the guess never matched, and it loaded every span of every trace at startup to make it. Removed; the scenario comes from the ownership row. - DCC-23 / GW-20: removed build_web_server and _expected_token (unused); corrected a comment that still said trust is keyed by agent alone. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 9 ++++++ README.md | 2 +- pyproject.toml | 2 ++ src/sentinel/control/app.py | 18 +++--------- src/sentinel/control/manager.py | 41 ++++++++++----------------- src/sentinel/control/mcp_gateway.py | 29 ++++++++++++++++--- src/sentinel/forensics/store.py | 29 ++----------------- src/sentinel/toolservers/web_fetch.py | 25 +--------------- tests/test_azure_wiring.py | 18 ++++++++++++ tests/test_forensic_history.py | 21 ++++++++++++++ tests/test_mcp_session_ownership.py | 27 ++++++++++++++++++ 11 files changed, 125 insertions(+), 96 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ae85fff..8481571 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,15 @@ All notable changes to this project are documented here. Format follows gateway paths are explicit checks, which `python -O` does not strip). ### Fixed +- Cosmos DB persistence could not start: its async Azure clients need aiohttp, + which is not installed, and setting `AZURE_COSMOS_ENDPOINT` failed at startup + with a bare import error. Install `sentinel-prox[cosmos]`; without it the + error now says so. An unused duplicate Cosmos factory was removed. +- Restoring the run index after a restart no longer loads every span of every + trace to guess a scenario label from redacted input (it never matched). +- The gateway logs an error if the MCP SDK stops exposing the session index it + reads to release ended sessions, instead of silently never releasing them; + a test pins that SDK attribute. - Replay reports a duplicated span id as a duplicate; it was reported as a cycle, which with one parent per span it cannot be. - SQLite stores each span's timestamp in an indexed column (existing databases diff --git a/README.md b/README.md index 6ba9af1..2d3f821 100644 --- a/README.md +++ b/README.md @@ -78,7 +78,7 @@ protects you from. | **Quarantine is recorded, not enforced** (default) | The trust score never recovers, so agents making only allowed calls cross the threshold within tens of calls. By default that crossing is logged, not acted on. `SENTINEL_ENFORCE_QUARANTINE=1` enforces it, for every agent that shares a credential. | | **Downstream servers: HTTP only** | No stdio servers, no per-server credentials or OAuth, one shared session per server, and no reconnect. Most published MCP servers are stdio. | | **State lives in one process** | Trust scores, MCP sessions, the event feed, the run index and policies replaced through `POST /tenants/{t}/policy` are in memory. A restart forgets API policy reloads (put durable policy in `SENTINEL_POLICY_FILE` / `SENTINEL_TENANT_POLICIES`), and two replicas share none of it, which is why the Azure template runs one. | -| **Cosmos persistence is partial** | With Cosmos DB, spans are stored and replayable, but run ownership, the run index after a restart, and retention purging exist only for the default SQLite store. | +| **Cosmos persistence is partial** | It needs `pip install 'sentinel-prox[cosmos]'` (for aiohttp). With Cosmos DB, spans are stored and replayable, but run ownership, the run index after a restart, and retention purging exist only for the default SQLite store. | | **Some state is visible across tenants** | Event ids on `/events` are one global counter, so a tenant can infer how busy the deployment is. `/downstream` and `/capabilities` describe the whole deployment to any authenticated caller. | | **Calls the SDK rejects are not recorded** | Arguments that fail a tool's input schema are refused by the MCP SDK before SENTINEL sees the call, so they leave no forensic record. | diff --git a/pyproject.toml b/pyproject.toml index 2af4ef2..76f6c45 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -83,6 +83,8 @@ Demo = "https://sentinel-i63x.onrender.com" sentinel = "sentinel.cli:main" [project.optional-dependencies] +# Cosmos DB persistence uses the async Azure clients, which need aiohttp. +cosmos = ["aiohttp>=3.9,<4"] dev = [ "pytest>=8.3", "pytest-asyncio>=0.24", diff --git a/src/sentinel/control/app.py b/src/sentinel/control/app.py index 90151f7..18d7a66 100644 --- a/src/sentinel/control/app.py +++ b/src/sentinel/control/app.py @@ -179,16 +179,6 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: self._on_close() -def _expected_token() -> str | None: - """The bearer token every endpoint expects (None when auth is disabled). - - Resolved fresh per request rather than cached, so changing - ``SENTINEL_API_TOKEN`` and restarting takes effect. Deliberately tiny: the - auth posture is a single chokepoint. - """ - return get_settings().api_token - - def _anonymous_allowed() -> bool: """Whether running with no authentication has been explicitly requested.""" return get_settings().allow_anonymous @@ -448,10 +438,10 @@ def visible(trace: str) -> bool: def _owned_agent(agent_id: str, request: Request) -> None: """Refuse an agent that has never acted in the caller's tenant. - Trust scores are keyed by agent, not by tenant, so without this one - tenant could read — or RESET — another's quarantine. Reset is the - sharper end: clearing a quarantine you do not own re-enables an agent - someone else's policy stopped. 404 for the same reason as runs. + Trust is kept per (tenant, agent), so another tenant's agent of the + same id is a different score already. This check adds that a scoped + caller cannot probe agent ids its tenant never ran: 404, for the same + reason as runs. """ tenant = _scope(request) if tenant is None: diff --git a/src/sentinel/control/manager.py b/src/sentinel/control/manager.py index 416fdd7..08d9729 100644 --- a/src/sentinel/control/manager.py +++ b/src/sentinel/control/manager.py @@ -16,6 +16,7 @@ import asyncio import functools +import importlib.util import logging import os import time @@ -48,7 +49,6 @@ AgentQuarantined, AuthorizationDecided, InjectionScanned, - InputReceived, ToolBlocked, ) from sentinel.forensics.replay import TraceReplay, replay @@ -64,21 +64,6 @@ from sentinel.trust.scorer import TrustScorer -def _infer_scenario(user_input: str) -> str: - """Best-effort scenario classification for a restored run record. - - Only used for display in the run index. Authoritative replay is from the - forensic spans; this just guesses a label from the user prompt because - the scenario name isn't itself stored as a span. - """ - text = user_input.lower() - if "bulk-delete" in text or "permanently remove" in text: - return "privilege" if "permanently remove" in text else "trust-collapse" - if "pricing page" in text or "research" in text: - return "hero-obvious" - return "restored" - - def _build_cosmos_container(settings: Settings) -> Any: # noqa: ANN401 - SDK container """Build the live async Cosmos container client (AZURE MODE). @@ -87,11 +72,20 @@ def _build_cosmos_container(settings: Settings) -> Any: # noqa: ANN401 - SDK co deploy time). Constructed lazily so the azure SDKs are imported only when Cosmos is actually configured. Not exercised offline (needs a live account). """ + # The async Cosmos client and credential need aiohttp, which SENTINEL does + # not install by default. Without this check, setting AZURE_COSMOS_ENDPOINT + # failed at startup with a bare "aiohttp package is not installed". + if importlib.util.find_spec("aiohttp") is None: + raise RuntimeError( + "Cosmos DB persistence needs aiohttp: pip install 'sentinel-prox[cosmos]' " + "(or unset AZURE_COSMOS_ENDPOINT to use the default SQLite store)" + ) from azure.cosmos.aio import CosmosClient # noqa: PLC0415 - Azure-only path from azure.identity.aio import DefaultAzureCredential # noqa: PLC0415 endpoint = settings.azure_cosmos_endpoint - assert endpoint is not None # precondition: only called when configured + if endpoint is None: # only called when configured + raise RuntimeError("AZURE_COSMOS_ENDPOINT is not set") client = CosmosClient(endpoint, DefaultAzureCredential()) database = client.get_database_client(settings.azure_cosmos_database) return database.get_container_client(settings.azure_cosmos_container) @@ -588,17 +582,12 @@ async def restore_persisted_runs(self) -> int: for trace_id in trace_ids: if trace_id in self._runs: continue - spans = await self._inner_store.get_spans(trace_id) - if not spans: - continue + # The scenario and agent come from the ownership row. They used to be + # guessed from each trace's user input, which is redacted before it + # is stored, so the guess never matched, and every span of every + # trace was loaded at startup to make it. scenario = "restored" agent_id = f"agent-{trace_id[:8]}" - for s in spans: - if isinstance(s.payload, InputReceived): - # The user input doesn't tell us the scenario name, but the - # control plane only uses scenario for display. - scenario = _infer_scenario(s.payload.content) - break # Ownership comes from what was recorded when the run started. A # trace with no record predates that (or was written by something # else) and is attributed to NOBODY rather than to the default diff --git a/src/sentinel/control/mcp_gateway.py b/src/sentinel/control/mcp_gateway.py index ea73cb3..f86670b 100644 --- a/src/sentinel/control/mcp_gateway.py +++ b/src/sentinel/control/mcp_gateway.py @@ -432,6 +432,7 @@ def __init__( self._transport_sessions: dict[str, _ReservedTransportSession] = {} self._pending_transport_sessions: dict[str, int] = {} self._transport_session_lock = asyncio.Lock() + self._warned_no_session_index = False self._front = self._build_front_server() # stateless=False: keep one ServerSession per MCP session so a session's # provenance frontier persists across its successive tool calls. @@ -792,16 +793,36 @@ def _record_session_owner(self, session_id: str, owner: str, *, tenant: str) -> # Forget sessions the transport no longer has. Only ever drops ids the # SDK has already discarded, so no live session loses its owner (which # would lock its creator out, since an unowned id is refused). - live = getattr(self._session_manager, "_server_instances", None) - if isinstance(live, dict): + live = self._live_session_ids() + if live is not None: for session in [s for s in self._session_owners if s not in live]: del self._session_owners[session] self._transport_sessions.pop(session, None) + def _live_session_ids(self) -> frozenset[str] | None: + """The session ids the SDK still holds, or None if it cannot be read. + + The SDK exposes no public way to ask, so this reads the private + ``_server_instances``. Without it, ended sessions are never released: + the owner table grows and each tenant hits its session cap for good. + tests/test_mcp_session_ownership.py pins the attribute, so an SDK that + drops it fails CI; at runtime the loss is logged rather than silent. + """ + live = getattr(self._session_manager, "_server_instances", None) + if isinstance(live, dict): + return frozenset(live) + if not self._warned_no_session_index: + self._warned_no_session_index = True + _LOG.error( + "the MCP SDK no longer exposes its session index; ended sessions " + "cannot be released, so tenants will reach their session cap" + ) + return None + def _prune_transport_sessions(self) -> None: """Drop reservations only after the SDK has ended the transport session.""" - live = getattr(self._session_manager, "_server_instances", None) - if not isinstance(live, dict): + live = self._live_session_ids() + if live is None: return for session_id in [sid for sid in self._transport_sessions if sid not in live]: self._transport_sessions.pop(session_id, None) diff --git a/src/sentinel/forensics/store.py b/src/sentinel/forensics/store.py index 26c258e..0c13e92 100644 --- a/src/sentinel/forensics/store.py +++ b/src/sentinel/forensics/store.py @@ -33,7 +33,6 @@ if TYPE_CHECKING: from azure.cosmos.aio import ContainerProxy - from sentinel.config import Settings _T = TypeVar("_T") @@ -180,7 +179,8 @@ class CosmosForensicStore: Takes an already-constructed async ``ContainerProxy`` so the class is unit- testable against a fake that honors the real azure-cosmos async contract (``await upsert_item(body=...)``; ``query_items(...)`` returns an async - iterable). The :func:`create_cosmos_store` factory wires the real client. + iterable). The run manager builds the real client (needs the ``cosmos`` + extra, for aiohttp). """ def __init__(self, container: ContainerProxy) -> None: @@ -219,31 +219,6 @@ async def get_spans(self, trace_id: str) -> list[Span]: return _by_seq(spans) -def create_cosmos_store(settings: Settings) -> CosmosForensicStore: - """Build a :class:`CosmosForensicStore` from settings (Azure mode only). - - Lazy-imports the Azure SDKs so DEMO MODE never pays for them and the package - builds on machines without Azure installed. Auth is identity-based - (``DefaultAzureCredential`` → system-assigned managed identity in prod, - per §2); no keys in code. The caller owns shutdown via - :meth:`CosmosForensicStore`'s underlying client — wired by the proxy in a - later phase. Fails loudly if the endpoint is unset rather than guessing. - """ - if not settings.azure_cosmos_endpoint: - raise ValueError( - "AZURE_COSMOS_ENDPOINT is not set; cannot build a Cosmos store. " - "(In DEMO MODE use InMemoryForensicStore instead.)" - ) - - from azure.cosmos.aio import CosmosClient - from azure.identity.aio import DefaultAzureCredential - - credential = DefaultAzureCredential() - client = CosmosClient(url=settings.azure_cosmos_endpoint, credential=credential) - database = client.get_database_client(settings.azure_cosmos_database) - container = database.get_container_client(settings.azure_cosmos_container) - return CosmosForensicStore(container) - class SqliteForensicStore: """File-backed forensic store — DEMO MODE persistence across restarts. diff --git a/src/sentinel/toolservers/web_fetch.py b/src/sentinel/toolservers/web_fetch.py index f1ebac9..ec82a0a 100644 --- a/src/sentinel/toolservers/web_fetch.py +++ b/src/sentinel/toolservers/web_fetch.py @@ -35,13 +35,10 @@ import socket from collections.abc import Callable, Sequence from dataclasses import dataclass -from typing import TYPE_CHECKING, Final +from typing import Final import httpx -if TYPE_CHECKING: - from mcp.server.fastmcp import FastMCP - _USER_AGENT: Final[str] = "SENTINEL-web-fetch/1.0 (+https://github.com/Harshith029/SENTINEL)" _TEXTUAL_HINTS: Final[tuple[str, ...]] = ( "text/", @@ -277,23 +274,3 @@ async def _fetch( finally: if owns_client: await client.aclose() - - -def build_web_server() -> FastMCP: - """A FastMCP ``web`` server whose ``web_fetch`` performs a REAL fetch. - - Drop-in replacement for the demo's mock ``web`` server: same tool name and - signature, so it sits behind the proxy unchanged. Imported lazily so the - package has no hard dependency on ``mcp.server.fastmcp`` at import time. - """ - from mcp.server.fastmcp import FastMCP - - web = FastMCP("web") - - @web.tool() - async def web_fetch(url: str) -> str: - """Fetch a web page over HTTP(S) and return its text (guarded).""" - result = await fetch_url(url) - return result.as_tool_output() - - return web diff --git a/tests/test_azure_wiring.py b/tests/test_azure_wiring.py index 80faee5..7f7a6db 100644 --- a/tests/test_azure_wiring.py +++ b/tests/test_azure_wiring.py @@ -72,3 +72,21 @@ def test_an_explicit_azure_shield_is_not_downgraded_by_demo_mode() -> None: azure_content_safety_key="k", ) assert InputShield.from_settings(auto).backend == "local" + + +def test_cosmos_without_aiohttp_fails_with_the_fix_named(monkeypatch) -> None: # type: ignore[no-untyped-def] + """It failed at startup with a bare "aiohttp package is not installed".""" + import importlib.util + + import pytest + + from sentinel.control.manager import _build_cosmos_container + + real_find_spec = importlib.util.find_spec + + def no_aiohttp(name: str, *args: object) -> object: + return None if name == "aiohttp" else real_find_spec(name, *args) # type: ignore[arg-type] + + monkeypatch.setattr(importlib.util, "find_spec", no_aiohttp) + with pytest.raises(RuntimeError, match=r"sentinel-prox\[cosmos\]"): + _build_cosmos_container(Settings(demo_mode=False, azure_cosmos_endpoint="https://x.example")) diff --git a/tests/test_forensic_history.py b/tests/test_forensic_history.py index 1bf2d9a..e4ccba6 100644 --- a/tests/test_forensic_history.py +++ b/tests/test_forensic_history.py @@ -196,3 +196,24 @@ async def test_a_store_passed_in_is_left_open_for_its_owner(tmp_path: Path) -> N await manager.aclose() assert await store.list_trace_ids() == [] # still usable: not ours to close store.close() + + +async def test_restoring_the_run_index_does_not_load_every_trace() -> None: + """Restore read every span of every trace to guess a label it never found (CM-10).""" + run_id, _ = await _acme_run() + manager = RunManager() + store = manager._inner_store # noqa: SLF001 + loads: list[str] = [] + real = store.get_spans + + async def counting(trace_id: str): # type: ignore[no-untyped-def] + loads.append(trace_id) + return await real(trace_id) + + store.get_spans = counting # type: ignore[method-assign] + try: + assert await manager.restore_persisted_runs() >= 1 + assert loads == [] + assert manager.get_run(run_id).scenario == "hero-obvious" # type: ignore[union-attr] + finally: + await manager.aclose() diff --git a/tests/test_mcp_session_ownership.py b/tests/test_mcp_session_ownership.py index 2fba02e..08eccec 100644 --- a/tests/test_mcp_session_ownership.py +++ b/tests/test_mcp_session_ownership.py @@ -390,3 +390,30 @@ class _Session: # weak-referenceable stand-in for the SDK's ServerSession assert "retired" in message and "new session" in message assert "Retry shortly" not in message assert "capacity" in gateway._no_proxy_message(_Session()) # noqa: SLF001 + + +def test_the_sdk_still_exposes_the_session_index_the_gateway_reads() -> None: + """GW-18: owner and reservation release read the SDK's private session index. + + If an SDK upgrade drops it, ended sessions are never released and tenants + reach their cap for good. This fails first, in CI, instead. + """ + from mcp.server.lowlevel import Server + from mcp.server.streamable_http_manager import StreamableHTTPSessionManager + + manager = StreamableHTTPSessionManager(app=Server("probe")) + assert isinstance(getattr(manager, "_server_instances", None), dict) + + +async def test_a_missing_session_index_is_logged_not_silent( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + async with _gateway(monkeypatch) as (_client, gateway, _store): + real = gateway._session_manager # noqa: SLF001 + gateway._session_manager = object() # type: ignore[assignment] # noqa: SLF001 + try: + with caplog.at_level("ERROR", logger="sentinel.control.mcp_gateway"): + assert gateway._live_session_ids() is None # noqa: SLF001 + finally: + gateway._session_manager = real # noqa: SLF001 + assert "session index" in caplog.text From a603abedcb0ee9c2ed2565b6294e660d517bf502 Mon Sep 17 00:00:00 2001 From: Harshith Date: Mon, 19 Oct 2026 01:40:01 +0530 Subject: [PATCH 3/7] fix(demo,dashboard): baseline matches the run's driver; kill chain and gauge from data Findings-catalogue items; checked in a browser where noted: - DEMO-04: the before/after baseline always ran the scripted transcript, so with a live model the dashboard compared a model against a script. It now uses the run's driver choice and web mode; the direct client gained list_tools so a live-model driver can run there. Test fails before. - DEMO-09: the kill chain named web_fetch and block-untrusted-origin, and the verdict claimed "the exfiltration would have succeeded", whatever the run did. Both come from the recorded steps now. In a browser, the privilege scenario shows no untrusted input, a clean lineage, and delete_record refused by rule never; the hero names web_fetch. - DEMO-26: the trust gauge hard-coded a threshold of 40. The configured value is in /capabilities (and the trust endpoint) and the dashboard reads it. - DEMO-16: the examples said an external /mcp run appears in the live view; it is listed in the history tab. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 8 ++++++++ examples/README.md | 5 +++-- examples/connect_over_http.py | 5 +++-- examples/run_live_agent.py | 3 ++- src/sentinel/control/manager.py | 13 ++++++++++++- src/sentinel/control/static/index.html | 18 +++++++++++------- src/sentinel/demo/driver.py | 8 ++++++++ src/sentinel/trust/scorer.py | 4 ++++ tests/test_agent_default_driver.py | 20 ++++++++++++++++++++ tests/test_phase8_enterprise.py | 15 +++++++++++++++ 10 files changed, 86 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8481571..47b9d81 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,14 @@ All notable changes to this project are documented here. Format follows gateway paths are explicit checks, which `python -O` does not strip). ### Fixed +- The before/after baseline uses the same driver (scripted or live model) and + web mode as the protected run; it was always the scripted transcript. +- The dashboard's kill chain and verdict come from the run's data. They named + `web_fetch`, `block-untrusted-origin` and "the exfiltration" whatever the + run did. Its trust gauge uses the configured quarantine threshold (now in + `/capabilities` and the trust endpoint) instead of a hard-coded 40. +- The examples no longer say an external `/mcp` run appears in the live view; + it is listed in the history tab. - Cosmos DB persistence could not start: its async Azure clients need aiohttp, which is not installed, and setting `AZURE_COSMOS_ENDPOINT` failed at startup with a bare import error. Install `sentinel-prox[cosmos]`; without it the diff --git a/examples/README.md b/examples/README.md index 1198606..83fae5a 100644 --- a/examples/README.md +++ b/examples/README.md @@ -70,8 +70,9 @@ python examples/connect_over_http.py # a real mcp client drives the hero atta SENTINEL-specific code** — to connect to `/mcp`, list tools, and run the hero kill chain. The poisoned-lineage `send_email` comes back `isError=True` with `SENTINEL blocked …`. Swap in your own client the same way; to your agent, -SENTINEL is just an MCP tool server. Each connection is its own `live-mcp` trace -on the dashboard. +SENTINEL is just an MCP tool server. Each connection is its own `live-mcp` run, +listed in the dashboard's history tab (the live view follows dashboard-launched +runs only). For a **real LLM agent loop** (the model decides; SENTINEL is the backstop): diff --git a/examples/connect_over_http.py b/examples/connect_over_http.py index c37e050..756bfce 100644 --- a/examples/connect_over_http.py +++ b/examples/connect_over_http.py @@ -22,8 +22,9 @@ What you'll see: tool discovery and the legit read calls succeed, and the poisoned-lineage ``send_email`` to the attacker is BLOCKED at the MCP boundary with a clean error — the same result as the in-process demo, now over real HTTP. -Open the dashboard (http://127.0.0.1:8765) to watch the live run appear as a -``live-mcp`` trace. +The session is recorded as a ``live-mcp`` run: open the dashboard +(http://127.0.0.1:8765), go to the history tab and select it to replay its trace. +The live view follows runs launched from the dashboard itself. """ from __future__ import annotations diff --git a/examples/run_live_agent.py b/examples/run_live_agent.py index 368c3ce..ff784c1 100644 --- a/examples/run_live_agent.py +++ b/examples/run_live_agent.py @@ -21,7 +21,8 @@ # ...or with NO credentials, using the offline susceptible stand-in model: python examples/run_live_agent.py --offline -Watch http://localhost:8765 — the run appears live as a `live-mcp` trace. +The session is recorded as a `live-mcp` run; find it in the dashboard's history +tab (http://localhost:8765). The live view follows dashboard-launched runs only. Exit status: 0 when SENTINEL blocked at least one action, 4 when nothing was blocked, 3 when no live model is available, 2 when the run failed. diff --git a/src/sentinel/control/manager.py b/src/sentinel/control/manager.py index 08d9729..a9ca31c 100644 --- a/src/sentinel/control/manager.py +++ b/src/sentinel/control/manager.py @@ -507,13 +507,18 @@ async def run_baseline(self, build: ScenarioBuild) -> dict[str, Any]: the dashboard — shows the two outboxes side-by-side so a viewer can FEEL the win, not just read the BLOCKED banner. """ + # The same driver choice and web mode as the protected run. The baseline + # used to be the scripted transcript always, so with a live model the + # dashboard compared a model's run against a script. (A live model may + # still behave differently on the second run; that is the model.) result = await run_demo_session( - build.driver, + self._select_driver(build), user_input=build.user_input, demo_mode=self._demo_mode, agent_id="baseline-no-sentinel", extra_pages=build.extra_pages, skip_sentinel=True, + real_web=self._settings.real_web_fetch, ) return { "outbox": result.outbox, @@ -624,6 +629,7 @@ def trust(self, agent_id: str, tenant: str = DEFAULT_TENANT) -> dict[str, Any]: "quarantined": self._scorer.is_quarantined(agent_id, tenant=tenant), "below_threshold": self._scorer.crossed_threshold(agent_id, tenant=tenant), "quarantine_enforced": self._scorer.enforces_quarantine, + "quarantine_threshold": self._scorer.quarantine_threshold, } def reset(self, agent_id: str, tenant: str = DEFAULT_TENANT) -> dict[str, Any]: @@ -646,6 +652,11 @@ def capabilities(self) -> dict[str, Any]: store=self._inner_store, demo_mode=self._demo_mode, ) + # The dashboard draws its trust gauge against this; it used to hard-code 40. + summary["trust"] = { + "quarantine_threshold": self._scorer.quarantine_threshold, + "quarantine_enforced": self._scorer.enforces_quarantine, + } summary["deployment_verification"] = { "status": "unverified", "detail": ( diff --git a/src/sentinel/control/static/index.html b/src/sentinel/control/static/index.html index d5c5e13..2b30d58 100644 --- a/src/sentinel/control/static/index.html +++ b/src/sentinel/control/static/index.html @@ -410,6 +410,7 @@

SENTINEL_