feat(serve): emit the bounded-autonomy seal on the live preflight path

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",

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)

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()
 
 

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("<b>Bounded-autonomy seal</b> &nbsp; (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()

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.

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
 
 

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: