From ef1756b53ebf51c0acd9c9f68c731436359b2a86 Mon Sep 17 00:00:00 2001 From: Carlos Hernandez Date: Thu, 15 Oct 2026 02:23:29 +0200 Subject: [PATCH] feat(shadow-ai): attribute cMCP audits with trusted session bindings Resolve session identity from an explicit source-scoped operator mapping, preserve incomplete scan reports, and test the released cMCP 0.2.0 and 0.7.0 audit producers. Add genuine fixtures, scoped compatibility documentation, and a hosted Python/version matrix. Fixes #215 Signed-off-by: Carlos Hernandez --- .github/workflows/shadow-ai-tests.yml | 38 ++++ README.md | 3 +- noxfile.py | 13 +- shadow-ai/README.md | 87 +++++++- shadow-ai/src/cmcp_shadow_ai.py | 118 +++++++++++ shadow-ai/src/shadow_ai_discovery.py | 8 +- shadow-ai/tests/cmcp_producer.py | 41 ++++ shadow-ai/tests/fixtures/README.md | 31 +++ .../tests/fixtures/cmcp-0.2.0/audit.jsonl | 7 + .../tests/fixtures/cmcp-0.2.0/catalog.json | 22 +++ .../tests/fixtures/cmcp-0.2.0/provenance.json | 10 + .../tests/fixtures/cmcp-0.7.0/audit.jsonl | 7 + .../tests/fixtures/cmcp-0.7.0/catalog.json | 36 ++++ .../tests/fixtures/cmcp-0.7.0/provenance.json | 10 + shadow-ai/tests/test_cmcp_interop.py | 105 ++++++++++ shadow-ai/tests/test_cmcp_shadow_ai.py | 187 ++++++++++++++++++ 16 files changed, 712 insertions(+), 11 deletions(-) create mode 100644 .github/workflows/shadow-ai-tests.yml create mode 100644 shadow-ai/src/cmcp_shadow_ai.py create mode 100644 shadow-ai/tests/cmcp_producer.py create mode 100644 shadow-ai/tests/fixtures/README.md create mode 100644 shadow-ai/tests/fixtures/cmcp-0.2.0/audit.jsonl create mode 100644 shadow-ai/tests/fixtures/cmcp-0.2.0/catalog.json create mode 100644 shadow-ai/tests/fixtures/cmcp-0.2.0/provenance.json create mode 100644 shadow-ai/tests/fixtures/cmcp-0.7.0/audit.jsonl create mode 100644 shadow-ai/tests/fixtures/cmcp-0.7.0/catalog.json create mode 100644 shadow-ai/tests/fixtures/cmcp-0.7.0/provenance.json create mode 100644 shadow-ai/tests/test_cmcp_interop.py create mode 100644 shadow-ai/tests/test_cmcp_shadow_ai.py diff --git a/.github/workflows/shadow-ai-tests.yml b/.github/workflows/shadow-ai-tests.yml new file mode 100644 index 00000000..ff594032 --- /dev/null +++ b/.github/workflows/shadow-ai-tests.yml @@ -0,0 +1,38 @@ +name: Shadow AI tests + +on: + pull_request: + paths: ['shadow-ai/**', 'noxfile.py', 'requirements/validate.txt', '.github/workflows/shadow-ai-tests.yml'] + push: + branches: [main] + paths: ['shadow-ai/**', 'noxfile.py', 'requirements/validate.txt', '.github/workflows/shadow-ai-tests.yml'] + +permissions: + contents: read + +jobs: + test: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python-version: ['3.11', '3.12', '3.13'] + cmcp-version: ['0.2.0', '0.7.0'] + env: + PYTHONPATH: shadow-ai/src + SHADOW_AI_REQUIRE_CMCP: '1' + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: ${{ matrix.python-version }} + - name: Install tests + run: pip install --require-hashes -r requirements/validate.txt + - name: Install released cMCP audit API + # AuditChain and SqliteAuditStore use only the standard library. This + # exercises their wire format, not a gateway/TEE or its dependencies. + run: pip install --no-deps "cmcp-runtime==${{ matrix.cmcp-version }}" + - name: Test scanner and cMCP adapter + run: python -m pytest shadow-ai/tests -q diff --git a/README.md b/README.md index a48fc129..029c6846 100644 --- a/README.md +++ b/README.md @@ -126,7 +126,8 @@ select an `integrates_with` value from the current schema. See the note below. [Shadow AI Discovery](shadow-ai/) is also standalone tooling outside the index and Marketplace. It scans enriched records against an agent-to-tools registry; -it has no direct cMCP adapter and does not read or write Agent Manifest records. +its cMCP audit adapter requires operator-controlled source/session bindings. It +does not read or write Agent Manifest records. All seven engines share [`agentrust-capture-core`](packages/agentrust-capture-core), which owns fingerprinting, comparison, baseline sealing and the report honesty rules. diff --git a/noxfile.py b/noxfile.py index 6f8a66c9..dce8956c 100644 --- a/noxfile.py +++ b/noxfile.py @@ -6,7 +6,7 @@ nox.options.default_venv_backend = "venv" -nox.options.sessions = ["capture_core", "trace_adapters", "capture_engines", "framework_adapters", "openai_agents_adapter", "google_adk_adapter", "shadow_ai", "wcm_integrations"] +nox.options.sessions = ["capture_core", "trace_adapters", "capture_engines", "framework_adapters", "openai_agents_adapter", "google_adk_adapter", "shadow_ai", "shadow_ai_cmcp", "wcm_integrations"] def pytest(session: nox.Session, *paths: str) -> None: @@ -92,6 +92,17 @@ def shadow_ai(session: nox.Session) -> None: pytest(session, "shadow-ai/tests") +@nox.session(python="3.12") +@nox.parametrize("cmcp_version", ["0.2.0", "0.7.0"]) +def shadow_ai_cmcp(session: nox.Session, cmcp_version: str) -> None: + """Published cMCP audit APIs only; these modules use the standard library.""" + session.install("--require-hashes", "-r", "requirements/validate.txt") + session.install("--no-deps", f"cmcp-runtime=={cmcp_version}") + session.env["PYTHONPATH"] = "shadow-ai/src" + session.env["SHADOW_AI_REQUIRE_CMCP"] = "1" + pytest(session, "shadow-ai/tests") + + @nox.session(python=["3.11", "3.12", "3.13"]) def wcm_integrations(session: nox.Session) -> None: """The Weight Custody Manifest integrations, against the published SDK. diff --git a/shadow-ai/README.md b/shadow-ai/README.md index 4e814cd3..60c28d39 100644 --- a/shadow-ai/README.md +++ b/shadow-ai/README.md @@ -1,9 +1,10 @@ # Shadow AI Discovery -Finds AI agents and tool use that nobody registered. It reads tool-call audit records (with agent IDs added) and compares them with a separate registry that lists each agent and the tools it is allowed to use, then reports agents that are not in the registry and tools an agent was never declared to use. It does not read cMCP output directly; see [Inputs and limitations](#inputs-and-limitations). +Finds AI agents and tool use that nobody registered. It compares tool-call audit records with a separate registry that lists each agent and the tools it may use, then reports agents that are not in the registry and undeclared tool calls. The base scanner needs enriched records with agent IDs; the [cMCP adapter](#cmcp-adapter) can attribute cMCP audit records using explicit operator-controlled session bindings. This is standalone tooling, outside the integration index and Marketplace. It -has no working cMCP or Agent Manifest adapter and claims neither integration. +includes a cMCP audit adapter tested against the published 0.2.0 and 0.7.0 audit +APIs. It has no Agent Manifest adapter and makes no gateway/TEE verification claim. The source moved from `integrations/shadow-ai/` to `shadow-ai/`; update local `PYTHONPATH` settings to `shadow-ai/src`. Run its tests with `nox -s shadow_ai`. @@ -23,7 +24,7 @@ A tool-call record without a usable `agent_id` is never classified. The scanner | Stack component | How | |---|---| -| **cMCP** | Intended source of tool-call audit records, once they are enriched with a trustworthy agent identity. Not consumed directly today; see [Inputs and limitations](#inputs-and-limitations). | +| **cMCP** | `CmcpAuditAdapter` consumes `AuditChain` / `SqliteAuditStore` JSON records using a separate trusted source/session-to-agent mapping. | | **Agent Manifest** | `suggested_manifest_id` is a naming hint for registering a discovered agent. The scanner does not read, write or validate Agent Manifest records. | ## Inputs and limitations @@ -52,7 +53,7 @@ JSON objects, passed as a list or read from a JSONL file (one object per line): | `agent_id` | Required on tool-call records. Usable only if it is a string that is not empty or whitespace-only; used verbatim, never trimmed, lowercased or converted from another type. | | `timestamp` | Optional. Copied as-is into results, `""` if absent. Not parsed, validated or normalized, and its type is not checked at runtime: a non-string value is copied unchanged. | -These are enriched records: whatever produces them must supply a trustworthy `agent_id`. cMCP v0.2.0 audit entries carry no `agent_id` and name their timestamp `timestamp_utc`, so they cannot be scanned directly. The scanner does not map `session_id` to an agent, does not read `timestamp_utc`, and does not infer identity from the tool catalog. Direct cMCP compatibility is not claimed until an adapter with a trustworthy session-to-agent mapping exists and is tested ([integrations#215](https://github.com/agentrust-io/integrations/issues/215)). +These are enriched records: whatever produces them must supply a trustworthy `agent_id`. cMCP audit entries carry `session_id` instead of `agent_id` and name their timestamp `timestamp_utc`. `ShadowAIScanner` still does not map session identity or read `timestamp_utc`; use the separate adapter below for identity attribution. Neither class infers identity from a tool catalog. The adapter addresses the remaining identity-mapping work in [integrations#215](https://github.com/agentrust-io/integrations/issues/215). JSONL files are read with Python's default text encoding (`Path.read_text()`). Lines are split only on `\n`, `\r\n` and `\r`; blank lines are skipped but still counted, so reported line numbers are physical line numbers. A line that is not valid JSON raises `json.JSONDecodeError`, and a record that is not a JSON object raises `AttributeError`; both stop the scan, from every method. @@ -75,6 +76,75 @@ if not report.complete: print(f"unclassified tool call at {item.location_kind} {item.location}: {item.reason}") ``` +## cMCP adapter + +`CmcpAuditAdapter` inherits the same four scan methods and result types. Supply +bindings from an operator-controlled configuration or an authenticated session +control plane. For example, after that control plane authenticates the agent and +records the session ID returned by the selected gateway: + +```python +from cmcp_shadow_ai import CmcpAuditAdapter, SessionAgentBinding + +source = "billing-gateway/boot-42" +bindings = [ + SessionAgentBinding(source, "session-id-from-that-gateway", "billing-agent"), +] +scanner = CmcpAuditAdapter( + "agents.json", + source_id=source, + trusted_bindings=bindings, +) +report = scanner.scan_audit_log_report("verified-billing-gateway-audit.jsonl") +``` + +The caller must establish that the audit stream belongs to that source and that +the session belongs to the named agent. A successful constructor does not +authenticate those assertions. The adapter does not authenticate a caller, +verify an audit signature/chain, perform attestation or read a manifest. Do not +build bindings from a tool catalog, an unverified manifest, audit fields, or an +agent's own identity assertion. Keep the agent registry and bindings outside the +audited agent's write access. + +`source_id` identifies the gateway and its session namespace/epoch. Use a new +source ID if session IDs can be reused after a restart, reset or import. Keep +each input stream within its selected source; a `source_id` inside a record +cannot choose a binding. The adapter matches `(source_id, session_id)` exactly, +without trimming, case folding or prefix matching. It snapshots the supplied +bindings at construction and rejects duplicate source/session keys, including +duplicates with the same agent, so an entry cannot overwrite another identity. + +For tool-call records, the adapter reports these additional unavailable reasons: + +| Reason | Meaning | +|---|---| +| `session_missing` | No `session_id` field | +| `session_null` | Session ID is `null` | +| `session_blank` | Session ID is empty or whitespace-only | +| `session_non_string` | Session ID is not a string; `observed_type` reports its JSON type | +| `session_unmapped` | No binding for the selected source and exact session ID | +| `agent_conflict` | An extra `agent_id` in the record disagrees with the trusted binding | + +These diagnostics have no agent ID or suggested manifest ID. Reports retain +valid findings and mark classification incomplete; strict list methods raise +`IdentityUnavailableError`. A record-supplied agent ID is never a fallback for a +missing binding. The adapter leaves input records unchanged and rejects a +non-empty malformed tool name rather than treating it as a clean scan. + +Timestamp mapping remains separate from identity attribution: `timestamp` is +copied with the existing scanner semantics, and `timestamp_utc` is not mapped or +normalized. Findings from native cMCP records therefore have `timestamp=""`. +Records with non-empty tool names include denied attempts; a finding does not +establish that a tool executed or that an effect occurred. + +The tested contract is JSON serialization from the **published** +`cmcp-runtime==0.2.0` and `cmcp-runtime==0.7.0` `AuditChain` and +`SqliteAuditStore` APIs. Tests install those packages with `--no-deps` because +these audit modules use the standard library. They exercise neither the full +gateway dependency stack nor a live gateway, TEE or authenticated identity +provider. Other releases are not claimed as tested. Genuine catalog/audit +fixtures and their provenance are under [tests/fixtures](tests/fixtures/README.md). + ## Methods | Method | Returns | Tool-call record without a usable `agent_id` | @@ -135,10 +205,17 @@ JSONL lines are now split only on line breaks. Previously a character such as U+ ## Running tests +From the repository root: + ```bash -python -m pytest tests/ -v +nox -s shadow_ai shadow_ai_cmcp ``` +`shadow_ai` tests the dependency-free scanner and adapter; producer tests skip +when cMCP is not installed. `shadow_ai_cmcp` runs the suite with each pinned +published cMCP release, requiring the producer tests to run. GitHub Actions runs +both releases on Python 3.11, 3.12 and 3.13. + ## License Apache 2.0 diff --git a/shadow-ai/src/cmcp_shadow_ai.py b/shadow-ai/src/cmcp_shadow_ai.py new file mode 100644 index 00000000..d40b2a89 --- /dev/null +++ b/shadow-ai/src/cmcp_shadow_ai.py @@ -0,0 +1,118 @@ +"""Attribute cMCP audit records using operator-supplied session bindings. + +Bindings and the audit source identifier are trusted policy inputs. This module +does not authenticate logs, attest a gateway, or derive identity from its catalog. +""" +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from types import MappingProxyType +from typing import Any, Iterable, Iterator + +from shadow_ai_discovery import ( + IdentityUnavailable, + ScanReport, + ShadowAIScanner, + _json_type_name, +) + + +def _require_identifier(value: str, field: str) -> None: + if not isinstance(value, str) or not value.strip(): + raise ValueError(f"{field} must be a non-blank string") + + +@dataclass(frozen=True) +class SessionAgentBinding: + """One authenticated session's identity, supplied by an operator/control plane. + + Use a source_id that identifies the gateway and its session namespace/epoch. + Do not construct these from audit fields or a tool catalog. Identifiers are + exact strings, with no trimming, case folding or session-prefix matching. + """ + + source_id: str + session_id: str + agent_id: str + + def __post_init__(self) -> None: + for field in ("source_id", "session_id", "agent_id"): + _require_identifier(getattr(self, field), field) + + +class CmcpAuditAdapter(ShadowAIScanner): + """The scanner's four scan methods, with explicit cMCP session attribution. + + source_id comes from the trusted caller's audit-source selection, never a + record. trusted_bindings are a snapshot: subsequent caller mutations cannot + rebind a session. Any duplicate source/session key is a configuration error. + + Input records stay unchanged. timestamp retains the scanner's existing + behavior; timestamp_utc mapping/normalization is deliberately separate. + """ + + def __init__( + self, + catalog_path: str | Path, + *, + source_id: str, + trusted_bindings: Iterable[SessionAgentBinding], + ) -> None: + _require_identifier(source_id, "source_id") + bindings: dict[tuple[str, str], str] = {} + for binding in trusted_bindings: + if not isinstance(binding, SessionAgentBinding): + raise TypeError("trusted_bindings must contain SessionAgentBinding objects") + key = (binding.source_id, binding.session_id) + if key in bindings: + raise ValueError("duplicate source/session binding; attribution is ambiguous") + bindings[key] = binding.agent_id + self._source_id = source_id + self._bindings = MappingProxyType(bindings) + super().__init__(catalog_path) + + def _classify( + self, located_records: Iterable[tuple[int, dict[str, Any]]], location_kind: str + ) -> ScanReport: + unavailable: list[IdentityUnavailable] = [] + + def attributed() -> Iterator[tuple[int, dict[str, Any]]]: + for location, record in located_records: + tool_name = record.get("tool_name") + if not tool_name: + continue + if not isinstance(tool_name, str) or not tool_name.strip(): + raise ValueError(f"invalid tool_name at {location_kind} {location}") + reason, observed_type = self._session_reason(record) + if reason is None: + agent_id = self._bindings[(self._source_id, record["session_id"])] + # An extra claim in a record cannot replace the trusted + # binding or conceal a conflicting identity assertion. + if "agent_id" in record and record["agent_id"] != agent_id: + reason = "agent_conflict" + if reason is not None: + unavailable.append(IdentityUnavailable( + location_kind, location, tool_name, record.get("timestamp", ""), + reason, observed_type, + )) + continue + yield location, {**record, "agent_id": agent_id} + + report = super()._classify(attributed(), location_kind) + diagnostics = sorted((*unavailable, *report.unavailable), key=lambda item: item.location) + return ScanReport(report.findings, tuple(diagnostics)) + + def _session_reason(self, record: dict[str, Any]) -> tuple[str | None, str | None]: + if "session_id" not in record: + return "session_missing", None + session_id = record["session_id"] + if session_id is None: + return "session_null", None + if not isinstance(session_id, str): + return "session_non_string", _json_type_name(session_id) + if not session_id.strip(): + return "session_blank", None + if (self._source_id, session_id) not in self._bindings: + return "session_unmapped", None + return None, None diff --git a/shadow-ai/src/shadow_ai_discovery.py b/shadow-ai/src/shadow_ai_discovery.py index 4579bed3..eaac14ea 100644 --- a/shadow-ai/src/shadow_ai_discovery.py +++ b/shadow-ai/src/shadow_ai_discovery.py @@ -38,8 +38,8 @@ def to_dict(self) -> dict[str, str]: @dataclass(frozen=True) class IdentityUnavailable: """ - A tool-call record that could not be classified because it carries no usable - ``agent_id``. The rejected agent_id is not stored as an identity field: there + A tool-call record that could not be attributed to a usable ``agent_id``. + The rejected identity is not stored as an identity field: there is no agent_id or suggested_manifest_id. Other record fields (tool_name, timestamp) are copied as-is, without redaction. """ @@ -48,8 +48,8 @@ class IdentityUnavailable: location: int tool_name: str timestamp: str # copied from the record without type validation; "" if absent - reason: str # "missing" | "null" | "blank" | "non_string" - observed_type: str | None = None # JSON type name, set only for "non_string" + reason: str # scanner identity reasons, or the cMCP adapter's session reasons + observed_type: str | None = None # JSON type name for non-string identity/session IDs def to_dict(self) -> dict[str, Any]: return { diff --git a/shadow-ai/tests/cmcp_producer.py b/shadow-ai/tests/cmcp_producer.py new file mode 100644 index 00000000..8088560b --- /dev/null +++ b/shadow-ai/tests/cmcp_producer.py @@ -0,0 +1,41 @@ +"""Produce real cMCP SQLite payloads using an installed, released audit API.""" +from __future__ import annotations + +import json +import sqlite3 +import sys +import tempfile +from importlib.metadata import version +from pathlib import Path + +from cmcp_runtime.audit.chain import AuditChain +from cmcp_runtime.audit.store import SqliteAuditStore + + +def write_audit(path: Path) -> list[dict]: + """Exercise AuditChain and the exact payload serialization in SqliteAuditStore. + + No gateway, network, TEE or caller authentication is exercised here. Session + identity comes from the separate operator bindings supplied by the tests. + """ + if version("cmcp-runtime") not in {"0.2.0", "0.7.0"}: + raise ValueError("fixture producer requires cmcp-runtime 0.2.0 or 0.7.0") + with tempfile.TemporaryDirectory() as directory: + database = Path(directory) / "audit.db" + store = SqliteAuditStore(database) + try: + for session in ("session-a", "session-b", "session-c"): + chain = AuditChain(session, store=store) + chain.append("tool_call", call_id=session + "/1", tool_name="echo", policy_decision="allow") + if session == "session-a": + chain.append("tool_call", call_id=session + "/2", tool_name="delete_all", policy_decision="deny") + finally: + store.close() + with sqlite3.connect(database) as connection: + payloads = [row[0] for row in connection.execute("SELECT payload FROM audit_entries ORDER BY rowid")] + path.write_text("\n".join(payloads) + "\n", encoding="utf-8") + return [json.loads(payload) for payload in payloads] + + +if __name__ == "__main__": + write_audit(Path(sys.argv[1])) diff --git a/shadow-ai/tests/fixtures/README.md b/shadow-ai/tests/fixtures/README.md new file mode 100644 index 00000000..20b0918a --- /dev/null +++ b/shadow-ai/tests/fixtures/README.md @@ -0,0 +1,31 @@ +# cMCP producer fixtures + +Each release directory contains: + +- `catalog.json`: exact bytes of `examples/minimal/catalog.json` at the release + tag. `provenance.json` records its immutable source URL and Git blob ID. +- `audit.jsonl`: actual `SqliteAuditStore.payload` values produced by + `AuditChain.append()` from the corresponding published `cmcp-runtime` wheel, + via [cmcp_producer.py](../cmcp_producer.py). The records contain no agent IDs. +- `provenance.json`: the producer version and SHA-256 of each fixture file. + +The fixture uses three synthetic session IDs. Session A attempts `echo` and +`delete_all`; sessions B and C attempt `echo`. Tests supply separate bindings +and an agent registry, then check registered/declaration controls, an +unregistered agent and an unmapped session. The cMCP tool catalog remains +invalid as an agent registry. + +To reproduce the audit wire format in an isolated environment: + +```bash +python -m venv /tmp/shadow-ai-cmcp-02 +/tmp/shadow-ai-cmcp-02/bin/python -m pip install --no-deps cmcp-runtime==0.2.0 +/tmp/shadow-ai-cmcp-02/bin/python shadow-ai/tests/cmcp_producer.py /tmp/cmcp-02-audit.jsonl +``` + +Repeat with `0.7.0` in a separate environment. UUIDs and timestamps vary on each +run. Interoperability tests generate fresh records, read them through the +adapter, and independently check their hashes and predecessor links. Fixture +hash checks detect accidental changes; they do not authenticate a gateway or +prove the operator-supplied session identity. No hardware or remote service is +used. diff --git a/shadow-ai/tests/fixtures/cmcp-0.2.0/audit.jsonl b/shadow-ai/tests/fixtures/cmcp-0.2.0/audit.jsonl new file mode 100644 index 00000000..4cacc61c --- /dev/null +++ b/shadow-ai/tests/fixtures/cmcp-0.2.0/audit.jsonl @@ -0,0 +1,7 @@ +{"call_id":null,"detail":null,"entry_hash":"b59061203fe1c634f50c1ecc58bed279c8bc599fd896d54a643cfaec1b1248d1","entry_id":"4a8302c1-fc5c-48c4-9bd6-579ead6b8e56","entry_type":"session_start","latency_us":null,"policy_decision":"n/a","policy_rule_matched":null,"prev_entry_hash":"genesis","request_payload_hash":null,"response_inspection_result":"n/a","response_payload_hash":null,"sequence_number":0,"server_identity":null,"session_id":"session-a","session_sensitivity_after":"public","session_sensitivity_before":null,"timestamp_utc":"2026-10-15T00:20:08.490358+00:00","tool_name":null,"workflow_id":null} +{"call_id":"session-a/1","detail":null,"entry_hash":"5a2b28b194c9d33d557aa748aea62ac8b3259f528aa7078d2f452f63dd078f98","entry_id":"bacceba0-5f5a-4f17-ac52-9582effba9d5","entry_type":"tool_call","latency_us":null,"policy_decision":"allow","policy_rule_matched":null,"prev_entry_hash":"b59061203fe1c634f50c1ecc58bed279c8bc599fd896d54a643cfaec1b1248d1","request_payload_hash":null,"response_inspection_result":null,"response_payload_hash":null,"sequence_number":1,"server_identity":null,"session_id":"session-a","session_sensitivity_after":null,"session_sensitivity_before":null,"timestamp_utc":"2026-10-15T00:20:08.490667+00:00","tool_name":"echo","workflow_id":null} +{"call_id":"session-a/2","detail":null,"entry_hash":"e3a55ad6e222db0b1f137db01ecce4783f2b472b57178415059448ebc109ba36","entry_id":"e09e129a-4446-4176-a2ba-df67977ec699","entry_type":"tool_call","latency_us":null,"policy_decision":"deny","policy_rule_matched":null,"prev_entry_hash":"5a2b28b194c9d33d557aa748aea62ac8b3259f528aa7078d2f452f63dd078f98","request_payload_hash":null,"response_inspection_result":null,"response_payload_hash":null,"sequence_number":2,"server_identity":null,"session_id":"session-a","session_sensitivity_after":null,"session_sensitivity_before":null,"timestamp_utc":"2026-10-15T00:20:08.490814+00:00","tool_name":"delete_all","workflow_id":null} +{"call_id":null,"detail":null,"entry_hash":"85770d8af633473ec156ae4ef50feb4625eeb39cd951f09df9f238657584a592","entry_id":"00ef76c6-c1d5-45aa-b2a1-cfe593e62b9c","entry_type":"session_start","latency_us":null,"policy_decision":"n/a","policy_rule_matched":null,"prev_entry_hash":"genesis","request_payload_hash":null,"response_inspection_result":"n/a","response_payload_hash":null,"sequence_number":0,"server_identity":null,"session_id":"session-b","session_sensitivity_after":"public","session_sensitivity_before":null,"timestamp_utc":"2026-10-15T00:20:08.490937+00:00","tool_name":null,"workflow_id":null} +{"call_id":"session-b/1","detail":null,"entry_hash":"c3a2dd693962c6123c9cf7e86b9149a3ab71ac3329e27236f15b9cb08d45bd68","entry_id":"99111c23-41c5-4623-a881-97ab131d2e77","entry_type":"tool_call","latency_us":null,"policy_decision":"allow","policy_rule_matched":null,"prev_entry_hash":"85770d8af633473ec156ae4ef50feb4625eeb39cd951f09df9f238657584a592","request_payload_hash":null,"response_inspection_result":null,"response_payload_hash":null,"sequence_number":1,"server_identity":null,"session_id":"session-b","session_sensitivity_after":null,"session_sensitivity_before":null,"timestamp_utc":"2026-10-15T00:20:08.491041+00:00","tool_name":"echo","workflow_id":null} +{"call_id":null,"detail":null,"entry_hash":"7ad88cc07233c89e20894e50a409669354d31145d7c3d0c9b379d9cb3cf77d6b","entry_id":"e8334a66-6a8b-454d-b9b6-c549746fedec","entry_type":"session_start","latency_us":null,"policy_decision":"n/a","policy_rule_matched":null,"prev_entry_hash":"genesis","request_payload_hash":null,"response_inspection_result":"n/a","response_payload_hash":null,"sequence_number":0,"server_identity":null,"session_id":"session-c","session_sensitivity_after":"public","session_sensitivity_before":null,"timestamp_utc":"2026-10-15T00:20:08.491123+00:00","tool_name":null,"workflow_id":null} +{"call_id":"session-c/1","detail":null,"entry_hash":"669b35e8fadf87b6bd58940712c48906e11c8431e5f2730d3d18c6199cc5057b","entry_id":"bee9ff1d-299c-49ed-bcc2-d63e114834ea","entry_type":"tool_call","latency_us":null,"policy_decision":"allow","policy_rule_matched":null,"prev_entry_hash":"7ad88cc07233c89e20894e50a409669354d31145d7c3d0c9b379d9cb3cf77d6b","request_payload_hash":null,"response_inspection_result":null,"response_payload_hash":null,"sequence_number":1,"server_identity":null,"session_id":"session-c","session_sensitivity_after":null,"session_sensitivity_before":null,"timestamp_utc":"2026-10-15T00:20:08.491225+00:00","tool_name":"echo","workflow_id":null} diff --git a/shadow-ai/tests/fixtures/cmcp-0.2.0/catalog.json b/shadow-ai/tests/fixtures/cmcp-0.2.0/catalog.json new file mode 100644 index 00000000..206c2c89 --- /dev/null +++ b/shadow-ai/tests/fixtures/cmcp-0.2.0/catalog.json @@ -0,0 +1,22 @@ +[ + { + "tool_name": "echo", + "server": { + "display_name": "Echo MCP Server (mock)", + "url": "http://localhost:9001/mcp", + "tls_fingerprint": "SHA256:REPLACE_WITH_REAL_FINGERPRINT", + "transport": "http-sse", + "rotation_mode": "key-pinned" + }, + "approved_definition": { + "description": "Returns its input unchanged. For testing only.", + "input_schema": {"type": "object", "properties": {"message": {"type": "string"}}}, + "output_schema": {"type": "object", "properties": {"message": {"type": "string"}}} + }, + "compliance_domain": "public", + "requires_baa": false, + "sensitivity_level": "public", + "added_at": "2026-06-04T00:00:00Z", + "approved_by": "developer@example.com" + } +] diff --git a/shadow-ai/tests/fixtures/cmcp-0.2.0/provenance.json b/shadow-ai/tests/fixtures/cmcp-0.2.0/provenance.json new file mode 100644 index 00000000..15ddb653 --- /dev/null +++ b/shadow-ai/tests/fixtures/cmcp-0.2.0/provenance.json @@ -0,0 +1,10 @@ +{ + "cmcp_runtime": "0.2.0", + "catalog_source": "https://github.com/agentrust-io/cmcp/blob/572ba840dde70ab5bca24399cae51a72a479dcea/examples/minimal/catalog.json", + "catalog_git_blob": "206c2c89c04a499ac03d9e474e524e284ea84db5", + "audit_producer": "AuditChain.append + SqliteAuditStore payload, generated by tests/cmcp_producer.py from the published PyPI wheel", + "sha256": { + "catalog.json": "f9ef889469ff5dedcb9e8ea6bc2bb8ff9ddbbda2cea001adad5bf56c805099d4", + "audit.jsonl": "c29718477843f41851bf3c68dca12c9e4e9c22f4e96729c31f3efab51aa71fab" + } +} diff --git a/shadow-ai/tests/fixtures/cmcp-0.7.0/audit.jsonl b/shadow-ai/tests/fixtures/cmcp-0.7.0/audit.jsonl new file mode 100644 index 00000000..4a0d1d08 --- /dev/null +++ b/shadow-ai/tests/fixtures/cmcp-0.7.0/audit.jsonl @@ -0,0 +1,7 @@ +{"call_id":null,"detail":null,"effective_data_class":null,"entry_hash":"1294796a01e9d73457a12433cca4e33f24c30b11d8ab70d677ec02d3cbdd72f0","entry_id":"c4c6be0f-c2d0-46bb-bb0c-abd6a3a15aac","entry_type":"session_start","evidence_class":"hash-only","execution_id":null,"external_execution_evidence":null,"latency_us":null,"policy_decision":"n/a","policy_rule_matched":null,"prev_entry_hash":"genesis","request_payload_hash":null,"response_inspection_result":"n/a","response_payload_hash":null,"sequence_number":0,"server_identity":null,"session_id":"session-a","session_sensitivity_after":"public","session_sensitivity_before":null,"timestamp_utc":"2026-10-15T00:20:08.572427+00:00","tool_name":null,"workflow_id":null} +{"call_id":"session-a/1","detail":null,"effective_data_class":null,"entry_hash":"9834a154e09b8c2f0de4172a62d7bc273764c250538a888cdb2f97e7cd686e26","entry_id":"8ce472af-8837-4381-8a45-05bfb34d430a","entry_type":"tool_call","evidence_class":"hash-only","execution_id":null,"external_execution_evidence":null,"latency_us":null,"policy_decision":"allow","policy_rule_matched":null,"prev_entry_hash":"1294796a01e9d73457a12433cca4e33f24c30b11d8ab70d677ec02d3cbdd72f0","request_payload_hash":null,"response_inspection_result":null,"response_payload_hash":null,"sequence_number":1,"server_identity":null,"session_id":"session-a","session_sensitivity_after":null,"session_sensitivity_before":null,"timestamp_utc":"2026-10-15T00:20:08.572668+00:00","tool_name":"echo","workflow_id":null} +{"call_id":"session-a/2","detail":null,"effective_data_class":null,"entry_hash":"3a3b853183bc506715e9b1ba839fdb93c55f41ae2c8ce36e01939e9d37b0d905","entry_id":"3aa047c0-6bb9-4dcd-a258-eede38a49a4e","entry_type":"tool_call","evidence_class":"hash-only","execution_id":null,"external_execution_evidence":null,"latency_us":null,"policy_decision":"deny","policy_rule_matched":null,"prev_entry_hash":"9834a154e09b8c2f0de4172a62d7bc273764c250538a888cdb2f97e7cd686e26","request_payload_hash":null,"response_inspection_result":null,"response_payload_hash":null,"sequence_number":2,"server_identity":null,"session_id":"session-a","session_sensitivity_after":null,"session_sensitivity_before":null,"timestamp_utc":"2026-10-15T00:20:08.572792+00:00","tool_name":"delete_all","workflow_id":null} +{"call_id":null,"detail":null,"effective_data_class":null,"entry_hash":"33c1fac59802d1f5475a29e287b462eb5d66fe09c86df2a6c02569d20e330b8e","entry_id":"e8991cd1-c51a-4993-b4a7-9be3a3179631","entry_type":"session_start","evidence_class":"hash-only","execution_id":null,"external_execution_evidence":null,"latency_us":null,"policy_decision":"n/a","policy_rule_matched":null,"prev_entry_hash":"genesis","request_payload_hash":null,"response_inspection_result":"n/a","response_payload_hash":null,"sequence_number":0,"server_identity":null,"session_id":"session-b","session_sensitivity_after":"public","session_sensitivity_before":null,"timestamp_utc":"2026-10-15T00:20:08.572904+00:00","tool_name":null,"workflow_id":null} +{"call_id":"session-b/1","detail":null,"effective_data_class":null,"entry_hash":"3a75a729ab90a747ec6a8178cd6d69623653a5782932d1a3868e578edc3c5fc4","entry_id":"10831251-f039-4e39-a610-57cd2ac6ee4c","entry_type":"tool_call","evidence_class":"hash-only","execution_id":null,"external_execution_evidence":null,"latency_us":null,"policy_decision":"allow","policy_rule_matched":null,"prev_entry_hash":"33c1fac59802d1f5475a29e287b462eb5d66fe09c86df2a6c02569d20e330b8e","request_payload_hash":null,"response_inspection_result":null,"response_payload_hash":null,"sequence_number":1,"server_identity":null,"session_id":"session-b","session_sensitivity_after":null,"session_sensitivity_before":null,"timestamp_utc":"2026-10-15T00:20:08.573011+00:00","tool_name":"echo","workflow_id":null} +{"call_id":null,"detail":null,"effective_data_class":null,"entry_hash":"a5bf6f48ff00ac752d7628064b956d054afc0ebdc2a919215611c4e8e1c8f9b9","entry_id":"a9226208-eef9-481d-9a43-e45f5bee2839","entry_type":"session_start","evidence_class":"hash-only","execution_id":null,"external_execution_evidence":null,"latency_us":null,"policy_decision":"n/a","policy_rule_matched":null,"prev_entry_hash":"genesis","request_payload_hash":null,"response_inspection_result":"n/a","response_payload_hash":null,"sequence_number":0,"server_identity":null,"session_id":"session-c","session_sensitivity_after":"public","session_sensitivity_before":null,"timestamp_utc":"2026-10-15T00:20:08.573123+00:00","tool_name":null,"workflow_id":null} +{"call_id":"session-c/1","detail":null,"effective_data_class":null,"entry_hash":"0ab519eaef0ebdb50877132e7d7c5c2ea39af89a605bed51f1795774aeab7a77","entry_id":"293f9d11-92ab-485e-8280-571a0c5d8f82","entry_type":"tool_call","evidence_class":"hash-only","execution_id":null,"external_execution_evidence":null,"latency_us":null,"policy_decision":"allow","policy_rule_matched":null,"prev_entry_hash":"a5bf6f48ff00ac752d7628064b956d054afc0ebdc2a919215611c4e8e1c8f9b9","request_payload_hash":null,"response_inspection_result":null,"response_payload_hash":null,"sequence_number":1,"server_identity":null,"session_id":"session-c","session_sensitivity_after":null,"session_sensitivity_before":null,"timestamp_utc":"2026-10-15T00:20:08.573225+00:00","tool_name":"echo","workflow_id":null} diff --git a/shadow-ai/tests/fixtures/cmcp-0.7.0/catalog.json b/shadow-ai/tests/fixtures/cmcp-0.7.0/catalog.json new file mode 100644 index 00000000..5880c818 --- /dev/null +++ b/shadow-ai/tests/fixtures/cmcp-0.7.0/catalog.json @@ -0,0 +1,36 @@ +[ + { + "tool_name": "echo", + "server": { + "display_name": "Echo MCP Server (mock)", + "url": "http://localhost:9001/mcp", + "tls_fingerprint": "SHA256:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=", + "transport": "http-sse" + }, + "approved_definition": { + "description": "Returns its input unchanged. For testing only.", + "input_schema": { + "type": "object", + "properties": { + "message": { + "type": "string" + } + } + }, + "output_schema": { + "type": "object", + "properties": { + "message": { + "type": "string" + } + } + } + }, + "compliance_domain": "public", + "requires_baa": false, + "sensitivity_level": "public", + "added_at": "2026-06-04T00:00:00Z", + "approved_by": "developer@example.com", + "definition_hash": "sha256:130436578985268754fd1925a01a7a12e2f94bcfd439f4f43050f816f866e8d6" + } +] diff --git a/shadow-ai/tests/fixtures/cmcp-0.7.0/provenance.json b/shadow-ai/tests/fixtures/cmcp-0.7.0/provenance.json new file mode 100644 index 00000000..b43bf042 --- /dev/null +++ b/shadow-ai/tests/fixtures/cmcp-0.7.0/provenance.json @@ -0,0 +1,10 @@ +{ + "cmcp_runtime": "0.7.0", + "catalog_source": "https://github.com/agentrust-io/cmcp/blob/66761889678915ba6f709a60c998ce7bd7eb779f/examples/minimal/catalog.json", + "catalog_git_blob": "5880c818e595912ae5c4edb3c58431471d85af25", + "audit_producer": "AuditChain.append + SqliteAuditStore payload, generated by tests/cmcp_producer.py from the published PyPI wheel", + "sha256": { + "catalog.json": "6cb08755ce77e88b9de11a2171a960596ca970322a78054de9343e3670735419", + "audit.jsonl": "e17f4d561c3ee59c5218fce7f841115fd173cc0a1aed079c457cf09d38636933" + } +} diff --git a/shadow-ai/tests/test_cmcp_interop.py b/shadow-ai/tests/test_cmcp_interop.py new file mode 100644 index 00000000..e3ba41a7 --- /dev/null +++ b/shadow-ai/tests/test_cmcp_interop.py @@ -0,0 +1,105 @@ +"""Producer/consumer controls against two explicitly pinned published audit APIs.""" +import hashlib +import json +import os +import sys +from importlib.metadata import version +from pathlib import Path + +import pytest + +if os.environ.get("SHADOW_AI_REQUIRE_CMCP") != "1": + pytest.importorskip("cmcp_runtime.audit.chain", reason="run nox -s shadow_ai_cmcp for released producer tests") + +sys.path.insert(0, str(Path(__file__).parent)) +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) +from cmcp_producer import write_audit +from cmcp_shadow_ai import CmcpAuditAdapter, SessionAgentBinding +from shadow_ai_discovery import IdentityUnavailableError, ShadowAIScanner + + +@pytest.fixture +def registry(tmp_path): + path = tmp_path / "agents.json" + path.write_text(json.dumps({"billing-agent": ["echo"]})) + return path + + +def adapter(registry, *, include_c=True): + bindings = [ + SessionAgentBinding("gateway-a/boot-1", "session-a", "billing-agent"), + SessionAgentBinding("gateway-a/boot-1", "session-b", "rogue-agent"), + ] + if include_c: + bindings.append(SessionAgentBinding("gateway-a/boot-1", "session-c", "billing-agent")) + return CmcpAuditAdapter(registry, source_id="gateway-a/boot-1", trusted_bindings=bindings) + + +def assert_producer_contract(records): + assert len(records) == 7 + assert all("agent_id" not in r and "timestamp" not in r for r in records) + assert all("session_id" in r and "timestamp_utc" in r for r in records) + previous = {} + for record in records: + body = {k: v for k, v in record.items() if k != "entry_hash"} + canonical = json.dumps(body, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode() + assert hashlib.sha256(canonical).hexdigest() == record["entry_hash"] + assert record["prev_entry_hash"] == previous.get(record["session_id"], "genesis") + previous[record["session_id"]] = record["entry_hash"] + + +def assert_classification(scanner, records): + report = scanner.scan_records_report(records) + assert report.complete + assert [(f.agent_id, f.tool_name, f.reason) for f in report.findings] == [ + ("billing-agent", "delete_all", "undeclared_tool"), + ("rogue-agent", "echo", "unregistered_agent"), + ] + # Timestamp normalization remains a separate change from identity mapping. + assert all(f.timestamp == "" for f in report.findings) + return report + + +def test_released_audit_api_and_sqlite_serialization(registry, tmp_path): + assert version("cmcp-runtime") in {"0.2.0", "0.7.0"} + path = tmp_path / "audit.jsonl" + records = write_audit(path) + assert_producer_contract(records) + scanner = adapter(registry) + report = assert_classification(scanner, records) + assert scanner.scan_audit_log_report(path) == report + assert not ShadowAIScanner(registry).scan_records_report(records).complete + + +def test_unmapped_real_session_keeps_findings_and_strict_methods_fail(registry, tmp_path): + path = tmp_path / "audit.jsonl" + records = write_audit(path) + scanner = adapter(registry, include_c=False) + report = scanner.scan_records_report(records) + assert not report.complete and len(report.findings) == 2 + assert [(d.location_kind, d.location, d.reason) for d in report.unavailable] == [ + ("index", 6, "session_unmapped"), + ] + file_report = scanner.scan_audit_log_report(path) + assert [(d.location_kind, d.location, d.reason) for d in file_report.unavailable] == [ + ("line", 7, "session_unmapped"), + ] + with pytest.raises(IdentityUnavailableError) as exc: + scanner.scan_audit_log(path) + assert exc.value.report == file_report + + +@pytest.mark.parametrize("release", ["0.2.0", "0.7.0"]) +def test_checked_in_genuine_fixtures_and_catalog(registry, release): + fixture = Path(__file__).parent / "fixtures" / ("cmcp-" + release) + provenance = json.loads((fixture / "provenance.json").read_text()) + for name, digest in provenance["sha256"].items(): + assert hashlib.sha256((fixture / name).read_bytes()).hexdigest() == digest + catalog = json.loads((fixture / "catalog.json").read_text()) + assert isinstance(catalog, list) and catalog[0]["tool_name"] == "echo" + # The actual producer catalog cannot authorize an agent or replace bindings. + with pytest.raises(ValueError, match="Unrecognized catalog format"): + CmcpAuditAdapter(fixture / "catalog.json", source_id="gateway-a/boot-1", trusted_bindings=[]) + records = [json.loads(line) for line in (fixture / "audit.jsonl").read_text().splitlines()] + assert_producer_contract(records) + assert_classification(adapter(registry), records) diff --git a/shadow-ai/tests/test_cmcp_shadow_ai.py b/shadow-ai/tests/test_cmcp_shadow_ai.py new file mode 100644 index 00000000..f3ad9f9b --- /dev/null +++ b/shadow-ai/tests/test_cmcp_shadow_ai.py @@ -0,0 +1,187 @@ +"""Session attribution must not turn missing identity into a clean scan.""" +import copy +import json +import sys +from dataclasses import FrozenInstanceError +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) +from cmcp_shadow_ai import CmcpAuditAdapter, SessionAgentBinding +from shadow_ai_discovery import IdentityUnavailableError, ShadowAIScanner + + +@pytest.fixture +def registry(tmp_path): + path = tmp_path / "agents.json" + path.write_text(json.dumps({"billing-agent": ["echo"]})) + return path + + +def adapter(registry, *, source="gateway-a/boot-1", bindings=None): + if bindings is None: + bindings = [SessionAgentBinding("gateway-a/boot-1", "session-a", "billing-agent")] + return CmcpAuditAdapter(registry, source_id=source, trusted_bindings=bindings) + + +def record(**changes): + return {"session_id": "session-a", "tool_name": "echo", **changes} + + +def via(scanner, records, tmp_path, mode, strict=False): + if mode == "records": + method = scanner.scan_records if strict else scanner.scan_records_report + return method(records) + path = tmp_path / "audit.jsonl" + # Preserve blank physical lines and CRLF accounting on the file path. + path.write_bytes(("\r\n" + "\r\n\r\n".join(json.dumps(r) for r in records)).encode()) + method = scanner.scan_audit_log if strict else scanner.scan_audit_log_report + return method(path) + + +@pytest.mark.parametrize("mode", ["records", "jsonl"]) +def test_positive_and_negative_controls(registry, tmp_path, mode): + bindings = [ + SessionAgentBinding("gateway-a/boot-1", "session-a", "billing-agent"), + SessionAgentBinding("gateway-a/boot-1", "session-b", "rogue-agent"), + ] + scanner = adapter(registry, bindings=bindings) + records = [record(), record(tool_name="delete_all"), record(session_id="session-b")] + before = copy.deepcopy(records) + report = via(scanner, records, tmp_path, mode) + assert report.complete + assert [(e.agent_id, e.reason) for e in report.findings] == [ + ("billing-agent", "undeclared_tool"), ("rogue-agent", "unregistered_agent"), + ] + assert via(scanner, records, tmp_path, mode, strict=True) == list(report.findings) + assert records == before + + +@pytest.mark.parametrize("mode", ["records", "jsonl"]) +@pytest.mark.parametrize("fields,reason,observed", [ + ({}, "session_missing", None), + ({"session_id": None}, "session_null", None), + ({"session_id": ""}, "session_blank", None), + ({"session_id": " \t "}, "session_blank", None), + ({"session_id": 424242}, "session_non_string", "number"), + ({"session_id": True}, "session_non_string", "boolean"), + ({"session_id": ["SENTINEL"]}, "session_non_string", "array"), + ({"session_id": {"id": "SENTINEL"}}, "session_non_string", "object"), + ({"session_id": "missing-session"}, "session_unmapped", None), + ({"session_id": "SESSION-A"}, "session_unmapped", None), + ({"session_id": " session-a "}, "session_unmapped", None), +]) +def test_unavailable_sessions_keep_real_findings(registry, tmp_path, mode, fields, reason, observed): + scanner = adapter(registry) + records = [record(tool_name="delete_all"), {"tool_name": "echo", **fields}, record()] + report = via(scanner, records, tmp_path, mode) + assert not report.complete + assert [e.reason for e in report.findings] == ["undeclared_tool"] + diagnostic, = report.unavailable + assert (diagnostic.reason, diagnostic.observed_type) == (reason, observed) + assert (diagnostic.location_kind, diagnostic.location) == ( + ("index", 1) if mode == "records" else ("line", 4) + ) + serialized = json.dumps(diagnostic.to_dict()) + assert "suggested_manifest_id" not in serialized + assert "agent_id" not in serialized and "424242" not in serialized and "SENTINEL" not in serialized + with pytest.raises(IdentityUnavailableError) as exc: + via(scanner, records, tmp_path, mode, strict=True) + assert exc.value.report == report + assert reason in str(exc.value) + assert "424242" not in str(exc.value) and "SENTINEL" not in str(exc.value) + + +@pytest.mark.parametrize("claim", ["rogue-agent", "", " ", None, 123, True, [], {}]) +def test_record_identity_cannot_override_mapping(registry, claim): + report = adapter(registry).scan_records_report([record(agent_id=claim)]) + assert report.findings == () + assert report.unavailable[0].reason == "agent_conflict" + assert not report.complete + + +def test_record_identity_is_never_a_fallback(registry): + report = adapter(registry, bindings=[]).scan_records_report([ + record(agent_id="billing-agent", source_id="gateway-a/boot-1"), + ]) + assert not report.complete and not report.findings + assert report.unavailable[0].reason == "session_unmapped" + + +def test_matching_record_claim_is_only_corroborating(registry): + assert adapter(registry).scan_records_report([record(agent_id="billing-agent")]).complete + + +def test_source_namespace_is_caller_supplied(registry): + bindings = [ + SessionAgentBinding("gateway-a/boot-1", "session-a", "billing-agent"), + SessionAgentBinding("gateway-b/boot-1", "session-a", "rogue-agent"), + ] + scanner = adapter(registry, source="gateway-b/boot-1", bindings=bindings) + report = scanner.scan_records_report([record(source_id="gateway-a/boot-1")]) + assert report.complete and report.findings[0].agent_id == "rogue-agent" + assert adapter(registry, source="gateway-a/boot-2", bindings=bindings).scan_records_report([ + record(agent_id="billing-agent"), + ]).unavailable[0].reason == "session_unmapped" + + +@pytest.mark.parametrize("other_agent", ["billing-agent", "rogue-agent"]) +def test_duplicate_bindings_are_rejected_even_if_identical(registry, other_agent): + bindings = [ + SessionAgentBinding("gateway-a/boot-1", "session-a", "billing-agent"), + SessionAgentBinding("gateway-a/boot-1", "session-a", other_agent), + ] + with pytest.raises(ValueError, match="duplicate source/session"): + adapter(registry, bindings=bindings) + + +@pytest.mark.parametrize("field", ["source_id", "session_id", "agent_id"]) +@pytest.mark.parametrize("invalid", [None, "", " \t", 1, True, [], {}]) +def test_invalid_binding_identifiers_fail_at_load(field, invalid): + fields = dict(source_id="gateway-a/boot-1", session_id="session-a", agent_id="billing-agent") + fields[field] = invalid + with pytest.raises(ValueError, match=field): + SessionAgentBinding(**fields) + + +def test_mapping_is_a_snapshot_and_binding_is_immutable(registry): + binding = SessionAgentBinding("gateway-a/boot-1", "session-a", "billing-agent") + bindings = [binding] + scanner = adapter(registry, bindings=bindings) + bindings.clear() + assert scanner.scan_records_report([record()]).complete + with pytest.raises(FrozenInstanceError): + binding.agent_id = "rogue-agent" + with pytest.raises(TypeError): + adapter(registry, bindings=[{"session_id": "session-a", "agent_id": "billing-agent"}]) + + +def test_exact_agent_ids_are_not_normalized(registry): + bindings = [SessionAgentBinding("gateway-a/boot-1", "session-a", " billing-agent ")] + report = adapter(registry, bindings=bindings).scan_records_report([record()]) + assert report.findings[0].agent_id == " billing-agent " + assert report.findings[0].reason == "unregistered_agent" + + +def test_non_tool_calls_and_timestamp_contract_are_unchanged(registry): + scanner = adapter(registry) + assert scanner.scan_records_report([{}, {"tool_name": None}, {"tool_name": ""}]).complete + report = scanner.scan_records_report([ + record(tool_name="delete_all", timestamp_utc="2026-10-15T00:00:00Z"), + record(tool_name="delete_all", timestamp="verbatim", timestamp_utc="ignored"), + ]) + assert [e.timestamp for e in report.findings] == ["", "verbatim"] + assert not ShadowAIScanner(registry).scan_records_report([record()]).complete + + +def test_jsonl_unicode_separator_is_not_a_physical_line_break(registry, tmp_path): + path = tmp_path / "audit.jsonl" + path.write_text(json.dumps(record(note="a\u2028b"), ensure_ascii=False)) + assert adapter(registry).scan_audit_log_report(path).complete + + +@pytest.mark.parametrize("tool_name", [" ", 123, ["echo"], {"name": "echo"}]) +def test_malformed_tool_call_does_not_become_a_clean_scan(registry, tool_name): + with pytest.raises(ValueError, match="invalid tool_name at index 0"): + adapter(registry).scan_records_report([record(tool_name=tool_name)])