diff --git a/core/seal/__init__.py b/core/seal/__init__.py new file mode 100644 index 0000000..ec98b9b --- /dev/null +++ b/core/seal/__init__.py @@ -0,0 +1,30 @@ +"""core/seal/ — the CROSS-3 bounded-autonomy SEAL: a detached, offline-verifiable receipt +binding the model identity (#4 wedge) + the bounded authority (#5 ceiling) over a specific +evidence bundle, signed with a pinned key (a YubiKey PIV slot in production).""" +from __future__ import annotations + +from .seal import ( + SEAL_VERSION, + SealError, + SealVerdict, + build_claims, + canonical_claims_bytes, + seal_bundle, + verify_seal, +) +from .signing import Ed25519Signer, Ed25519Verifier, Signer, Verifier, key_id + +__all__ = [ + "SEAL_VERSION", + "SealError", + "SealVerdict", + "build_claims", + "canonical_claims_bytes", + "seal_bundle", + "verify_seal", + "Signer", + "Verifier", + "Ed25519Signer", + "Ed25519Verifier", + "key_id", +] diff --git a/core/seal/schema/seal.schema.json b/core/seal/schema/seal.schema.json new file mode 100644 index 0000000..e5d0f02 --- /dev/null +++ b/core/seal/schema/seal.schema.json @@ -0,0 +1,53 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "$id": "https://aegis.local/schema/seal.schema.json", + "title": "AEGIS CROSS-3 Bounded-Autonomy Seal (detached receipt)", + "type": "object", + "additionalProperties": false, + "required": ["seal_version", "claims", "signature"], + "properties": { + "seal_version": { "const": "1.0" }, + "claims": { + "type": "object", + "additionalProperties": false, + "required": ["seal_version", "bundle_sha256", "model", "authority", "sealed_at_utc"], + "properties": { + "seal_version": { "const": "1.0" }, + "bundle_sha256": { "type": "string", "pattern": "^[a-f0-9]{64}$" }, + "model": { + "type": "object", + "additionalProperties": false, + "required": ["provider", "model", "model_hash", "model_hash_kind"], + "properties": { + "provider": { "type": "string" }, + "model": { "type": "string" }, + "model_hash": { "type": ["string", "null"], "pattern": "^[a-f0-9]{64}$" }, + "model_hash_kind": { "enum": ["weights-sha256", "identity-claim"] } + } + }, + "authority": { + "type": "object", + "additionalProperties": false, + "required": ["required", "max_authorized", "allowed", "effective"], + "properties": { + "required": { "enum": ["auto", "hitl", "hotl", "block"] }, + "max_authorized": { "enum": ["auto", "hitl", "hotl", "block"] }, + "allowed": { "type": "boolean" }, + "effective": { "enum": ["auto", "hitl", "hotl", "block"] } + } + }, + "sealed_at_utc": { "type": "string" } + } + }, + "signature": { + "type": "object", + "additionalProperties": false, + "required": ["alg", "key_id", "value"], + "properties": { + "alg": { "enum": ["ed25519", "piv-ecdsa-p256"] }, + "key_id": { "type": "string" }, + "value": { "type": "string" } + } + } + } +} diff --git a/core/seal/seal.py b/core/seal/seal.py new file mode 100644 index 0000000..da15deb --- /dev/null +++ b/core/seal/seal.py @@ -0,0 +1,150 @@ +"""core/seal/seal.py — the CROSS-3 bounded-autonomy SEAL. + +A DETACHED receipt that fuses, into one offline-verifiable object, three facts no +competitor binds together: + (#4 wedge) WHICH model produced the change — provider + model + model-hash + (#5 ceiling) that it was BOUNDED — required authority <= max_authorized + (integrity) over THIS exact evidence bundle — bundle_sha256 +signed with a pinned key (a YubiKey PIV slot in production). + +The receipt is DETACHED — it does not mutate the bundle, so the bundle's own integrity +hash stays stable — and binds to the bundle by its sha256. VERIFY runs OFFLINE against +only the public key (no secret), and a VALID verdict is a portable cryptographic proof +that a named+hashed model produced this exact change AND that it was bounded by the ceiling +— the thing competitors assert but cannot hand an auditor offline. +""" +from __future__ import annotations + +import base64 +import json +from dataclasses import dataclass + +from ...evidence.bundler import verify as verify_bundle +from ..risk import Tier +from .signing import Signer, Verifier + +SEAL_VERSION = "1.0" + + +class SealError(Exception): + """Refused to produce a seal (the bundle is unbounded or fails integrity).""" + + +def _canonical(obj) -> bytes: + return json.dumps(obj, sort_keys=True, separators=(",", ":")).encode("utf-8") + + +def canonical_claims_bytes(claims: dict) -> bytes: + return _canonical(claims) + + +def build_claims(bundle: dict, sealed_at_utc: str) -> dict: + """The exact claim set the seal signs — binds model + authority + the bundle hash.""" + mi = bundle["change"]["model_identity"] + au = bundle["change"]["authority"] + return { + "seal_version": SEAL_VERSION, + "bundle_sha256": bundle["integrity"]["sha256"], + "model": { + "provider": mi["provider"], "model": mi["model"], + "model_hash": mi["model_hash"], "model_hash_kind": mi["model_hash_kind"], + }, + "authority": { + "required": au["required"], "max_authorized": au["max_authorized"], + "allowed": au["allowed"], "effective": au["effective"], + }, + "sealed_at_utc": sealed_at_utc, + } + + +def _is_bounded(authority: dict) -> bool: + if not authority.get("allowed") or authority.get("effective") == "block": + return False + try: + return Tier[str(authority["required"]).upper()] <= Tier[str(authority["max_authorized"]).upper()] + except KeyError: + return False + + +def seal_bundle(bundle: dict, signer: Signer, *, sealed_at_utc: str) -> dict: + """Produce a detached seal receipt for a BOUNDED, integrity-valid bundle. Refuses + (SealError) to seal a tampered bundle or one whose change exceeds the autonomy ceiling + — you never certify a self-escalation.""" + if not verify_bundle(bundle): + raise SealError("refuse to seal: bundle fails its own integrity check") + if not _is_bounded(bundle["change"]["authority"]): + raise SealError("refuse to seal: change is not within the autonomy ceiling " + "(no-self-escalation) — nothing to certify") + claims = build_claims(bundle, sealed_at_utc) + sig = signer.sign(canonical_claims_bytes(claims)) + return { + "seal_version": SEAL_VERSION, + "claims": claims, + "signature": { + "alg": signer.alg(), + "key_id": signer.key_id(), + "value": base64.b64encode(sig).decode("ascii"), + }, + } + + +@dataclass(frozen=True) +class SealVerdict: + valid: bool + gate: str # the gate that decided (G0..G6) or "OK" + reason: str + + +def verify_seal(bundle: dict, seal: dict, verifier: Verifier) -> SealVerdict: + """Offline verification — uses ONLY the pinned public key. Runs gates G0..G6; the first + failure decides.""" + # G0 — structure + try: + claims = seal["claims"] + sig = seal["signature"] + alg, kid, value = sig["alg"], sig["key_id"], sig["value"] + b_sha, model, authority = claims["bundle_sha256"], claims["model"], claims["authority"] + except (KeyError, TypeError): + return SealVerdict(False, "G0", "malformed seal") + + # G1 — key pinning: only the pinned key + alg are trusted + if alg != verifier.alg() or kid != verifier.key_id(): + return SealVerdict(False, "G1", f"unpinned key/alg ({alg}/{kid})") + + # G2 — signature over the canonical claims + try: + ok = verifier.verify(canonical_claims_bytes(claims), base64.b64decode(value)) + except Exception: # noqa: BLE001 — malformed base64, etc. + return SealVerdict(False, "G2", "signature decode error") + if not ok: + return SealVerdict(False, "G2", "signature does not verify against the pinned key") + + # G3 — the seal covers THIS bundle + if b_sha != bundle.get("integrity", {}).get("sha256"): + return SealVerdict(False, "G3", "seal bundle_sha256 does not match the bundle") + + # G4 — model identity bound + honest (an identity-claim never carries a faked weight hash) + mi = bundle.get("change", {}).get("model_identity", {}) + if (model.get("provider") != mi.get("provider") or model.get("model") != mi.get("model") + or model.get("model_hash") != mi.get("model_hash") + or model.get("model_hash_kind") != mi.get("model_hash_kind")): + return SealVerdict(False, "G4", "sealed model identity does not match the bundle") + if model.get("model_hash_kind") == "identity-claim" and model.get("model_hash") is not None: + return SealVerdict(False, "G4", "identity-claim must not carry a weight hash") + + # G5 — bounded autonomy: matches the bundle AND is within the ceiling + au = bundle.get("change", {}).get("authority", {}) + if (authority.get("required") != au.get("required") + or authority.get("max_authorized") != au.get("max_authorized") + or authority.get("allowed") != au.get("allowed") + or authority.get("effective") != au.get("effective")): + return SealVerdict(False, "G5", "sealed authority does not match the bundle") + if not _is_bounded(authority): + return SealVerdict(False, "G5", "sealed change exceeds the autonomy ceiling") + + # G6 — the bundle itself integrity-verifies + if not verify_bundle(bundle): + return SealVerdict(False, "G6", "bundle fails its own integrity check") + + return SealVerdict(True, "OK", "named+hashed model and bounded authority sealed over " + "this bundle; signature valid against the pinned key") diff --git a/core/seal/signing.py b/core/seal/signing.py new file mode 100644 index 0000000..a356e69 --- /dev/null +++ b/core/seal/signing.py @@ -0,0 +1,106 @@ +"""core/seal/signing.py — detached-signature primitives for the CROSS-3 seal. + +A `Signer` produces a detached signature over canonical claim bytes; a `Verifier` checks +one against a PINNED public key. The seal is verifiable OFFLINE with no secret — the +verifier holds only the public key. + +Ed25519Signer/Verifier are the software implementation (real asymmetric crypto via the +`cryptography` library). The HARDWARE path — a YubiKey PIV slot via PKCS#11 C_Sign, with a +slot-9a attestation chain pinned to the Yubico PIV root CA — implements the SAME `Signer` +protocol and is the documented swap-in (it needs the physical token, so it is not built +here; this module ships the seal mechanics + offline verifier the hardware signer plugs +into). No hardware signer is stubbed-as-working — that would be a false-green crypto claim. +""" +from __future__ import annotations + +import hashlib +from typing import Protocol, runtime_checkable + +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import ( + Ed25519PrivateKey, + Ed25519PublicKey, +) + +_ALG = "ed25519" + + +def key_id(public_raw: bytes) -> str: + """Short, stable id of a public key — sha256(raw)[:16]. The verifier PINS this; a seal + signed by any other key is rejected at gate G1.""" + return hashlib.sha256(public_raw).hexdigest()[:16] + + +@runtime_checkable +class Signer(Protocol): + def alg(self) -> str: ... + def key_id(self) -> str: ... + def public_key_bytes(self) -> bytes: ... + def sign(self, payload: bytes) -> bytes: ... + + +@runtime_checkable +class Verifier(Protocol): + def alg(self) -> str: ... + def key_id(self) -> str: ... + def verify(self, payload: bytes, signature: bytes) -> bool: ... + + +class Ed25519Signer: + """Software Ed25519 signer (real asymmetric crypto). Production replaces this with a + PIV/PKCS#11 signer implementing the same protocol; the seal format is identical.""" + + def __init__(self, private_key: Ed25519PrivateKey) -> None: + self._sk = private_key + self._pub = private_key.public_key().public_bytes( + serialization.Encoding.Raw, serialization.PublicFormat.Raw) + + @classmethod + def generate(cls) -> "Ed25519Signer": + return cls(Ed25519PrivateKey.generate()) + + @classmethod + def from_private_bytes(cls, raw: bytes) -> "Ed25519Signer": + return cls(Ed25519PrivateKey.from_private_bytes(raw)) + + def alg(self) -> str: + return _ALG + + def key_id(self) -> str: + return key_id(self._pub) + + def public_key_bytes(self) -> bytes: + return self._pub + + def sign(self, payload: bytes) -> bytes: + return self._sk.sign(payload) + + def verifier(self) -> "Ed25519Verifier": + """The matching offline verifier (public key only).""" + return Ed25519Verifier.from_public_bytes(self._pub) + + +class Ed25519Verifier: + """Offline verifier — holds only the PINNED public key (no secret material).""" + + def __init__(self, public_key: Ed25519PublicKey, public_raw: bytes) -> None: + self._pk = public_key + self._pub = public_raw + + @classmethod + def from_public_bytes(cls, raw: bytes) -> "Ed25519Verifier": + return cls(Ed25519PublicKey.from_public_bytes(raw), raw) + + def alg(self) -> str: + return _ALG + + def key_id(self) -> str: + return key_id(self._pub) + + def verify(self, payload: bytes, signature: bytes) -> bool: + try: + self._pk.verify(signature, payload) + return True + except InvalidSignature: + return False diff --git a/requirements.txt b/requirements.txt index 5d8f2b8..227a5e4 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,4 @@ jsonschema==3.2.0 reportlab>=4.0 flask>=3.0 +cryptography>=42 diff --git a/tests/seal_test.py b/tests/seal_test.py new file mode 100644 index 0000000..52ae0d4 --- /dev/null +++ b/tests/seal_test.py @@ -0,0 +1,231 @@ +"""aegis/tests/seal_test.py — CROSS-3 bounded-autonomy SEAL: detached receipt, offline +verify, gates G0..G6. + +House style: pytest-discoverable test_* + a dep-light __main__ runner +(CI: python3 -m aegis.tests.seal_test). Uses real Ed25519 (cryptography). The schema test +self-skips if jsonschema is absent. +""" +from __future__ import annotations + +import base64 +import json +import random +import sys +from pathlib import Path + +from aegis.core.backends.simulator import SimulatorBackend +from aegis.core.orchestrator.pipeline import run_preflight +from aegis.core.seal import ( + Ed25519Signer, + Ed25519Verifier, + SealError, + build_claims, + canonical_claims_bytes, + seal_bundle, + verify_seal, +) +from aegis.evidence.bundler import compute_sha256 + +try: + import jsonschema + _HAVE = True +except Exception: # noqa: BLE001 + _HAVE = False + +_AT = "2026-06-01T12:00:00+00:00" + + +def _root() -> Path: + here = Path(__file__).resolve() + for p in here.parents: + if (p / "core" / "seal").exists(): + return p + return here.parent.parent + + +def _seal_schema() -> dict: + return json.loads((_root() / "core" / "seal" / "schema" / "seal.schema.json").read_text("utf-8")) + + +def _reseal(b: dict) -> dict: + b["integrity"]["sha256"] = "" + b["integrity"]["sha256"] = compute_sha256(b) + return b + + +def _bounded_bundle(required: str = "hitl", max_auth: str = "hotl") -> dict: + b = run_preflight("add vlan 10 to leaf-1", backend=SimulatorBackend(), lab="clos-evpn") + b["change"]["authority"] = { + "severity": "low", "required": required, "max_authorized": max_auth, + "allowed": True, "effective": required, + "change_class": {"touches_asn": False, "touches_rd_rt": False, + "touches_spine": False, "touches_underlay": False}, + } + return _reseal(b) + + +def _unbounded_bundle() -> dict: + b = _bounded_bundle() + b["change"]["authority"].update(required="block", effective="block", allowed=False) + return _reseal(b) + + +def _signed_seal(bundle: dict, signer: Ed25519Signer, claims: dict) -> dict: + return { + "seal_version": "1.0", "claims": claims, + "signature": {"alg": signer.alg(), "key_id": signer.key_id(), + "value": base64.b64encode(signer.sign(canonical_claims_bytes(claims))).decode("ascii")}, + } + + +# ── claims binding ─────────────────────────────────────────────────────────────── + +def test_build_claims_binds_model_authority_and_hash() -> None: + b = _bounded_bundle() + c = build_claims(b, _AT) + assert c["bundle_sha256"] == b["integrity"]["sha256"] + assert c["model"]["provider"] == b["change"]["model_identity"]["provider"] + assert c["authority"]["required"] == "hitl" + + +# ── round-trip + offline verify ────────────────────────────────────────────────── + +def test_seal_then_verify_is_valid() -> None: + b = _bounded_bundle() + s = Ed25519Signer.generate() + seal = seal_bundle(b, s, sealed_at_utc=_AT) + v = verify_seal(b, seal, s.verifier()) + assert v.valid and v.gate == "OK", v.reason + + +def test_offline_verify_with_public_key_only() -> None: + b = _bounded_bundle() + s = Ed25519Signer.generate() + seal = seal_bundle(b, s, sealed_at_utc=_AT) + pub_only = Ed25519Verifier.from_public_bytes(s.public_key_bytes()) # no secret + assert verify_seal(b, seal, pub_only).valid + + +# ── G1/G2 — wrong key, forged signature ────────────────────────────────────────── + +def test_wrong_key_rejected() -> None: + b = _bounded_bundle() + s = Ed25519Signer.generate() + other = Ed25519Signer.generate() + seal = seal_bundle(b, s, sealed_at_utc=_AT) + v = verify_seal(b, seal, other.verifier()) # not the pinned key + assert not v.valid and v.gate in ("G1", "G2") + + +def test_forged_signature_rejected_g2() -> None: + b = _bounded_bundle() + s = Ed25519Signer.generate() + seal = seal_bundle(b, s, sealed_at_utc=_AT) + seal["signature"]["value"] = "AAAA" + seal["signature"]["value"][4:] # corrupt the sig + v = verify_seal(b, seal, s.verifier()) + assert not v.valid and v.gate == "G2" + + +# ── G3/G6 — bundle binding + integrity ─────────────────────────────────────────── + +def test_tampered_bundle_reseal_breaks_g3() -> None: + b = _bounded_bundle() + s = Ed25519Signer.generate() + seal = seal_bundle(b, s, sealed_at_utc=_AT) + b["change"]["generated_configs"][0]["config"] += "\n# injected" + _reseal(b) # new integrity hash != the seal's claimed bundle_sha256 + v = verify_seal(b, seal, s.verifier()) + assert not v.valid and v.gate == "G3" + + +def test_tampered_bundle_no_reseal_breaks_g6() -> None: + b = _bounded_bundle() + s = Ed25519Signer.generate() + seal = seal_bundle(b, s, sealed_at_utc=_AT) + b["change"]["generated_configs"][0]["config"] += "\n# injected" # no re-seal + v = verify_seal(b, seal, s.verifier()) + assert not v.valid and v.gate in ("G3", "G6") + + +# ── G4 — model swap (re-signed with the right key to get past G2) ───────────────── + +def test_swapped_model_in_claims_rejected_g4() -> None: + b = _bounded_bundle() + s = Ed25519Signer.generate() + claims = build_claims(b, _AT) + claims["model"]["model"] = "evil/other-model" + seal = _signed_seal(b, s, claims) + v = verify_seal(b, seal, s.verifier()) + assert not v.valid and v.gate == "G4" + + +# ── G5 — no self-escalation: never seal / never accept an unbounded change ──────── + +def test_refuse_to_seal_unbounded() -> None: + b = _unbounded_bundle() + s = Ed25519Signer.generate() + try: + seal_bundle(b, s, sealed_at_utc=_AT) + raise AssertionError("must refuse to seal an unbounded change") + except SealError: + pass + + +def test_verify_rejects_unbounded_seal_g5() -> None: + b = _unbounded_bundle() + s = Ed25519Signer.generate() + seal = _signed_seal(b, s, build_claims(b, _AT)) # bypass seal_bundle's refusal + v = verify_seal(b, seal, s.verifier()) + assert not v.valid and v.gate == "G5" + + +# ── 0-violation property: every valid seal certifies a bounded change ───────────── + +def test_every_valid_seal_is_bounded_property() -> None: + s = Ed25519Signer.generate() + ver = s.verifier() + rng = random.Random(11) + sealed = 0 + for _ in range(120): + b = _bounded_bundle(required=rng.choice(["auto", "hitl", "hotl"]), max_auth="hotl") + seal = seal_bundle(b, s, sealed_at_utc=_AT) + v = verify_seal(b, seal, ver) + assert v.valid, v.reason + a = seal["claims"]["authority"] + assert a["allowed"] and a["effective"] != "block" + sealed += 1 + assert sealed == 120 + + +# ── schema ──────────────────────────────────────────────────────────────────────── + +def test_seal_matches_schema() -> None: + if not _HAVE: + print(" SKIP test_seal_matches_schema (jsonschema absent)") + return + b = _bounded_bundle() + s = Ed25519Signer.generate() + jsonschema.validate(seal_bundle(b, s, sealed_at_utc=_AT), _seal_schema()) + + +# ── runner ─────────────────────────────────────────────────────────────────────── + +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())