From a758b4a79c5dabf87e4ce9c60b52c65be39dd232 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 22 Aug 2026 17:06:03 +0000 Subject: [PATCH 1/2] =?UTF-8?q?feat(scripts):=20edge-ingest=20demo=20?= =?UTF-8?q?=E2=80=94=20the=20$100=20Agent=20story,=20signed=20and=20fed=20?= =?UTF-8?q?to=20the=20platform?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wires the demo assets to the new ingest endpoint (#484): generates the four mandate-draw records of the $100-agent storyline as chained, DSSE-signed OCSF events in exactly the emitter's shape, and plays four acts against POST /api/v1/audit/ingest — the story verifies with chain positions, the replay dedupes, a record edited after signing quarantines with fingerprint mismatch, and a record after a lost epoch verifies with the stream gap surfaced. Offline mode (--out) writes the signed NDJSON + demo public key so the same artifact feeds the immutable-ledger cpex adapter — one generator, both sinks of the joint demo. Auto-registers the edge given a Clerk token, or uses EDGE_INGEST_KEY. The end-to-end test loads the script and drives all four acts through the real endpoint, so demo and verifier can't silently drift apart. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_017L4gxSTs5uuCEANgpL1i9F --- api/tests/test_edge_ingest_demo.py | 78 +++++++ scripts/demo_edge_ingest.py | 339 +++++++++++++++++++++++++++++ 2 files changed, 417 insertions(+) create mode 100644 api/tests/test_edge_ingest_demo.py create mode 100644 scripts/demo_edge_ingest.py diff --git a/api/tests/test_edge_ingest_demo.py b/api/tests/test_edge_ingest_demo.py new file mode 100644 index 0000000..bba5253 --- /dev/null +++ b/api/tests/test_edge_ingest_demo.py @@ -0,0 +1,78 @@ +"""The demo assets feed the real ingest endpoint end-to-end. + +Loads scripts/demo_edge_ingest.py and drives its four acts through the +actual API: the signed $100-agent story verifies, the replay dedupes, +the tampered record quarantines, and the lost-epoch record verifies with +the gap surfaced. If the demo script and the verifier ever drift apart, +this is the test that says so before a live demo does. +""" + +import importlib.util +import pathlib + +import pytest + +from common.models import EdgeAuditEvent, OrgMembership + +_SCRIPT = pathlib.Path(__file__).resolve().parents[2] / "scripts" / "demo_edge_ingest.py" +_spec = importlib.util.spec_from_file_location("demo_edge_ingest", _SCRIPT) +demo = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(demo) + + +@pytest.fixture +def demo_edge(client, db_session, test_user, auth_headers): + """An edge registered with the demo script's chain identity and key.""" + db_session.add(OrgMembership(org_id=test_user.org_id, user_id=test_user.id, role="owner")) + db_session.commit() + resp = client.post( + "/api/v1/edges", + headers=auth_headers, + json={ + "name": "Edge demo — $100 Agent", + "chain_uid": demo.CHAIN_UID, + "verify_key_pem": demo.demo_public_key_pem(), + "key_id": "edge-demo-key-1", + }, + ) + assert resp.status_code == 201, resp.text + return {"Authorization": f"Bearer {resp.json()['ingest_key']}"} + + +def test_demo_story_plays_all_four_acts(client, db_session, demo_edge): + story = demo.build_story() + + # Act 1 — the $100 agent: three draws land, the fourth is denied + act1 = client.post("/api/v1/audit/ingest", headers=demo_edge, content=demo.ndjson(story)) + assert act1.status_code == 202, act1.text + data = act1.json() + assert data["verified"] == 4 and data["quarantined"] == 0 and data["anomalies"] == 0 + assert [r["chain_position"] for r in data["results"]] == [1, 2, 3, 4] + denied = db_session.query(EdgeAuditEvent).filter_by(metadata_uid="draw-004").one() + assert denied.event["action"] == "Denied" + assert denied.event["status_code"] == "mandate_exceeded" + # The draw-receipt join key rides inside the verified bytes + assert denied.event["unmapped"]["cmf.request.request_id"] == "corr-d4e55b37" + + # Act 2 — replay: at-least-once, nothing re-stored + act2 = client.post("/api/v1/audit/ingest", headers=demo_edge, content=demo.ndjson(story)).json() + assert act2["duplicates"] == 4 and act2["verified"] == 0 + assert db_session.query(EdgeAuditEvent).count() == 4 + + # Act 3 — a record edited after signing: quarantined, stored verbatim + tampered = demo.build_tampered(story) + act3 = client.post( + "/api/v1/audit/ingest", headers=demo_edge, content=demo.ndjson([tampered]) + ).json() + assert act3["quarantined"] == 1 + assert "fingerprint mismatch" in act3["results"][0]["reason"] + row = db_session.query(EdgeAuditEvent).filter_by(metadata_uid="draw-evil").one() + assert row.verification_status == "quarantined" + + # Act 4 — after a lost epoch: verified WITH the gap surfaced, chain re-anchored + gap = demo.build_gap_record(story) + act4 = client.post("/api/v1/audit/ingest", headers=demo_edge, content=demo.ndjson([gap])).json() + assert act4["verified"] == 1 and act4["anomalies"] == 1 + result = act4["results"][0] + assert result["status"] == "verified" and result["chain_position"] == 5 + assert "stream_seq gap" in result["anomalies"] diff --git a/scripts/demo_edge_ingest.py b/scripts/demo_edge_ingest.py new file mode 100644 index 0000000..6fdcf00 --- /dev/null +++ b/scripts/demo_edge_ingest.py @@ -0,0 +1,339 @@ +#!/usr/bin/env python3 +"""Edge-ingest demo — the "$100 Agent" story, signed and fed to the platform. + +Generates the four mandate-draw decision records (three allowed, the fourth +denied over-limit) in exactly the shape the cpex-ocsf-audit plugin emits — +attestation chain, DSSE ECDSA-P256 signature, seam stream stamps — and +streams them into ``POST /api/v1/audit/ingest``, where each record is +cryptographically verified on arrival. Then the tamper beat: + + Act 1 four signed records → 4x verified, chain positions 1-4 + Act 2 replay the same batch → 4x duplicate, nothing re-stored + Act 3 a tampered copy (edited after → quarantined: fingerprint mismatch, + signing) stored verbatim with the reason + Act 4 a record after a lost epoch → verified WITH anomaly: the gap is + (stream_seq jumps) surfaced, the chain re-anchors + +Offline mode writes the signed NDJSON + demo public key to disk instead — +the same file feeds the immutable-ledger cpex adapter, so one artifact +drives both sinks of the joint demo: + python3 scripts/demo_edge_ingest.py --out /tmp/spend-story + +Online mode (API running; deps: the api venv — cryptography, rfc8785): + export AI_IDENTITY_TOKEN= # to auto-register the edge + python3 scripts/demo_edge_ingest.py + # or with a pre-registered edge: + EDGE_INGEST_KEY=aid_edge_... python3 scripts/demo_edge_ingest.py + +Optional: API_URL (default http://localhost:8000), DEMO_CHAIN_UID. +The signing key is a fixed demo scalar — never a real credential; the +platform only ever sees the public half. +""" + +from __future__ import annotations + +import argparse +import base64 +import copy +import hashlib +import json +import os +import sys +import urllib.error +import urllib.request + +import rfc8785 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import ec + +API_URL = os.environ.get("API_URL", "http://localhost:8000") +CHAIN_UID = os.environ.get("DEMO_CHAIN_UID", "edge-demo-100-dollar-agent") +STREAM_ID = "gw-demo/boot-1" +TIMEOUT = 20 + +GREEN = "\033[32m" +RED = "\033[31m" +YELLOW = "\033[33m" +BOLD = "\033[1m" +RESET = "\033[0m" + +# Fixed demo scalar (public half registered with the platform; the demo is +# about verification, not key custody — a real edge uses its own keypair). +DEMO_KEY = ec.derive_private_key(0x0100D011A2A6E27, ec.SECP256R1()) + +# (seq, request id, time, amount, allowed, remaining after) +DRAWS = [ + (1, "corr-d1a04f10", "2026-08-22T17:00:01.000Z", "40.00", True, "60.00"), + (2, "corr-d2b93c22", "2026-08-22T17:00:02.000Z", "35.00", True, "25.00"), + (3, "corr-d3c71e08", "2026-08-22T17:00:03.000Z", "20.00", True, "5.00"), + (4, "corr-d4e55b37", "2026-08-22T17:00:04.000Z", "15.00", False, "5.00"), +] + + +def demo_public_key_pem(key=DEMO_KEY) -> str: + return ( + key.public_key() + .public_bytes(serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) + .decode() + ) + + +def _covered_bytes(event: dict) -> bytes: + # The emitter's sign::signing_input rule, verbatim. + ev = copy.deepcopy(event) + att = ev["attestation_list"][0] + att.pop("fingerprint", None) + att.pop("signatures", None) + ev["unmapped"].pop("signature_b64", None) + ev["unmapped"].pop("signature_key_id", None) + return rfc8785.dumps(ev) + + +def _dsse_pae(payload: bytes) -> bytes: + t = b"application/vnd.ocsf.event+json" + return b"DSSEv1 %d %s %d %s" % (len(t), t, len(payload), payload) + + +def _sign_and_chain(event: dict, prev: tuple[str, str] | None, key) -> tuple[dict, tuple[str, str]]: + if prev is not None: + event["attestation_list"][0]["prev_event"] = { + "uid": prev[0], + "type_uid": 600399, + "fingerprint": {"algorithm_id": 3, "value": prev[1]}, + } + cb = _covered_bytes(event) + fingerprint = hashlib.sha256(cb).hexdigest() + event["attestation_list"][0]["fingerprint"] = { + "algorithm_id": 3, + "encoding_id": 1, + "serialization_id": 2, + "value": fingerprint, + } + signature = key.sign(_dsse_pae(cb), ec.ECDSA(hashes.SHA256())) + event["unmapped"]["signature_b64"] = base64.b64encode(signature).decode() + event["unmapped"]["signature_key_id"] = "edge-demo-key-1" + return event, (event["metadata"]["uid"], fingerprint) + + +def _draw_record(seq: int, req: str, ts: str, amount: str, allowed: bool, remaining: str) -> dict: + # Field shapes match the committed cpex-ocsf-audit sample records + # (SAMPLE-OUTPUT-DECISIONS.md records 5 and 3): a delegated mandate + # draw, denied over-limit on the fourth. metadata.uid is what a real + # decision record will carry once spec §4 emitter item 5 lands. + event = { + "action": "Allowed", + "action_id": 1, + "activity_id": 99, + "activity_name": "Invoke Tool", + "actor": {"roles": ["hr"], "user": {"groups": [], "uid": "alice@corp.com"}}, + "api": {"request": {"uid": f"call-{seq}"}}, + "category_uid": 6, + "class_uid": 6003, + "delegation": { + "actor_subject_uid": "agent-7", + "chain": [ + { + "audience": "payments-mcp", + "scopes_granted": ["spend"], + "subject_uid": "agent-7", + "timestamp": "1970-01-01T00:00:00+00:00", + "ttl_seconds": 300, + } + ], + "depth": 1, + "origin_subject_uid": "alice@corp.com", + }, + "disposition": "Allowed", + "disposition_id": 1, + "metadata": { + "uid": f"draw-{seq:03d}", + "product": {"name": "AI Identity OCSF Audit", "vendor_name": "AI Identity"}, + "profiles": ["ai_operation", "security_control"], + "version": "1.9.0", + }, + "severity_id": 1, + "time": ts, + "tool": {"name": "make_purchase", "namespace": "procurement", "uid": f"call-{seq}"}, + "type_uid": 600399, + "attestation_list": [ + {"uid": f"att-draw-{seq:03d}", "chain_uid": CHAIN_UID, "authority_uid": "org-demo"} + ], + "unmapped": { + "cmf.request.request_id": req, + "cmf.security.labels": ["FINANCIAL"], + "cpex.stream": { + "epoch": 1, + "stream_id": STREAM_ID, + "stream_seq": seq, + "emission_seq": seq, + }, + }, + } + if allowed: + event["unmapped"]["cpex.decision"] = { + "steps": [ + {"action": "allowed", "phase": "sequential", "plugin": "mandate-check"}, + {"action": "allowed", "phase": "sequential", "plugin": "cedar-pdp"}, + ], + "verdict": "allow", + } + else: + reason = ( + f"mandate-check: draw {amount} exceeds remaining {remaining} " + f"of 100.00 mandate for agent-7" + ) + event.update( + { + "action": "Denied", + "action_id": 2, + "disposition": "Blocked", + "disposition_id": 2, + "status_code": "mandate_exceeded", + "status_detail": reason, + "status_id": 2, + } + ) + event["unmapped"]["cpex.decision"] = { + "steps": [{"action": "denied", "phase": "sequential", "plugin": "mandate-check"}], + "verdict": {"deny": {"code": "mandate_exceeded", "reason": reason}}, + } + return event + + +def build_story(key=DEMO_KEY) -> list[dict]: + """The four signed, chained draw records of the $100-agent story.""" + events, prev = [], None + for seq, req, ts, amount, allowed, remaining in DRAWS: + event = _draw_record(seq, req, ts, amount, allowed, remaining) + event, prev = _sign_and_chain(event, prev, key) + events.append(event) + return events + + +def build_tampered(story: list[dict]) -> dict: + """A copy of draw 2 edited AFTER signing — the forgery the math catches.""" + tampered = copy.deepcopy(story[1]) + tampered["metadata"]["uid"] = "draw-evil" + tampered["status_detail"] = "amount quietly rewritten" + return tampered + + +def build_gap_record(story: list[dict], key=DEMO_KEY) -> dict: + """A crypto-valid record whose predecessors were lost — seq jumps 4 → 7.""" + event = _draw_record(7, "corr-d7f90a11", "2026-08-22T17:00:07.000Z", "1.00", True, "4.00") + last = story[-1] + prev = (last["metadata"]["uid"], last["attestation_list"][0]["fingerprint"]["value"]) + event, _ = _sign_and_chain(event, prev, key) + return event + + +def ndjson(events: list[dict]) -> bytes: + return b"\n".join(json.dumps(e, sort_keys=True).encode() for e in events) + + +# ── Online demo ───────────────────────────────────────────────────── + + +def _die(msg: str) -> None: + print(f"{RED}FATAL{RESET} {msg}", file=sys.stderr) + sys.exit(2) + + +def _post(path: str, body: bytes, headers: dict) -> dict: + req = urllib.request.Request(API_URL + path, data=body, method="POST", headers=headers) + try: + with urllib.request.urlopen(req, timeout=TIMEOUT) as resp: + return json.loads(resp.read()) + except urllib.error.HTTPError as exc: + _die(f"POST {path} → {exc.code}: {exc.read().decode()[:300]}") + + +def _register_edge() -> str: + token = os.environ.get("AI_IDENTITY_TOKEN") + if not token: + _die( + "set EDGE_INGEST_KEY (pre-registered edge) or AI_IDENTITY_TOKEN " + "(Clerk session token, to register one)" + ) + data = _post( + "/api/v1/edges", + json.dumps( + { + "name": "Edge demo — $100 Agent", + "chain_uid": CHAIN_UID, + "verify_key_pem": demo_public_key_pem(), + "key_id": "edge-demo-key-1", + } + ).encode(), + {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}, + ) + print(f"registered edge {BOLD}{data['id']}{RESET} (chain {CHAIN_UID})") + return data["ingest_key"] + + +def _ingest(ingest_key: str, events: list[dict]) -> dict: + return _post( + "/api/v1/audit/ingest", + ndjson(events), + {"Authorization": f"Bearer {ingest_key}", "Content-Type": "application/x-ndjson"}, + ) + + +def _show(title: str, response: dict) -> None: + print(f"\n{BOLD}{title}{RESET}") + for r in response["results"]: + status = r["status"] + color = {"verified": GREEN, "quarantined": RED}.get(status, YELLOW) + line = f" {color}{status:<11}{RESET} {r.get('uid') or r.get('dedupe_key') or '—'}" + if r.get("chain_position"): + line += f" chain #{r['chain_position']}" + if r.get("reason"): + line += f" {RED}{r['reason']}{RESET}" + if r.get("anomalies"): + line += f" {YELLOW}{r['anomalies']}{RESET}" + print(line) + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument( + "--out", + metavar="PREFIX", + help="offline: write PREFIX.jsonl (signed story) + PREFIX.pub.pem and exit", + ) + args = parser.parse_args() + + story = build_story() + + if args.out: + with open(f"{args.out}.jsonl", "wb") as f: + f.write(ndjson(story) + b"\n") + with open(f"{args.out}.pub.pem", "w") as f: + f.write(demo_public_key_pem()) + print(f"wrote {args.out}.jsonl (4 signed records) and {args.out}.pub.pem") + return + + ingest_key = os.environ.get("EDGE_INGEST_KEY") or _register_edge() + + _show( + "Act 1 — the $100 agent: three draws land, the fourth is denied", _ingest(ingest_key, story) + ) + _show( + "Act 2 — replaying the batch: at-least-once, nothing re-stored", _ingest(ingest_key, story) + ) + _show( + "Act 3 — a record edited after signing: the math catches it", + _ingest(ingest_key, [build_tampered(story)]), + ) + _show( + "Act 4 — after a lost epoch: the gap is surfaced, the chain re-anchors", + _ingest(ingest_key, [build_gap_record(story)]), + ) + print( + f"\n{GREEN}done{RESET} — every outcome above is stored org-scoped; " + "quarantined and anomalous rows carry their reasons verbatim." + ) + + +if __name__ == "__main__": + main() From 822f042bd0366c29cdd1fe4428ceaf2b699f5c1b Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 22 Aug 2026 17:06:33 +0000 Subject: [PATCH 2/2] docs(changelog): entry for the edge-ingest demo (#485) Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_017L4gxSTs5uuCEANgpL1i9F --- CHANGELOG.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index b47a896..93e957d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,8 @@ Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). ## [Unreleased] ### Added +- **Edge-ingest demo — the $100 Agent story, signed end-to-end.** `scripts/demo_edge_ingest.py` generates the four mandate-draw records of the spend storyline as chained, DSSE-signed OCSF events (the emitter's exact shape) and plays four acts against `POST /api/v1/audit/ingest`: the story verifies with chain positions, the replay dedupes, a record edited after signing quarantines with `fingerprint mismatch`, and a record after a lost epoch verifies with the stream gap surfaced. Offline mode (`--out`) writes the signed NDJSON + demo public key, so the same artifact feeds the immutable-ledger cpex adapter — one generator, both sinks of the joint demo. An end-to-end test drives all four acts through the real endpoint so the demo and the verifier can't silently drift apart. Script + test only — no product impact. (#485) + - **OCSF edge ingest — the platform now accepts signed audit records from enforcement points it doesn't host.** Component A of the edge-enforcement spec (#483): org admins register an edge deployment (`POST /api/v1/edges` — globally-unique `chain_uid`, validated ECDSA P-256 verification key, show-once `aid_edge_` credential hashed at rest), and the edge streams NDJSON OCSF records to `POST /api/v1/audit/ingest`, where every record is verified **on arrival** — covered bytes reconstructed exactly per the emitter's `sign::signing_input` (RFC 8785 minus the attestation envelope), fingerprint recomputed, DER ECDSA-P256 signature checked over the DSSE PAE, chain continuity checked against the last verified head, `stream_seq` density per stream. Integrity failures quarantine the record (stored verbatim with the reason, never dropped, never advancing chain state — a forged record can't re-root the chain); continuity observations on a crypto-valid record (gap, discontinuity after a crashed edge epoch) land as `anomalies` on a verified row and the chain re-anchors, so one lost epoch can't cascade-quarantine everything after it. Replays dedupe on `metadata.uid` (stream-stamp fallback) and report `duplicate`. Two new tables via forward-only migration `g4b5c6d7e8f9`; no new dependencies. 19 new tests sign real chained records the way the Rust emitter does and cover authz, tamper/wrong-key/unsigned quarantine, gap/re-anchor, replay, caps, and cross-org isolation. (#484) - **Edge-enforcement spec drafted — the on-prem/VPC roadmap item gets its concrete shape.** `docs/specs/praxis-edge-enforcement.md` designs the split the CPEX/Praxis collaboration makes possible: enforcement runs in the customer's cluster (Praxis + PPE with two AI Identity plugins), while authority and evidence stay with the platform — raw payloads never leave the cluster, only signed decision records carrying content hashes plus draw-settlement metadata. Three gated components: an org-scoped OCSF ingest endpoint that verifies fingerprints, DSSE, chain continuity, and `stream_seq` density on arrival (gaps quarantined and surfaced, never laundered — a gap is itself evidence); a `cpex-plugin-mandate` crate that verifies Biscuits offline via the published root public key (the token's embedded ceiling checks make forged caps fail cryptographically) but settles draws against the mandate service, keeping cumulative spend single-writer and the Ed25519 root key out of the edge entirely — fail-closed on every path; and a reference deployment smoke-tested by the committed $100-mandate scenario. Build is explicitly gated behind the praxis#11 port verification per `PRAXIS-PORT-PLAN.md`'s sequencing rule, and September stays committed to v0.5.0 — components target the v0.6.0 window alongside agent spend control. Design only — no product impact. (#483)