diff --git a/core/seal/__init__.py b/core/seal/__init__.py index ec98b9b..6b98517 100644 --- a/core/seal/__init__.py +++ b/core/seal/__init__.py @@ -10,6 +10,7 @@ build_claims, canonical_claims_bytes, seal_bundle, + seal_response, verify_seal, ) from .signing import Ed25519Signer, Ed25519Verifier, Signer, Verifier, key_id @@ -21,6 +22,7 @@ "build_claims", "canonical_claims_bytes", "seal_bundle", + "seal_response", "verify_seal", "Signer", "Verifier", diff --git a/core/seal/seal.py b/core/seal/seal.py index da15deb..f7df860 100644 --- a/core/seal/seal.py +++ b/core/seal/seal.py @@ -148,3 +148,14 @@ def verify_seal(bundle: dict, seal: dict, verifier: Verifier) -> SealVerdict: return SealVerdict(True, "OK", "named+hashed model and bounded authority sealed over " "this bundle; signature valid against the pinned key") + + +def seal_response(bundle: dict, signer: Signer, sealed_at_utc: str) -> tuple[dict | None, str | None]: + """Live-path convenience: returns (receipt, None) for a bounded, integrity-valid bundle, + or (None, reason) when the change cannot be sealed (unbounded / tampered). The caller + embeds the receipt under bundle['seal'] (excluded from the bundle hash) so a run emits + its proof in one response.""" + try: + return seal_bundle(bundle, signer, sealed_at_utc=sealed_at_utc), None + except SealError as exc: + return None, str(exc) diff --git a/evidence/bundler.py b/evidence/bundler.py index bed2681..3a5af8b 100644 --- a/evidence/bundler.py +++ b/evidence/bundler.py @@ -17,6 +17,8 @@ def compute_sha256(bundle: dict) -> str: """Hash of the bundle with integrity.sha256 emptied (stable, verifiable).""" clone = json.loads(json.dumps(bundle)) clone["integrity"]["sha256"] = "" + clone.pop("seal", None) # a detached CROSS-3 seal may ride in the response but is NOT + # part of the bundle's own integrity hash (it signs that hash) return hashlib.sha256(_canonical(clone)).hexdigest() diff --git a/evidence/pdf.py b/evidence/pdf.py index 80c7b52..2b534cd 100644 --- a/evidence/pdf.py +++ b/evidence/pdf.py @@ -164,5 +164,21 @@ def render_pdf(bundle: dict) -> bytes: f"{ig.get('signed', False)}", ss["AegisSub"])) el.append(Paragraph(f"sha256 {ig['sha256']}", ss["AegisMono"])) + # bounded-autonomy seal (CROSS-3) — the offline-verifiable receipt, when present + seal = bundle.get("seal") + if seal: + cl = seal.get("claims", {}); m = cl.get("model", {}); au = cl.get("authority", {}) + sig = seal.get("signature", {}) + el.append(Spacer(1, 6)) + el.append(Paragraph("Bounded-autonomy seal   (offline-verifiable receipt)", + ss["AegisSub"])) + el.append(_kv_table([ + ["Model", f"{m.get('provider','?')} · {m.get('model','?')} " + f"({m.get('model_hash_kind','?')})"], + ["Authority", f"required {au.get('required','?')} · ceiling " + f"{au.get('max_authorized','?')} · {'within bound' if au.get('allowed') else 'EXCEEDS'}"], + ["Signature", f"{sig.get('alg','?')} · key {sig.get('key_id','?')}"], + ])) + doc.build(el) return buf.getvalue() diff --git a/serve.py b/serve.py index fac2e1e..ec75a05 100644 --- a/serve.py +++ b/serve.py @@ -12,14 +12,17 @@ AEGIS_PORT=9000 python -m aegis.serve """ from __future__ import annotations +import base64 import os import re +from datetime import datetime, timezone from flask import Flask, Response, jsonify, request from aegis.core.orchestrator.pipeline import run_preflight, PreflightError from aegis.core.backends.simulator import SimulatorBackend from aegis.evidence.pdf import render_pdf +from aegis.core.seal import Ed25519Signer, Ed25519Verifier, seal_response, verify_seal _UI = os.path.join(os.path.dirname(__file__), "ui", "preflight_screen.html") @@ -42,6 +45,25 @@ app.config["MAX_CONTENT_LENGTH"] = 2 * 1024 * 1024 # 2 MB +def _load_signer() -> Ed25519Signer: + """Seal signing key. AEGIS_SEAL_KEY = 64 hex chars (an Ed25519 private seed) pins a + stable key whose receipts verify across restarts; otherwise an EPHEMERAL demo key is + used. Production swaps this for a YubiKey-PIV signer (same Signer protocol).""" + raw = (os.environ.get("AEGIS_SEAL_KEY") or "").strip() + if raw: + try: + return Ed25519Signer.from_private_bytes(bytes.fromhex(raw)) + except Exception as exc: # noqa: BLE001 + print(f"[aegis.seal] AEGIS_SEAL_KEY invalid ({exc}); using an ephemeral key") + print("[aegis.seal] no AEGIS_SEAL_KEY set - EPHEMERAL demo signing key " + "(production uses a YubiKey-PIV key; receipts will not persist across restarts)") + return Ed25519Signer.generate() + + +_SIGNER = _load_signer() +_VERIFIER: Ed25519Verifier = _SIGNER.verifier() + + @app.after_request def _security_headers(resp: Response) -> Response: resp.headers["X-Content-Type-Options"] = "nosniff" @@ -94,6 +116,11 @@ def preflight_run(): return jsonify({"error": str(e), "stage": "guard"}), 400 except Exception as e: # noqa: BLE001 return jsonify({"error": f"{type(e).__name__}: {e}", "stage": "pipeline"}), 502 + # CROSS-3: emit the bounded-autonomy seal in the same response (None if the change is + # not within the autonomy ceiling -- see change.authority for why). Detached: the seal + # rides under bundle["seal"] but is excluded from the bundle's own integrity hash. + seal, _skipped = seal_response(bundle, _SIGNER, datetime.now(timezone.utc).isoformat()) + bundle["seal"] = seal return jsonify(bundle) @@ -114,6 +141,24 @@ def evidence_pdf(): "Content-Disposition": f'attachment; filename="aegis-evidence-{run}.pdf"'}) +@app.get("/api/seal/pubkey") +def seal_pubkey(): + """The pinned public key a client uses to verify seals offline (no secret).""" + return jsonify({"alg": _VERIFIER.alg(), "key_id": _VERIFIER.key_id(), + "public_key_b64": base64.b64encode(_SIGNER.public_key_bytes()).decode("ascii")}) + + +@app.post("/api/seal/verify") +def seal_verify(): + """Offline verification of a {bundle, seal} pair against THIS server's pinned key.""" + data = request.get_json(silent=True) or {} + bundle, seal = data.get("bundle"), data.get("seal") + if not isinstance(bundle, dict) or not isinstance(seal, dict): + return jsonify({"error": "body must be {bundle, seal}"}), 400 + v = verify_seal(bundle, seal, _VERIFIER) + return jsonify({"valid": v.valid, "gate": v.gate, "reason": v.reason}), (200 if v.valid else 422) + + def main(): port = int(os.environ.get("AEGIS_PORT", "8088")) # Bind loopback by default; require an explicit AEGIS_HOST=0.0.0.0 to expose it. diff --git a/tests/api_test.py b/tests/api_test.py index a60d1a0..4f60483 100644 --- a/tests/api_test.py +++ b/tests/api_test.py @@ -111,12 +111,34 @@ def main() -> int: check("bundle.egress_none", (bundle.get("integrity") or {}).get("egress") == "none", str((bundle.get("integrity") or {}).get("egress"))) + # 13 — CROSS-3 seal emitted in the run response (bounded change -> receipt; else null) + au = (bundle.get("change") or {}).get("authority") or {} + seal = bundle.get("seal") + if au.get("allowed") and au.get("effective") != "block": + check("run.seal.present", isinstance(seal, dict) and "signature" in seal, + "expected a seal receipt for a bounded change") + else: + check("run.seal.absent_when_unbounded", seal is None, "unbounded change must not seal") + + # 14 — pinned public key published for offline verification + r = c.get("/api/seal/pubkey") + pk = r.get_json() if r.status_code == 200 else {} + check("seal.pubkey.200", r.status_code == 200, str(r.status_code)) + check("seal.pubkey.keyid", bool(pk.get("key_id")) and pk.get("alg") == "ed25519", str(pk)[:80]) + + # 15 — offline verify endpoint accepts a genuine {bundle, seal} pair + if isinstance(seal, dict): + r = c.post("/api/seal/verify", json={"bundle": bundle, "seal": seal}) + vd = r.get_json() if r.status_code == 200 else {} + check("seal.verify.valid", r.status_code == 200 and vd.get("valid") is True, + f"{r.status_code} {r.get_data(as_text=True)[:120]}") + if FAILS: print("\n=== API ENDPOINT TEST: FAIL ===") for f in FAILS: print(" -", f) return 1 - print(f"\n=== API ENDPOINT TEST: PASS ({CHECKS} checks across 5 routes) ===") + print(f"\n=== API ENDPOINT TEST: PASS ({CHECKS} checks across 7 routes) ===") return 0 diff --git a/tests/seal_test.py b/tests/seal_test.py index 52ae0d4..ed6bd6d 100644 --- a/tests/seal_test.py +++ b/tests/seal_test.py @@ -22,9 +22,10 @@ build_claims, canonical_claims_bytes, seal_bundle, + seal_response, verify_seal, ) -from aegis.evidence.bundler import compute_sha256 +from aegis.evidence.bundler import compute_sha256, verify try: import jsonschema @@ -208,6 +209,31 @@ def test_seal_matches_schema() -> None: jsonschema.validate(seal_bundle(b, s, sealed_at_utc=_AT), _seal_schema()) +# ── live-path wiring: seal_response + embedded-seal hash exclusion ──────────────── + +def test_seal_response_bounded_returns_receipt() -> None: + b = _bounded_bundle(); s = Ed25519Signer.generate() + seal, reason = seal_response(b, s, _AT) + assert reason is None and seal is not None + assert verify_seal(b, seal, s.verifier()).valid + + +def test_seal_response_unbounded_returns_reason() -> None: + b = _unbounded_bundle(); s = Ed25519Signer.generate() + seal, reason = seal_response(b, s, _AT) + assert seal is None and reason and "ceiling" in reason.lower() + + +def test_compute_sha256_excludes_embedded_seal() -> None: + """A detached seal embedded under bundle['seal'] (the live-path response shape) must NOT + change the bundle's own integrity hash, so verify(bundle) still holds.""" + b = _bounded_bundle(); s = Ed25519Signer.generate() + seal, _ = seal_response(b, s, _AT) + b["seal"] = seal + assert verify(b), "embedded seal must be excluded from the bundle integrity hash" + assert verify_seal(b, b["seal"], s.verifier()).valid + + # ── runner ─────────────────────────────────────────────────────────────────────── def _run_all() -> int: