diff --git a/core/backends/simulator.py b/core/backends/simulator.py index a99a4e2..823732d 100644 --- a/core/backends/simulator.py +++ b/core/backends/simulator.py @@ -144,3 +144,18 @@ def state_diff(self, twin_id: str) -> DiffResult: def teardown_twin(self, twin_id: str) -> None: # simulator: nothing to free. real backend: clab destroy + ns cleanup. return None + + # --- model identity (the #4 wedge) ------------------------------------ + def model_identity(self) -> dict: + """Deterministic attestation of the simulated generator. Fixed fields keep the + sealed bundle reproducible (stress INV-3). The real local/cloud backends attest a + weights-sha256 / identity-claim at the generate-path cutover.""" + return { + "provider": "simulator", + "model": "deterministic-sim-v1", + "model_hash": None, + "model_hash_kind": "identity-claim", + "api_version": None, + "capabilities": [], + "resolved_at_utc": "1970-01-01T00:00:00+00:00", + } diff --git a/core/llm/__init__.py b/core/llm/__init__.py index 3f3a265..f1115dc 100644 --- a/core/llm/__init__.py +++ b/core/llm/__init__.py @@ -32,7 +32,7 @@ classify, is_retryable, ) -from .identity import resolve_model_identity, unknown_identity +from .identity import IdentitySink, resolve_model_identity, unknown_identity __all__ = [ # egress (CROSS-1) @@ -70,4 +70,5 @@ # identity "resolve_model_identity", "unknown_identity", + "IdentitySink", ] diff --git a/core/llm/identity.py b/core/llm/identity.py index 700c6d7..1fe3450 100644 --- a/core/llm/identity.py +++ b/core/llm/identity.py @@ -84,3 +84,26 @@ def unknown_identity(created_utc: str) -> dict[str, object]: "capabilities": [], "resolved_at_utc": created_utc, } + + +class IdentitySink: + """Captures the ModelIdentity from the egress 'after_response' hook so the pipeline can + seal WHICH model produced a change into the evidence bundle (#4 wedge). + + Wire-up: egress.on("after_response", sink.capture) + Read: sink.last / sink.as_bundle_dict() -> pass to run_preflight(model_identity=...) + Process-local, last-write-wins (the egress makes one successful call per generate).""" + + def __init__(self) -> None: + self._last: ModelIdentity | None = None + + @property + def last(self) -> ModelIdentity | None: + return self._last + + def capture(self, provider: str, identity: ModelIdentity) -> None: + """after_response(provider, identity) observer — never raises into the egress.""" + self._last = identity + + def as_bundle_dict(self) -> dict[str, object] | None: + return self._last.to_bundle_dict() if self._last is not None else None diff --git a/core/orchestrator/pipeline.py b/core/orchestrator/pipeline.py index 65fe386..9e5e9f2 100644 --- a/core/orchestrator/pipeline.py +++ b/core/orchestrator/pipeline.py @@ -13,7 +13,7 @@ from ..backends.base import Backend, GeneratedConfig from . import guards, rollback -from ...evidence.bundler import build_bundle +from ...evidence.bundler import build_bundle, unattested_identity from ...evidence.compliance import map_controls @@ -26,7 +26,8 @@ def run_preflight(intent: str, *, backend: Backend, lab: str = "clos-evpn", operator: str = "unknown", source: str = "nl_intent", approver: str | None = None, - imported_configs: list[dict] | None = None) -> dict: + imported_configs: list[dict] | None = None, + model_identity: dict | None = None) -> dict: frameworks = frameworks or ["pci_dss_v4"] run_id = uuid.uuid4().hex created = datetime.now(timezone.utc).isoformat() @@ -89,6 +90,19 @@ def run_preflight(intent: str, *, backend: Backend, lab: str = "clos-evpn", decision, reason = _verdict(bf, twin, tier, needs_approval, approval_granted) + # seal WHICH model produced the change (#4 wedge). config_import is operator- + # supplied (no model) -> None -> build_bundle seals the honest _UNKNOWN_IDENTITY. + # nl_intent ALWAYS went through a model (generate_config), so the backend MUST + # attest which; if it cannot, seal an explicit UNATTESTED marker -- NEVER the + # operator-supplied identity (that would falsely deny model authorship). + if model_identity is not None: + mi = model_identity + elif source == "config_import": + mi = None + else: + attested = backend.model_identity() if hasattr(backend, "model_identity") else None + mi = attested if attested is not None else unattested_identity() + return build_bundle( run_id=run_id, created=created, operator=operator, intent=intent, source=source, configs=configs, batfish=bf, twin=twin, diff=diff, @@ -96,6 +110,7 @@ def run_preflight(intent: str, *, backend: Backend, lab: str = "clos-evpn", approver=approver if approval_granted else None, approval_utc=created if approval_granted else None, rollback_plan=plan, decision=decision, reason=reason, + model_identity=mi, ) finally: if twin_id is not None: diff --git a/evidence/bundler.py b/evidence/bundler.py index 87383e7..963e58a 100644 --- a/evidence/bundler.py +++ b/evidence/bundler.py @@ -27,12 +27,13 @@ def verify(bundle: dict) -> bool: def build_bundle(*, run_id, created, operator, intent, source, configs, batfish, twin, diff, compliance, tier, needs_approval, approver, approval_utc, rollback_plan, decision, reason, + model_identity: dict | None = None, signed: bool = False) -> dict: severity = _severity(diff["devices_affected"], len(diff["sessions_dropped"]), batfish["errors"], twin["converged"]) bundle = { - "bundle_version": "1.0", + "bundle_version": "1.1", "run_id": run_id, "created_utc": created, "operator": operator, @@ -50,6 +51,8 @@ def build_bundle(*, run_id, created, operator, intent, source, configs, batfish, "granted_by": approver, "granted_utc": approval_utc, }, + "model_identity": (model_identity if model_identity is not None + else _unknown_identity()), }, "twin": { "engine": "containerlab", @@ -98,3 +101,38 @@ def _severity(devices, dropped, errors, converged) -> str: if devices >= 1: return "low" return "none" + + +# Fixed timestamp for synthetic identities (UNKNOWN / simulator) so stress determinism +# (INV-3: same intent -> identical bundle) holds. Real backends carry their own resolved_at. +_SYNTHETIC_RESOLVED_AT = "1970-01-01T00:00:00+00:00" + + +def _unknown_identity() -> dict: + """The honest _UNKNOWN_IDENTITY for a change no model produced (config_import / + operator-supplied) — keeps the v1.1 bundle schema-valid WITHOUT claiming authorship.""" + return { + "provider": "none", + "model": "operator-supplied", + "model_hash": None, + "model_hash_kind": "identity-claim", + "api_version": None, + "capabilities": [], + "resolved_at_utc": _SYNTHETIC_RESOLVED_AT, + } + + +def unattested_identity() -> dict: + """A model DID produce this change (nl_intent) but the backend could not attest WHICH. + Deliberately DISTINCT from _unknown_identity (operator-supplied) so a missing + attestation can NEVER be read as 'no model was involved' (that would falsely deny AI + authorship). provider 'unknown' + model 'unattested' make the gap auditable.""" + return { + "provider": "unknown", + "model": "unattested", + "model_hash": None, + "model_hash_kind": "identity-claim", + "api_version": None, + "capabilities": [], + "resolved_at_utc": _SYNTHETIC_RESOLVED_AT, + } diff --git a/evidence/schema/evidence_bundle.schema.json b/evidence/schema/evidence_bundle.schema.json index d45c54d..98a44e6 100644 --- a/evidence/schema/evidence_bundle.schema.json +++ b/evidence/schema/evidence_bundle.schema.json @@ -7,7 +7,7 @@ "required": ["bundle_version", "run_id", "created_utc", "change", "twin", "validation", "verdict", "integrity"], "properties": { - "bundle_version": { "const": "1.0" }, + "bundle_version": { "const": "1.1" }, "run_id": { "type": "string", "minLength": 8 }, "created_utc": { "type": "string", "format": "date-time" }, "operator": { "type": "string" }, @@ -15,7 +15,8 @@ "change": { "type": "object", "additionalProperties": false, - "required": ["intent", "source", "generated_configs", "risk_tier", "approval"], + "required": ["intent", "source", "generated_configs", "risk_tier", "approval", + "model_identity"], "properties": { "intent": { "type": "string" }, "source": { "enum": ["nl_intent", "config_import"] }, @@ -43,6 +44,23 @@ "granted_by": { "type": ["string", "null"] }, "granted_utc": { "type": ["string", "null"], "format": "date-time" } } + }, + + "model_identity": { + "type": "object", + "additionalProperties": false, + "description": "The proposing model, sealed into the bundle SHA-256 (the #4 wedge). LOCAL backends carry a real weights-sha256; CLOUD backends carry an honest identity-claim with model_hash=null; config_import/no-model runs carry _UNKNOWN_IDENTITY (provider 'none'); the simulator attests provider 'simulator'.", + "required": ["provider", "model", "model_hash", "model_hash_kind", + "api_version", "capabilities", "resolved_at_utc"], + "properties": { + "provider": { "enum": ["openai-compatible-local", "anthropic-cloud", "simulator", "none", "unknown"] }, + "model": { "type": "string" }, + "model_hash": { "type": ["string", "null"], "pattern": "^[a-f0-9]{64}$" }, + "model_hash_kind": { "enum": ["weights-sha256", "identity-claim"] }, + "api_version": { "type": ["string", "null"] }, + "capabilities": { "type": "array", "items": { "enum": ["chat", "json", "tools"] } }, + "resolved_at_utc": { "type": "string" } + } } } }, @@ -112,7 +130,7 @@ "additionalProperties": false, "required": ["framework", "control", "status", "evidence"], "properties": { - "framework": { "enum": ["pci_dss_v4", "soc2", "nist_800-53"] }, + "framework": { "enum": ["pci_dss_v4", "soc2", "nist_800-53", "eu_ai_act"] }, "control": { "type": "string" }, "status": { "enum": ["pass", "fail", "n/a"] }, "evidence": { "type": "string" } diff --git a/tests/stress_test.py b/tests/stress_test.py index c314a33..bc85faa 100644 --- a/tests/stress_test.py +++ b/tests/stress_test.py @@ -77,6 +77,8 @@ def state_diff(self, twin_id): def teardown_twin(self, twin_id): self.torn += 1 return self.inner.teardown_twin(twin_id) + def model_identity(self): + return self.inner.model_identity() def _random_intent(rng: random.Random) -> str: diff --git a/tests/wedge_test.py b/tests/wedge_test.py new file mode 100644 index 0000000..e7c23fc --- /dev/null +++ b/tests/wedge_test.py @@ -0,0 +1,236 @@ +"""aegis/tests/wedge_test.py — #4 model-identity wedge: seal WHICH model made the change. + +House style: pytest-discoverable test_* + a dep-light __main__ runner +(CI: python3 -m aegis.tests.wedge_test). Tests run against the REAL bundle produced by +run_preflight + the REAL evidence_bundle schema. jsonschema-based asserts self-skip if +jsonschema is absent; the seal/verify/round-trip asserts always run. +""" +from __future__ import annotations + +import json +import sys +from pathlib import Path + +from aegis.core.backends.simulator import SimulatorBackend +from aegis.core.llm import ( + AdapterConfig, + BackendSpec, + HookBus, + IdentitySink, + LLMEgress, + LLMRequest, + ModelIdentity, + resolve_model_identity, +) +from aegis.core.orchestrator.pipeline import run_preflight +from aegis.evidence.bundler import verify + +try: + import jsonschema + _HAVE_JSONSCHEMA = True +except Exception: # noqa: BLE001 + _HAVE_JSONSCHEMA = False + +_INTENT = "add vlan 10 to leaf-1" + + +def _schema() -> dict: + return json.loads((_aegis_root() / "evidence" / "schema" / "evidence_bundle.schema.json").read_text("utf-8")) + + +def _validate(bundle: dict) -> None: + if _HAVE_JSONSCHEMA: + jsonschema.validate(bundle, _schema()) + + +def _bundle(**kw) -> dict: + return run_preflight(_INTENT, backend=SimulatorBackend(), lab="clos-evpn", **kw) + + +def _local_identity_dict() -> dict: + return { + "provider": "openai-compatible-local", + "model": "ai/qwen3:latest", + "model_hash": "a" * 64, + "model_hash_kind": "weights-sha256", + "api_version": None, + "capabilities": ["chat", "json"], + "resolved_at_utc": "2026-06-01T10:30:00+00:00", + } + + +# ── real pipeline bundles now carry + seal model_identity at v1.1 ──────────────── + +def test_nl_intent_seals_simulator_identity_v11() -> None: + b = _bundle() + assert b["bundle_version"] == "1.1" + mi = b["change"]["model_identity"] + assert mi["provider"] == "simulator" and mi["model"] == "deterministic-sim-v1" + assert verify(b) + _validate(b) + + +def test_config_import_seals_unknown_identity() -> None: + b = run_preflight("", backend=SimulatorBackend(), lab="single", source="config_import", + imported_configs=[{"device": "r1", "vendor": "frr", "config": "router bgp 65000"}]) + mi = b["change"]["model_identity"] + assert mi["provider"] == "none" and mi["model"] == "operator-supplied" + assert mi["model_hash"] is None and mi["model_hash_kind"] == "identity-claim" + assert verify(b) + _validate(b) + + +def test_explicit_model_identity_passthrough() -> None: + b = _bundle(model_identity=_local_identity_dict()) + mi = b["change"]["model_identity"] + assert mi["provider"] == "openai-compatible-local" + assert mi["model_hash"] == "a" * 64 and mi["model_hash_kind"] == "weights-sha256" + assert verify(b) + _validate(b) + + +def test_model_identity_is_under_the_seal_tamper_detected() -> None: + b = _bundle(model_identity=_local_identity_dict()) + assert verify(b) + b["change"]["model_identity"]["model"] = "evil-swapped-model" + assert not verify(b), "swapping the attested model must break the integrity seal" + + +def test_cloud_identity_claim_is_never_a_faked_weight_hash() -> None: + spec = BackendSpec(provider="anthropic-cloud", base_url="https://api.anthropic.com", + model="claude-haiku-4-5-20251001", api_version="2023-06-01") + mi = resolve_model_identity(spec).to_bundle_dict() + assert mi["model_hash"] is None and mi["model_hash_kind"] == "identity-claim" + b = _bundle(model_identity=mi) + assert b["change"]["model_identity"]["provider"] == "anthropic-cloud" + assert verify(b) + _validate(b) + + +# ── schema: v1.1 requires model_identity; bundle_version pinned to 1.1 ─────────── + +def test_v11_requires_model_identity() -> None: + if not _HAVE_JSONSCHEMA: + print(" SKIP test_v11_requires_model_identity (jsonschema absent)") + return + b = _bundle() + del b["change"]["model_identity"] + try: + jsonschema.validate(b, _schema()) + raise AssertionError("a v1.1 bundle without model_identity must fail validation") + except jsonschema.ValidationError: + pass + + +def test_schema_pins_bundle_version_to_11() -> None: + if not _HAVE_JSONSCHEMA: + print(" SKIP test_schema_pins_bundle_version_to_11 (jsonschema absent)") + return + b = _bundle() + b["bundle_version"] = "1.0" + try: + jsonschema.validate(b, _schema()) + raise AssertionError("schema must pin bundle_version to 1.1") + except jsonschema.ValidationError: + pass + + +# ── the WEDGE round-trip: egress after_response -> IdentitySink -> sealed bundle ─ + +class _FakeGenBackend: + """A generating backend whose model the egress attests; the sink hands it to the seal.""" + + provider = "openai-compatible-local" + + def identity(self) -> ModelIdentity: + return ModelIdentity( + provider=self.provider, model="ai/qwen3:latest", model_hash="b" * 64, + model_hash_kind="weights-sha256", api_version=None, + capabilities=("chat",), resolved_at_utc="2026-06-01T10:30:00+00:00", + ) + + async def complete(self, *, system, user, max_tokens, temperature=0.1): + return "router bgp 65000", self.identity() + + +def test_egress_after_response_sink_into_sealed_bundle() -> None: + import asyncio + + sink = IdentitySink() + hooks = HookBus() + hooks.register("after_response", sink.capture) + cfg = AdapterConfig(chain=(), airgap=False, cloud_available=False, + max_retries=0, base_backoff_s=0.0) + egress = LLMEgress(cfg, (_FakeGenBackend(),), hooks) + asyncio.run(egress.complete(LLMRequest(system="s", user=_INTENT, max_tokens=64))) + + assert sink.last is not None and sink.last.model == "ai/qwen3:latest" + b = _bundle(model_identity=sink.as_bundle_dict()) + mi = b["change"]["model_identity"] + assert mi["model"] == "ai/qwen3:latest" and mi["model_hash_kind"] == "weights-sha256" + assert verify(b) + _validate(b) + + +class _NoAttestBackend: + """A generating backend (nl_intent) that does NOT attest a model identity -- + e.g. the live HttpBackend before the generate-path cutover.""" + def __init__(self) -> None: + self._inner = SimulatorBackend() + def generate_config(self, intent, lab): + return self._inner.generate_config(intent, lab) + def batfish_check(self, configs): + return self._inner.batfish_check(configs) + def spawn_twin(self, lab): + return self._inner.spawn_twin(lab) + def apply_and_converge(self, twin_id, configs): + return self._inner.apply_and_converge(twin_id, configs) + def state_diff(self, twin_id): + return self._inner.state_diff(twin_id) + def teardown_twin(self, twin_id): + return self._inner.teardown_twin(twin_id) + + +def test_nl_intent_without_attestation_is_unattested_not_operator() -> None: + b = run_preflight(_INTENT, backend=_NoAttestBackend(), lab="clos-evpn") + mi = b["change"]["model_identity"] + assert mi["provider"] == "unknown" and mi["model"] == "unattested" + # MUST NOT be confused with a config_import operator-supplied identity: + assert (mi["provider"], mi["model"]) != ("none", "operator-supplied") + assert verify(b) + _validate(b) + + +def test_empty_sink_yields_none() -> None: + assert IdentitySink().as_bundle_dict() is None + + +# ── helpers + runner ──────────────────────────────────────────────────────────── + +def _aegis_root() -> Path: + here = Path(__file__).resolve() + for parent in here.parents: + if (parent / "evidence" / "schema").exists(): + return parent + return here.parent.parent + + +def _run_all() -> int: + funcs = [v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v)] + failures = 0 + for fn in funcs: + try: + fn() + print(f"PASS {fn.__name__}") + except AssertionError as e: + failures += 1 + print(f"FAIL {fn.__name__}: {e}") + except Exception as e: # noqa: BLE001 + failures += 1 + print(f"ERROR {fn.__name__}: {type(e).__name__}: {e}") + print(f"\n{len(funcs) - failures}/{len(funcs)} passed") + return 1 if failures else 0 + + +if __name__ == "__main__": + sys.exit(_run_all())