AID-H-033.001: Requested-vs-Effective Model Binding Records
Pillars: app, infra
Lifecycle: operation
Scope boundary
Owns protected, complete, durable and independently verifiable requested-versus-effective route provenance across every invocation and attempt, including recorder role custody, atomic intent/outbox acknowledgment, immutable terminal or unknown history, and independently enumerated coverage. The dispatch prerequisite prevents unrecorded sends; it does not authorize the selected route, decide fallback eligibility, detect malicious route drift, or claim exactly-once provider execution.
AID-H-033.005: AID-H-033.001 owns protected provenance and durable intent acknowledgment; AID-H-033.005 locally verifies the signed contract, resolves the final route, and owns route allow/deny immediately before transport. A recorded decision cannot authorize a route, and an eligible route cannot dispatch without the declared durable acknowledgment. Shared decision evidence does not earn duplicate provenance or enforcement credit.
AID-D-004.006: AID-H-033.001 establishes complete protected route history and canonical v2 producer bindings; AID-D-004.006 independently compares authorized decisions with actual routing, emits drift/visibility signals, and verifies detection/alert effect. The detector can reuse provenance but cannot earn provenance credit by replaying it; the producer cannot mark its own independent population complete.
AID-D-004.004: AID-H-033.001 records model substitution from inside the router; AID-D-004.004 detects source or namespace drift through independent DNS, egress, registry, or CI signals. Internal route evidence and external source-drift observation can reveal different failures.
Warning
Engineering prerequisite
A WAL commit or hash chain does not establish immutable custody. This reference path requires separate privileged recorder and client UIDs, protected storage and service configuration, independently observed terminal states, and a signed failure/capacity policy. It assures the declared local storage restart model, not storage-volume loss. Full disk, outbox or acknowledgment failure stops dispatch. An unresolved crash must not automatically retry; recovery cannot undo a provider send already accepted.Implementation guidance
- AID-H-033.001-G001: Preserve protected, durably acknowledged requested-versus-effective route history across every attempt, bind actual observations and independent ingress coverage, and export within an explicit storage failure model.
AID-H-033.001-G001
Concept: preserve authoritative route provenance without granting route eligibility
Apply this method to an AI gateway that translates model aliases, headers, retries, provider accounts, or fallback routes. A caller can otherwise cause an effective model or provider change while generic best-effort logging reports only the requested model, drops the attempt, or overwrites its predecessor. The protected record must survive the failures claimed by the signed deployment profile and preserve the requested identity, approved decision, each attempted transition, and independently observed outcome. H-033.005 still verifies and enforces route eligibility locally. The recorder's acknowledgment proves durable intent, never that the provider received a request or that an unapproved route is allowed.
Step 1: provision separate recorder and client identities
This implementation uses Python 3.10 or later, its SQLite standard library, Cosign, Linux Unix-domain sockets, and real peer credentials. The recorder exclusively owns its database, WAL, and outbox. Ingress, gateway, observer, verifier, and exporter use distinct service UIDs supplied by the signed policy. Clients can invoke only their assigned append or read operation; they cannot open the database directory or assert a role in JSON. Run the independently credentialed observer from an egress or provider-readback component that the gateway cannot edit. Do not use a read-only connection or a file owned by the gateway as evidence of immutable custody.
set -euo pipefail
sudo groupadd --system aidefend-route-ipc
for role in recorder ingress gateway observer verifier exporter; do
sudo useradd --system --no-create-home --shell /usr/sbin/nologin --groups aidefend-route-ipc "aidefend-route-$role"
done
sudo install -d -m 0700 -o aidefend-route-recorder -g aidefend-route-recorder /var/lib/aidefend-route-ledger
sudo install -d -m 0750 -o aidefend-route-recorder -g aidefend-route-ipc /run/aidefend-route-ledger
# Use existing deployment identities instead when these service accounts already exist.
# Install module, signed policy, bundles, and public keys under root-controlled paths.
The socket directory is owned by the recorder with no group write; its socket grants group connect access only. Restrict access to the service group, maintain OS audit and root-controlled service configuration, and prevent untrusted filesystem ancestors or policy/key replacement. Root or recorder compromise and loss of the storage volume remain outside this local custody claim. SQLite append-denial triggers are additional safeguards; WAL, fsync, and a hash chain do not by themselves make a same-owner file immutable.
Step 2: supply and verify the deployment's signed policy
Provision aidefend.route-recorder-policy.v2 with exactly schema_version, policy_version, subject (application_id, application_version, gateway_id), roles (the five distinct numeric UIDs), route_policies (routing-policy version to SHA-256 of the exact verified contract bytes), failure_model, max_pending_events, max_message_bytes, io_timeout_seconds, and sqlite_timeout_seconds. Obtain UIDs with id -u aidefend-route-ingress and the corresponding service names. Capacity, deadlines, allowed policy versions, rotation, and retention come from the versioned deployment profile, not a universal framework threshold. Ledger metadata and every event bind the canonical recorder-policy digest. Restarting the same ledger with different roles, capacity, subject, or routing-policy population is rejected. For a legitimate policy change, preserve and independently close/read back the old protected segment, provision a new protected segment under the new signed policy, and retain a signed continuity checkpoint linking both policy digests, the prior terminal hash and the complete independently enumerated windows. Resolve unknown dispatches and preserve pending outbox records across the transition; never reset or overwrite old history. This method does not implement automatic segment rotation. The only supported local failure model is protected_local_storage_survives_process_and_host_restart. Validate the storage stack's flush behavior, crash recovery, disk capacity, and confidentiality before relying on it. A deployment requiring volume-loss or site-loss durability must replace this path with an independently durable acknowledgment that meets that stronger model.
Step 3: install the runnable recorder and common consumer module
File: gateway/route_ledger.py. Ledger payloads and exports use explicit version 2 schemas; decision projections and window receipts use version 3. The module refuses duplicate JSON keys, non-finite numbers, mismatched subjects, unsupported field sets, fabricated predecessor bindings, substituted policy digests, and inappropriate peer UIDs. An event and its outbox entry are inserted in the same transaction with WAL and synchronous=FULL; acknowledgment is sent only after commit. Disk-full, commit, outbox, timeout, or acknowledgment failure stops the dispatch path. This is a reference Linux custody path; the portable SQLite tests do not establish Windows service identity separation.
# File: gateway/route_ledger.py
"""Canonical local route recorder; the framework embeds the module above TESTS.
Production custody uses Linux Unix-domain peer credentials and separate UIDs.
Portable tests exercise SQLite durability, state transitions and real crashes;
they do not certify operating-system credential separation on Windows.
"""
from __future__ import annotations
import argparse
import hashlib
import json
import math
import os
from pathlib import Path
import socket
import socketserver
import sqlite3
import stat
import struct
import subprocess
from datetime import datetime, timezone
ROUTE_FIELDS = {
"provider", "provider_account_id", "model_id", "deployment_id",
"endpoint", "region", "route_id", "fallback_decision_id",
}
ROLE_FOR = {
"INGRESS": "ingress", "DECISION": "gateway", "INTENT": "gateway",
"OBSERVED": "observer", "FINAL": "gateway",
}
def canonical(value):
return json.dumps(value, sort_keys=True, separators=(",", ":"),
ensure_ascii=False, allow_nan=False).encode("utf-8")
def strict_json(raw):
def pairs(items):
value = {}
for key, item in items:
if key in value:
raise ValueError("duplicate JSON key")
value[key] = item
return value
def reject(value):
raise ValueError("non-finite JSON constant: " + value)
return json.loads(raw.decode("utf-8", errors="strict"),
object_pairs_hook=pairs, parse_constant=reject)
def text(value, label):
if not isinstance(value, str) or not value.strip():
raise ValueError(label + " must be nonempty text")
return value
def digest(value):
if not isinstance(value, str) or len(value) != 64 or set(value) - set("0123456789abcdef"):
raise ValueError("invalid SHA-256")
return value
def timestamp(value):
parsed = datetime.fromisoformat(text(value, "timestamp").replace("Z", "+00:00"))
if parsed.tzinfo is None:
raise ValueError("timestamp needs timezone")
return parsed
def route(value):
if not isinstance(value, dict) or set(value) != ROUTE_FIELDS:
raise ValueError("effective route field set differs")
return {key: text(value[key], key) for key in sorted(ROUTE_FIELDS)}
def validate_policy(policy):
required = {"schema_version", "policy_version", "subject", "roles",
"route_policies", "failure_model", "max_pending_events",
"max_message_bytes", "io_timeout_seconds", "sqlite_timeout_seconds"}
if not isinstance(policy, dict) or set(policy) != required:
raise ValueError("recorder policy field set differs")
if policy["schema_version"] != "aidefend.route-recorder-policy.v2":
raise ValueError("unsupported recorder policy")
text(policy["policy_version"], "policy version")
if set(policy["subject"]) != {"application_id", "application_version", "gateway_id"}:
raise ValueError("subject field set differs")
for key, value in policy["subject"].items():
text(value, key)
if set(policy["roles"]) != {"ingress", "gateway", "observer", "verifier", "exporter"}:
raise ValueError("role field set differs")
values = list(policy["roles"].values())
if any(type(uid) is not int or uid < 0 for uid in values) or len(set(values)) != len(values):
raise ValueError("roles require distinct OS UIDs")
if not isinstance(policy["route_policies"], dict) or not policy["route_policies"]:
raise ValueError("verified routing-policy population is absent")
for version, sha in policy["route_policies"].items():
text(version, "route policy version")
digest(sha)
if policy["failure_model"] != "protected_local_storage_survives_process_and_host_restart":
raise ValueError("this local example cannot assure loss of its storage volume")
for key in ("max_pending_events", "max_message_bytes"):
if type(policy[key]) is not int or policy[key] < 1:
raise ValueError(key + " must come from positive policy input")
for key in ("io_timeout_seconds", "sqlite_timeout_seconds"):
value = policy[key]
if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or value <= 0:
raise ValueError(key + " must come from finite positive policy input")
return policy
class ClosingConnection(sqlite3.Connection):
def __exit__(self, *args):
try:
return super().__exit__(*args)
finally:
self.close()
class Ledger:
"""Instantiate only inside the privileged recorder, never in the gateway."""
def __init__(self, directory, policy):
self.directory = Path(directory)
self.policy = strict_json(canonical(validate_policy(policy)))
if self.directory.is_symlink():
raise ValueError("recorder directory may not be a symlink")
self.directory.mkdir(mode=0o700, parents=False, exist_ok=True)
self.db = self.directory / "routes.sqlite3"
if self.db.is_symlink():
raise ValueError("database may not be a symlink")
with self.connect() as db:
db.executescript("""
CREATE TABLE IF NOT EXISTS metadata(key TEXT PRIMARY KEY, value BLOB NOT NULL);
CREATE TABLE IF NOT EXISTS events(
seq INTEGER PRIMARY KEY, event_id TEXT NOT NULL UNIQUE,
kind TEXT NOT NULL, invocation_id TEXT NOT NULL, attempt_id TEXT,
record BLOB NOT NULL, sha256 TEXT NOT NULL UNIQUE);
CREATE TABLE IF NOT EXISTS outbox(
event_id TEXT PRIMARY KEY REFERENCES events(event_id), sha256 TEXT NOT NULL,
record BLOB NOT NULL);
CREATE TABLE IF NOT EXISTS export_acks(
event_id TEXT PRIMARY KEY REFERENCES events(event_id), sha256 TEXT NOT NULL,
authenticated_receipt TEXT NOT NULL);
CREATE INDEX IF NOT EXISTS invocation_events ON events(invocation_id,seq);
CREATE TRIGGER IF NOT EXISTS events_no_update BEFORE UPDATE ON events
BEGIN SELECT RAISE(ABORT,'append-only events'); END;
CREATE TRIGGER IF NOT EXISTS events_no_delete BEFORE DELETE ON events
BEGIN SELECT RAISE(ABORT,'append-only events'); END;
CREATE TRIGGER IF NOT EXISTS outbox_no_update BEFORE UPDATE ON outbox
BEGIN SELECT RAISE(ABORT,'append-only outbox'); END;
CREATE TRIGGER IF NOT EXISTS outbox_no_delete BEFORE DELETE ON outbox
BEGIN SELECT RAISE(ABORT,'append-only outbox'); END;
CREATE TRIGGER IF NOT EXISTS acks_no_update BEFORE UPDATE ON export_acks
BEGIN SELECT RAISE(ABORT,'append-only export acknowledgments'); END;
CREATE TRIGGER IF NOT EXISTS acks_no_delete BEFORE DELETE ON export_acks
BEGIN SELECT RAISE(ABORT,'append-only export acknowledgments'); END;
""")
identity = canonical({"schema_version": "aidefend.route-ledger.v2",
"subject": self.policy["subject"],
"recorder_policy_sha256": hashlib.sha256(canonical(self.policy)).hexdigest()})
db.execute("INSERT OR IGNORE INTO metadata VALUES('identity',?)", (identity,))
if db.execute("SELECT value FROM metadata WHERE key='identity'").fetchone()[0] != identity:
raise ValueError("existing ledger belongs to another application, gateway or recorder policy")
def connect(self):
db = sqlite3.connect(self.db, timeout=self.policy["sqlite_timeout_seconds"],
factory=ClosingConnection)
if db.execute("PRAGMA journal_mode=WAL").fetchone()[0].lower() != "wal":
db.close()
raise RuntimeError("WAL unavailable")
db.execute("PRAGMA synchronous=FULL")
db.execute("PRAGMA foreign_keys=ON")
if db.execute("PRAGMA synchronous").fetchone()[0] != 2:
db.close()
raise RuntimeError("durable commit configuration differs")
return db
def role(self, peer_uid, required):
if type(peer_uid) is not int or peer_uid != self.policy["roles"][required]:
raise PermissionError("OS peer is not the required recorder client")
def records(self, db, invocation=None):
rows = db.execute("SELECT seq,record,sha256 FROM events WHERE invocation_id=? ORDER BY seq", (invocation,)).fetchall() if invocation else db.execute("SELECT seq,record,sha256 FROM events ORDER BY seq").fetchall()
result, previous = [], None
for seq, raw, sha in rows:
record = strict_json(raw)
if invocation:
prior = db.execute("SELECT sha256 FROM events WHERE seq=?", (seq - 1,)).fetchone()
previous = prior[0] if prior else None
if (record["seq"] != seq or record["previous_sha256"] != previous
or hashlib.sha256(raw).hexdigest() != sha
or record["subject"] != self.policy["subject"]
or record["recorder_policy_sha256"] != hashlib.sha256(canonical(self.policy)).hexdigest()):
raise ValueError("ledger chain or subject differs")
result.append({**record, "record_sha256": sha})
previous = sha
return result
def enqueue(self, db, event_id, sha, raw):
db.execute("INSERT INTO outbox VALUES(?,?,?)", (event_id, sha, raw))
def append(self, peer_uid, event_id, kind, payload):
if kind not in ROLE_FOR:
raise PermissionError("unknown or mutative recorder operation")
self.role(peer_uid, ROLE_FOR[kind])
text(event_id, "event ID")
if not isinstance(payload, dict) or len(canonical(payload)) > self.policy["max_message_bytes"]:
raise ValueError("payload type or policy byte ceiling differs")
invocation = text(payload.get("invocation_id"), "invocation ID")
attempt = payload.get("attempt_id")
with self.connect() as db:
db.execute("BEGIN IMMEDIATE")
existing = db.execute("SELECT record,sha256 FROM events WHERE event_id=?", (event_id,)).fetchone()
if existing:
old = strict_json(existing[0])
if old["kind"] != kind or old["payload"] != payload:
raise ValueError("event ID reused for different content")
return {"created": False, "event_id": event_id, "record_sha256": existing[1]}
own = self.records(db, invocation)
tail = db.execute("SELECT seq,record,sha256 FROM events ORDER BY seq DESC LIMIT 1").fetchone()
if tail and hashlib.sha256(tail[1]).hexdigest() != tail[2]:
raise ValueError("current ledger tail digest differs")
decisions = {r["attempt_id"]: r for r in own if r["kind"] == "DECISION"}
intents = {r["attempt_id"]: r for r in own if r["kind"] == "INTENT"}
observed = {r["attempt_id"]: r for r in own if r["kind"] == "OBSERVED"}
if kind == "INGRESS":
if set(payload) != {"invocation_id", "requested_model", "received_at"} or own:
raise ValueError("duplicate intake or ingress field set differs")
text(payload["requested_model"], "requested model")
timestamp(payload["received_at"])
else:
if not any(r["kind"] == "INGRESS" for r in own):
raise ValueError("independent ingress binding is missing")
if kind in {"DECISION", "INTENT"} and any(r["kind"] == "FINAL" and
r["payload"]["outcome"] in {"COMPLETE", "VIOLATION"} for r in own):
raise PermissionError("reconciled invocation cannot authorize another dispatch; use new ingress identity")
if kind == "DECISION":
fields = {"invocation_id", "attempt_id", "previous_attempt_id", "decision_id",
"allow", "effective", "policy_version", "policy_sha256", "decided_at", "reason"}
if set(payload) != fields or type(payload["allow"]) is not bool or attempt in decisions:
raise ValueError("decision field set or attempt identity differs")
text(attempt, "attempt ID"); text(payload["decision_id"], "decision ID")
route(payload["effective"]); timestamp(payload["decided_at"]); text(payload["reason"], "reason")
if self.policy["route_policies"].get(payload["policy_version"]) != digest(payload["policy_sha256"]):
raise ValueError("decision is not bound to an approved routing-policy digest")
previous = payload["previous_attempt_id"]
if decisions:
last = next(reversed(decisions))
if previous != last or last not in observed or observed[last]["payload"]["outcome"] == "UNKNOWN":
raise ValueError("retry/fallback needs a resolved preceding attempt")
elif previous is not None:
raise ValueError("first attempt has a fabricated predecessor")
if any(a not in observed or observed[a]["payload"]["outcome"] == "UNKNOWN" for a in intents):
raise ValueError("unresolved dispatch forbids a new attempt")
elif kind == "INTENT":
if set(payload) != {"invocation_id", "attempt_id", "decision_sha256"}:
raise ValueError("intent field set differs")
decision = decisions.get(attempt)
if (not decision or decision["payload"]["allow"] is not True
or payload["decision_sha256"] != decision["record_sha256"]
or attempt in intents or attempt in observed):
raise PermissionError("no fresh allowed decision for dispatch")
pending = db.execute("SELECT count(*) FROM outbox o LEFT JOIN export_acks a USING(event_id) WHERE a.event_id IS NULL").fetchone()[0]
if pending >= self.policy["max_pending_events"]:
raise RuntimeError("signed backlog limit prevents another dispatch")
elif kind == "OBSERVED":
fields = {"invocation_id", "attempt_id", "outcome", "effective", "provider_request_id", "observed_at", "evidence_ref"}
if set(payload) != fields:
raise ValueError("observation field set differs")
text(attempt, "observed attempt ID")
if attempt not in decisions and payload["outcome"] != "DISPATCHED":
raise ValueError("only witnessed dispatch may introduce an unbound attempt")
timestamp(payload["observed_at"]); text(payload["evidence_ref"], "independent evidence")
outcome = payload["outcome"]
old = observed.get(attempt)
if old and (old["payload"]["outcome"] != "UNKNOWN" or outcome == "UNKNOWN"):
raise ValueError("terminal observation is immutable; only UNKNOWN may be reconciled")
if outcome == "DISPATCHED":
route(payload["effective"]); text(payload["provider_request_id"], "provider request identity")
elif outcome in {"NOT_DISPATCHED", "UNKNOWN"}:
if payload["effective"] is not None or payload["provider_request_id"] is not None:
raise ValueError("unknown/non-dispatch must not invent effective provider state")
else:
raise ValueError("unsupported observation outcome")
elif kind == "FINAL":
if set(payload) != {"invocation_id", "outcome"}:
raise ValueError("final field set differs")
outcomes = [observed[a]["payload"]["outcome"] if a in observed else "UNKNOWN" for a in decisions]
violations = any(r["payload"]["outcome"] == "DISPATCHED" and
(a not in decisions or a not in intents or decisions[a]["payload"]["allow"] is not True)
for a, r in observed.items())
derived = "VIOLATION" if violations else "UNRESOLVED" if not outcomes or "UNKNOWN" in outcomes else "COMPLETE"
if payload["outcome"] != derived:
raise ValueError("final outcome is not supported by attempt observations")
seq = tail[0] + 1 if tail else 1
record = {"schema_version": "aidefend.route-ledger-event.v2", "seq": seq,
"event_id": event_id, "kind": kind, "invocation_id": invocation,
"attempt_id": attempt, "subject": self.policy["subject"],
"recorder_policy_sha256": hashlib.sha256(canonical(self.policy)).hexdigest(),
"previous_sha256": tail[2] if tail else None,
"recorded_at": datetime.now(timezone.utc).isoformat(), "payload": payload}
raw = canonical(record); sha = hashlib.sha256(raw).hexdigest()
db.execute("INSERT INTO events VALUES(?,?,?,?,?,?,?)", (seq, event_id, kind, invocation, attempt, raw, sha))
self.enqueue(db, event_id, sha, raw)
db.commit() # The local durable acknowledgment is issued only AFTER commit.
return {"created": True, "event_id": event_id, "record_sha256": sha}
def export(self, peer_uid):
self.role(peer_uid, "verifier")
with self.connect() as db:
db.execute("BEGIN")
records = self.records(db)
return {"schema_version": "aidefend.route-ledger-export.v2", "subject": self.policy["subject"],
"recorder_policy_sha256": hashlib.sha256(canonical(self.policy)).hexdigest(), "events": records}
def export_page(self, peer_uid, after_seq, high_water):
self.role(peer_uid, "verifier")
if type(after_seq) is not int or after_seq < 0 or (high_water is not None and (type(high_water) is not int or high_water < after_seq)):
raise ValueError("export cursor is invalid")
with self.connect() as db:
db.execute("BEGIN")
maximum = db.execute("SELECT coalesce(max(seq),0) FROM events").fetchone()[0]
high_water = maximum if high_water is None else high_water
if high_water > maximum: raise ValueError("export checkpoint exceeds current ledger")
previous = db.execute("SELECT sha256 FROM events WHERE seq=?", (after_seq,)).fetchone()
if after_seq and not previous: raise ValueError("export checkpoint disappeared")
page = {"schema_version": "aidefend.route-ledger-page.v2", "subject": self.policy["subject"],
"recorder_policy_sha256": hashlib.sha256(canonical(self.policy)).hexdigest(),
"after_seq": after_seq, "next_seq": after_seq, "high_water": high_water,
"previous_sha256": previous[0] if previous else None, "complete": after_seq == high_water, "events": []}
prior = page["previous_sha256"]
cursor = db.execute("SELECT seq,record,sha256 FROM events WHERE seq>? AND seq<=? ORDER BY seq", (after_seq, high_water))
for seq, raw, sha in cursor:
row = strict_json(raw)
if seq != page["next_seq"] + 1 or row["previous_sha256"] != prior or hashlib.sha256(raw).hexdigest() != sha:
raise ValueError("export sequence/hash chain differs")
candidate = {**page, "next_seq": seq, "complete": seq == high_water,
"events": page["events"] + [{**row, "record_sha256": sha}]}
if len(canonical({"ok": True, "result": candidate})) + 1 > self.policy["max_message_bytes"]:
if not page["events"]: raise ValueError("a record exceeds the signed export-frame ceiling")
break
page, prior = candidate, sha
return page
def pending(self, peer_uid):
self.role(peer_uid, "exporter")
with self.connect() as db:
page = []
for i, s, r in db.execute("SELECT o.event_id,o.sha256,o.record FROM outbox o LEFT JOIN export_acks a USING(event_id) WHERE a.event_id IS NULL ORDER BY o.rowid"):
candidate = page + [{"event_id": i, "sha256": s, "record": strict_json(r)}]
if len(canonical({"ok": True, "result": candidate})) + 1 > self.policy["max_message_bytes"]:
if not page: raise ValueError("one outbox record exceeds the signed frame ceiling")
break
page = candidate
return page # Acknowledge this page, then poll again; never truncate a record.
def acknowledge_export(self, peer_uid, event_id, sha, authenticated_receipt):
self.role(peer_uid, "exporter")
text(authenticated_receipt, "authenticated external receipt")
with self.connect() as db:
db.execute("BEGIN IMMEDIATE")
row = db.execute("SELECT sha256 FROM outbox WHERE event_id=?", (event_id,)).fetchone()
if not row or row[0] != digest(sha):
raise ValueError("export acknowledgment does not bind exact durable bytes")
prior = db.execute("SELECT sha256,authenticated_receipt FROM export_acks WHERE event_id=?", (event_id,)).fetchone()
if prior and prior != (sha, authenticated_receipt):
raise ValueError("export receipt replacement denied")
db.execute("INSERT OR IGNORE INTO export_acks VALUES(?,?,?)", (event_id, sha, authenticated_receipt))
def guarded_dispatch(commit_intent, send_provider, *, event_id):
"""Call after H-033.005 re-verifies final route eligibility locally.
send_provider closes over the EXACT serialized request bound to that decision.
An existing intent never permits dispatch again; transport/receipt uncertainty
requires independent observation, not an automatic resend.
"""
ack = commit_intent()
if not isinstance(ack, dict) or set(ack) != {"created", "event_id", "record_sha256"}:
raise RuntimeError("local recorder acknowledgment field set differs")
if ack["created"] is not True:
raise PermissionError("existing intent cannot authorize a repeated dispatch")
digest(ack["record_sha256"]); text(ack["event_id"], "event ID")
if ack["event_id"] != text(event_id, "expected intent ID"):
raise PermissionError("durable acknowledgment belongs to another attempt")
return send_provider()
def project_decisions(export, window_id, *, expected_subject, expected_recorder_policy_sha256, invocation_ids=None):
"""Explicit v3 projection. Consumers key attempts by (invocation_id,attempt_id)."""
if set(export) != {"schema_version", "subject", "recorder_policy_sha256", "events"} or export["schema_version"] != "aidefend.route-ledger-export.v2":
raise ValueError("unsupported ledger export")
digest(expected_recorder_policy_sha256)
if (export["subject"] != expected_subject
or export["recorder_policy_sha256"] != expected_recorder_policy_sha256):
raise ValueError("projection belongs to another application/release or recorder policy")
text(window_id, "window ID")
selected = None if invocation_ids is None else set(invocation_ids)
if invocation_ids is not None and (len(selected) != len(invocation_ids) or
any(not isinstance(i, str) or not i for i in invocation_ids)):
raise ValueError("projection invocation population differs")
if not isinstance(export["events"], list): raise ValueError("ledger event population differs")
previous = None; sequence = 0; event_ids = set()
output = []; decision_index = {}; intent_index = {}
for event in export["events"]:
fields = {"schema_version", "seq", "event_id", "kind", "invocation_id", "attempt_id", "subject",
"recorder_policy_sha256", "previous_sha256", "recorded_at", "payload", "record_sha256"}
if (not isinstance(event, dict) or set(event) != fields or
event["schema_version"] != "aidefend.route-ledger-event.v2" or
type(event["seq"]) is not int or event["seq"] != sequence + 1 or
event["previous_sha256"] != previous or event["event_id"] in event_ids or
event["kind"] not in ROLE_FOR or not isinstance(event["payload"], dict) or
event["payload"].get("invocation_id") != event["invocation_id"] or
event["payload"].get("attempt_id") != event["attempt_id"]):
raise ValueError("ledger sequence, chain or event binding differs")
text(event["event_id"], "ledger event ID"); timestamp(event["recorded_at"])
raw = canonical({key: value for key, value in event.items() if key != "record_sha256"})
if (event["subject"] != expected_subject
or event["recorder_policy_sha256"] != expected_recorder_policy_sha256
or hashlib.sha256(raw).hexdigest() != event["record_sha256"]):
raise ValueError("projection source event subject or content digest differs")
previous = event["record_sha256"]; sequence = event["seq"]; event_ids.add(event["event_id"])
if selected is not None and event["invocation_id"] not in selected: continue
key = (event["invocation_id"], event["attempt_id"])
if event["kind"] == "INTENT":
prior = decision_index.get(key)
if (key in intent_index or prior is None or prior["allow"] is not True or
event["payload"]["decision_sha256"] != prior["decision_sha256"]):
raise ValueError("intent is duplicate or lacks its exact preceding allowed decision")
intent_index[key] = event["record_sha256"]
if event["kind"] == "DECISION":
if key in decision_index: raise ValueError("duplicate decision")
p = event["payload"]
output.append({"schema_version": "aidefend.route-decision.v3", "window_id": window_id,
"subject": export["subject"], "recorder_policy_sha256": export["recorder_policy_sha256"],
**p, "decision_sha256": event["record_sha256"], "durable_intent_sha256": None})
decision_index[key] = output[-1]
for key, row in decision_index.items(): row["durable_intent_sha256"] = intent_index.get(key)
return output
def evaluate_routes(decisions, observations, expected_invocations):
"""Independent observed-route comparison; this is NOT detector assurance."""
expected = set(expected_invocations)
if len(expected) != len(expected_invocations) or not expected:
raise ValueError("independent ingress population is empty or duplicated")
wanted, actual, findings, gaps = {}, {}, [], []
for row in decisions:
key = (text(row["invocation_id"], "invocation"), text(row["attempt_id"], "attempt"))
if row["schema_version"] != "aidefend.route-decision.v3" or key in wanted:
raise ValueError("decision identity/schema differs")
if type(row["allow"]) is not bool:
raise ValueError("decision allow is not Boolean")
route(row["effective"]); digest(row["decision_sha256"])
if row["durable_intent_sha256"] is not None: digest(row["durable_intent_sha256"])
wanted[key] = row
for row in observations:
key = (text(row["invocation_id"], "invocation"), text(row["attempt_id"], "attempt"))
if key in actual:
raise ValueError("duplicate independent observation")
actual[key] = row
observed_population = {inv for inv, _ in wanted}
for inv in sorted(expected - observed_population):
gaps.append({"invocation_id": inv, "reason": "missing_decision_population"})
for key in sorted(set(wanted) | set(actual)):
d, o = wanted.get(key), actual.get(key)
if key[0] not in expected or not d:
findings.append({"attempt": list(key), "reason": "unexpected_or_unbound_attempt"})
continue
if o is None or o.get("outcome") == "UNKNOWN":
gaps.append({"attempt": list(key), "reason": "missing_or_unknown_actual_dispatch"})
continue
if o.get("outcome") not in {"DISPATCHED", "NOT_DISPATCHED"}:
raise ValueError("invalid observation outcome")
if o["outcome"] == "NOT_DISPATCHED" and (o.get("effective") is not None or o.get("provider_request_id") is not None):
raise ValueError("non-dispatch observation invents provider state")
if o["outcome"] == "DISPATCHED":
seen = route(o["effective"]); text(o.get("provider_request_id"), "provider request ID")
if d["allow"] is not True:
findings.append({"attempt": list(key), "reason": "denied_attempt_dispatched"})
if d["durable_intent_sha256"] is None:
findings.append({"attempt": list(key), "reason": "dispatched_without_durable_intent"})
for field in sorted(ROUTE_FIELDS):
if seen[field] != d["effective"][field]:
findings.append({"attempt": list(key), "field": field,
"expected": d["effective"][field], "observed": seen[field]})
return {"window_result": "DRIFT_FOUND" if findings else "VISIBILITY_INCOMPLETE" if gaps else "NO_DRIFT_OBSERVED",
"findings": findings, "coverage_gaps": gaps}
def evaluate_window_bytes(captured, evaluation_time):
"""Internal evaluator: captured bytes MUST come from verify_window below."""
labels = {"policy", "inventory", "receipt", "ingress", "decisions", "gateway", "provider", "ledger"}
if set(captured) != labels:
raise ValueError("complete signed source population required")
hashes = {label: hashlib.sha256(raw).hexdigest() for label, raw in captured.items()}
policy = strict_json(captured["policy"])
policy_fields = {"schema_version", "policy_version", "subject", "route_policies", "recorder_policy_sha256",
"max_evidence_age_seconds", "max_future_skew_seconds", "minimum_invocation_count"}
if set(policy) != policy_fields or policy["schema_version"] != "aidefend.route-observation-policy.v2":
raise ValueError("window policy schema differs")
text(policy["policy_version"], "observation policy version")
if set(policy["subject"]) != {"application_id", "application_version", "gateway_id"}:
raise ValueError("observation subject field set differs")
for value in policy["subject"].values(): text(value, "observation subject")
digest(policy["recorder_policy_sha256"])
for key in ("max_evidence_age_seconds", "max_future_skew_seconds"):
value = policy[key]
if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or value < 0:
raise ValueError("invalid signed freshness/skew input")
if policy["max_evidence_age_seconds"] == 0 or type(policy["minimum_invocation_count"]) is not int or policy["minimum_invocation_count"] < 1:
raise ValueError("positive policy age/population floor required")
if not isinstance(policy["route_policies"], dict) or not policy["route_policies"]:
raise ValueError("approved routing-policy population is missing")
for key, value in policy["route_policies"].items(): text(key, "routing-policy version"); digest(value)
inventory = strict_json(captured["inventory"])
if (set(inventory) != {"schema_version", "subject", "gateway_present"}
or inventory["schema_version"] != "aidefend.route-architecture-inventory.v2"
or inventory["subject"] != policy["subject"] or type(inventory["gateway_present"]) is not bool):
raise ValueError("architecture inventory subject/schema differs")
receipt = strict_json(captured["receipt"])
fields = {"schema_version", "subject", "window_id", "window_start", "window_end",
"policy_sha256", "inventory_sha256", "expected_invocation_ids", "closed",
"source_digests", "source_counts", "source_complete", "source_cursors"}
if (set(receipt) != fields or receipt["schema_version"] != "aidefend.route-window-receipt.v3"
or receipt["subject"] != policy["subject"] or receipt["policy_sha256"] != hashes["policy"]
or receipt["inventory_sha256"] != hashes["inventory"] or receipt["closed"] is not True):
raise ValueError("closed receipt is not bound to the exact signed policy/inventory")
window = text(receipt["window_id"], "window ID")
start, end = timestamp(receipt["window_start"]), timestamp(receipt["window_end"])
if start >= end or evaluation_time.tzinfo is None:
raise ValueError("window or evaluation clock is invalid")
expected = receipt["expected_invocation_ids"]
if not isinstance(expected, list) or any(not isinstance(i, str) or not i for i in expected) or len(set(expected)) != len(expected):
raise ValueError("independent invocation population is invalid")
streams = {"ingress", "decisions", "gateway", "provider", "ledger"}
ledger = strict_json(captured["ledger"])
for name in ("source_digests", "source_counts", "source_complete", "source_cursors"):
if not isinstance(receipt[name], dict) or set(receipt[name]) != streams:
raise ValueError("receipt source population differs")
rows, gaps = {}, []
for label in sorted(streams):
rows[label] = ledger["events"] if label == "ledger" else [strict_json(line) for line in captured[label].splitlines() if line.strip()]
count = receipt["source_counts"][label]
cursor = receipt["source_cursors"][label]
if (receipt["source_digests"][label] != hashes[label] or type(count) is not int
or count != len(rows[label]) or type(receipt["source_complete"][label]) is not bool
or not isinstance(cursor, dict) or set(cursor) != {"first", "next"}
or type(cursor["first"]) is not int or type(cursor["next"]) is not int
or cursor["first"] < 0 or cursor["next"] - cursor["first"] != count):
raise ValueError("snapshot digest/count/cursor binding differs")
if not receipt["source_complete"][label]: gaps.append({"source": label, "reason": "source_incomplete"})
for row in rows[label] if label != "ledger" else []:
if row.get("window_id") != window:
raise ValueError("row belongs to another measurement window")
projected = project_decisions(ledger, window, expected_subject=policy["subject"],
expected_recorder_policy_sha256=policy["recorder_policy_sha256"], invocation_ids=expected)
projection_matches = captured["decisions"] == b"".join(canonical(row) + b"\n" for row in projected)
if not inventory["gateway_present"]:
if expected or any(rows.values()):
raise ValueError("absent gateway inventory contains route observations")
return {"window_result": "NOT_APPLICABLE", "findings": [], "coverage_gaps": [],
"subject": policy["subject"], "window_id": window, "input_sha256s": hashes}
age = (evaluation_time - end).total_seconds()
if age > policy["max_evidence_age_seconds"] or age < -policy["max_future_skew_seconds"]:
gaps.append({"source": "window", "reason": "stale_or_future_window"})
if len(expected) < policy["minimum_invocation_count"]:
gaps.append({"source": "ingress", "reason": "empty_or_below_policy_population"})
ingress_ids = []
for row in rows["ingress"]:
if set(row) != {"schema_version", "window_id", "invocation_id", "requested_model", "received_at"} or row["schema_version"] != "aidefend.trusted-route-ingress.v2":
raise ValueError("ingress field set/schema differs")
ingress_ids.append(text(row["invocation_id"], "ingress identity")); text(row["requested_model"], "requested model")
if not start <= timestamp(row["received_at"]) <= end: raise ValueError("ingress lies outside the closed window")
if len(set(ingress_ids)) != len(ingress_ids): raise ValueError("duplicate independent ingress")
if set(ingress_ids) != set(expected): gaps.append({"source": "ingress", "reason": "receipt_ingress_population_differs"})
for row in rows["decisions"]:
required = {"schema_version", "window_id", "invocation_id", "attempt_id", "previous_attempt_id", "decision_id",
"allow", "effective", "policy_version", "policy_sha256", "decided_at", "reason", "decision_sha256",
"subject", "recorder_policy_sha256", "durable_intent_sha256"}
if (set(row) != required or row["subject"] != policy["subject"]
or row["recorder_policy_sha256"] != policy["recorder_policy_sha256"]
or policy["route_policies"].get(row["policy_version"]) != row["policy_sha256"]):
raise ValueError("decision field set or exact routing-policy binding differs")
digest(row["policy_sha256"]); text(row["decision_id"], "decision identity"); text(row["reason"], "decision reason")
if not start <= timestamp(row["decided_at"]) <= end: raise ValueError("decision lies outside the closed window")
for row in rows["gateway"]:
if set(row) != {"window_id", "invocation_id", "attempt_id", "outcome", "effective", "provider_request_id", "observed_at", "decision_sha256"}:
raise ValueError("independent gateway observation field set differs")
digest(row["decision_sha256"])
if not start <= timestamp(row["observed_at"]) <= end: raise ValueError("gateway observation lies outside the window")
result = evaluate_routes(rows["decisions"], rows["gateway"], expected) if expected else {"window_result": "VISIBILITY_INCOMPLETE", "findings": [], "coverage_gaps": []}
if not projection_matches:
result["findings"].append({"source": "decisions", "reason": "decision_projection_differs_from_protected_ledger"})
decisions = {(r["invocation_id"], r["attempt_id"]): r for r in rows["decisions"]}
gateway = {(r["invocation_id"], r["attempt_id"]): r for r in rows["gateway"]}
# Inspect independently attested witnesses as well as decision projections.
# Full exports include other windows; scope findings to this window's IDs.
selected_invocations = set(expected)
latest_observed, latest_final = {}, {}
for event in ledger["events"]:
if event["invocation_id"] not in selected_invocations: continue
payload = event["payload"]
key = (event["invocation_id"], event["attempt_id"])
if event["kind"] == "OBSERVED": latest_observed[key] = payload
if event["kind"] == "FINAL": latest_final[event["invocation_id"]] = payload
if event["kind"] == "OBSERVED" and payload.get("outcome") == "DISPATCHED":
seen = gateway.get(key)
if seen is None or seen["outcome"] != "DISPATCHED":
result["findings"].append({"attempt": list(key), "reason": "ledger_dispatch_absent_from_gateway"})
elif any(payload.get(field) != seen[field] for field in ("effective", "provider_request_id")):
result["findings"].append({"attempt": list(key), "reason": "ledger_gateway_observation_differs"})
if event["kind"] == "FINAL" and payload.get("outcome") == "VIOLATION":
result["findings"].append({"invocation_id": event["invocation_id"], "reason": "ledger_final_violation"})
for key in sorted(set(gateway) | set(decisions) | set(latest_observed)):
observed, seen = latest_observed.get(key), gateway.get(key)
if observed is None:
gaps.append({"attempt": list(key), "reason": "ledger_observation_missing"})
elif observed["outcome"] == "UNKNOWN":
gaps.append({"attempt": list(key), "reason": "ledger_observation_unknown"})
elif observed["outcome"] == "NOT_DISPATCHED" and seen is not None and seen["outcome"] == "DISPATCHED":
result["findings"].append({"attempt": list(key), "reason": "ledger_gateway_outcome_differs"})
for invocation, final in latest_final.items():
if final["outcome"] == "UNRESOLVED":
gaps.append({"invocation_id": invocation, "reason": "ledger_final_unresolved"})
chains = {}
for row in rows["decisions"]:
previous = chains.get(row["invocation_id"])
if row["previous_attempt_id"] != previous:
raise ValueError("retry/fallback predecessor chain differs")
if previous is not None:
prior = gateway.get((row["invocation_id"], previous))
if not prior or prior["outcome"] == "UNKNOWN":
result["findings"].append({"attempt": [row["invocation_id"], row["attempt_id"]], "reason": "retry_after_unknown_dispatch"})
chains[row["invocation_id"]] = row["attempt_id"]
provider = {}
for row in rows["provider"]:
if set(row) != {"window_id", "invocation_id", "attempt_id", "provider_request_id", "effective", "observed_at"}:
raise ValueError("provider observation field set differs")
key = (text(row["invocation_id"], "provider invocation"), text(row["attempt_id"], "provider attempt"))
if key in provider: raise ValueError("duplicate provider attempt")
text(row["provider_request_id"], "provider request identity")
if not isinstance(row["effective"], dict) or set(row["effective"]) != ROUTE_FIELDS - {"route_id", "fallback_decision_id"}:
gaps.append({"attempt": list(key), "reason": "provider_field_visibility_incomplete"})
if not start <= timestamp(row["observed_at"]) <= end: raise ValueError("provider observation lies outside the window")
provider[key] = row
dispatched = {key for key, row in gateway.items() if row["outcome"] == "DISPATCHED"}
for key in sorted(dispatched - set(provider)):
gaps.append({"attempt": list(key), "reason": "missing_independent_provider_readback"})
for key in sorted(set(provider) - dispatched):
result["findings"].append({"attempt": list(key), "reason": "provider_dispatch_without_gateway_observation"})
for key, row in gateway.items():
if key in decisions and row["decision_sha256"] != decisions[key]["decision_sha256"]:
result["findings"].append({"attempt": list(key), "reason": "gateway_decision_digest_differs"})
if key in provider and row["outcome"] == "DISPATCHED":
seen = provider[key]
if row["provider_request_id"] != seen["provider_request_id"]:
result["findings"].append({"attempt": list(key), "reason": "provider_request_identity_differs"})
for field in ROUTE_FIELDS - {"route_id", "fallback_decision_id"}:
if field in seen["effective"] and seen["effective"][field] != row["effective"][field]:
result["findings"].append({"attempt": list(key), "field": "provider." + field,
"expected": row["effective"][field], "observed": seen["effective"][field]})
result["coverage_gaps"].extend(gaps)
result["window_result"] = "DRIFT_FOUND" if result["findings"] else "VISIBILITY_INCOMPLETE" if result["coverage_gaps"] else "NO_DRIFT_OBSERVED"
result.update({"subject": policy["subject"], "window_id": window,
"evaluated_at": evaluation_time.isoformat(), "input_sha256s": hashes})
return result
def verify_window(root, key_root, timeout, maximum, evaluation_time):
names = {"policy": "policy.json", "inventory": "inventory.json", "receipt": "receipt.json",
"ingress": "ingress.jsonl", "decisions": "decisions.jsonl", "gateway": "gateway.jsonl", "provider": "provider.jsonl", "ledger": "ledger.json"}
require_ledger_key_separation(key_root, maximum)
captured = {label: verify_file(Path(root) / name, Path(root) / (name + ".bundle.json"),
Path(key_root) / (label + ".pub"), timeout, maximum)
for label, name in names.items()}
return evaluate_window_bytes(captured, evaluation_time)
def require_ledger_key_separation(key_root, maximum):
# Admission pins this protected role-key directory. Gateway/window signers
# cannot possess the independently protected recorder-attestation private key.
from cryptography.hazmat.primitives.serialization import load_pem_public_key, Encoding, PublicFormat
def identity(role):
key = load_pem_public_key(read_regular(Path(key_root) / (role + ".pub"), maximum))
return key.public_bytes(Encoding.DER, PublicFormat.SubjectPublicKeyInfo)
ledger_key = identity("ledger")
if any(identity(role) == ledger_key for role in ("decisions", "receipt", "gateway", "provider")):
raise PermissionError("ledger attestation requires an independent recorder role key")
def read_regular(path, maximum):
path = Path(path)
before = path.lstat()
if not stat.S_ISREG(before.st_mode) or path.is_symlink() or before.st_size > maximum:
raise ValueError("input is not a bounded regular file")
with path.open("rb") as handle:
opened = os.fstat(handle.fileno()); raw = handle.read(maximum + 1)
after = path.lstat()
identity = lambda s: (s.st_dev, s.st_ino, s.st_size, s.st_mtime_ns)
if len(raw) > maximum or identity(before) != identity(opened) or identity(before) != identity(after):
raise ValueError("input changed during capture")
return raw
def verify_file(path, bundle, key, timeout, maximum):
"""Private snapshots: verify exact captured bytes, never reopen live inputs."""
import tempfile
captured = [read_regular(p, maximum) for p in (path, bundle, key)]
with tempfile.TemporaryDirectory(prefix="route-verify-") as directory:
names = [Path(directory) / name for name in ("input", "bundle", "key")]
for name, raw in zip(names, captured):
name.write_bytes(raw); name.chmod(0o400)
subprocess.run(["cosign", "verify-blob", "--key", str(names[2]),
"--bundle", str(names[1]), str(names[0])],
timeout=timeout, check=True, stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE)
return captured[0]
def call_recorder(socket_path, expected_recorder_uid, request, timeout, maximum):
if not hasattr(socket, "SO_PEERCRED") or os.name != "posix":
raise RuntimeError("production peer-credential transport requires Linux")
raw = canonical(request) + b"\n"
if len(raw) > maximum:
raise ValueError("request exceeds signed byte ceiling")
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as client:
client.settimeout(timeout); client.connect(str(socket_path))
_, uid, _ = struct.unpack("3i", client.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12))
if uid != expected_recorder_uid:
raise PermissionError("recorder socket peer identity differs")
client.sendall(raw)
response = client.makefile("rb").readline(maximum + 1)
if len(response) > maximum or not response.endswith(b"\n"):
raise ValueError("recorder reply is truncated or oversized")
value = strict_json(response)
if set(value) != {"ok", "result"} or value["ok"] is not True:
raise RuntimeError("recorder rejected the operation")
return value["result"]
def collect_export(socket_path, recorder_uid, timeout, frame_limit, snapshot_limit):
cursor, high_water, previous, events, total = 0, None, None, [], 0
subject = policy_sha = None
while True:
page = call_recorder(socket_path, recorder_uid, {"operation": "export",
"arguments": {"after_seq": cursor, "high_water": high_water}}, timeout, frame_limit)
if (page["schema_version"] != "aidefend.route-ledger-page.v2" or page["after_seq"] != cursor
or page["previous_sha256"] != previous or type(page["complete"]) is not bool):
raise ValueError("export page continuity differs")
if high_water is None:
high_water, subject, policy_sha = page["high_water"], page["subject"], page["recorder_policy_sha256"]
if (page["high_water"] != high_water or page["subject"] != subject or page["recorder_policy_sha256"] != policy_sha
or page["next_seq"] != cursor + len(page["events"])):
raise ValueError("export snapshot checkpoint/subject differs")
for row in page["events"]:
raw = canonical({k: v for k, v in row.items() if k != "record_sha256"})
if row["seq"] != cursor + 1 or row["previous_sha256"] != previous or hashlib.sha256(raw).hexdigest() != row["record_sha256"]:
raise ValueError("exported event identity/hash chain differs")
cursor += 1; previous = row["record_sha256"]; total += len(raw)
if total > snapshot_limit: raise ValueError("snapshot exceeds the signed window capacity")
events.append(row)
if page["complete"]:
if cursor != high_water: raise ValueError("premature export completion")
return {"schema_version": "aidefend.route-ledger-export.v2", "subject": subject,
"recorder_policy_sha256": policy_sha, "events": events}
if not page["events"]: raise ValueError("export cursor did not advance")
def serve(directory, socket_path, policy):
if os.name != "posix" or not hasattr(socket, "SO_PEERCRED"):
raise RuntimeError("this production recorder deployment requires Linux")
directory, socket_path = Path(directory), Path(socket_path)
if os.geteuid() in policy["roles"].values():
raise PermissionError("recorder UID must be separate from every client UID")
info = directory.lstat()
if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.geteuid() or info.st_mode & 0o077:
raise PermissionError("recorder must exclusively own a 0700 state directory")
parent_info = socket_path.parent.lstat()
if (not stat.S_ISDIR(parent_info.st_mode) or socket_path.parent.is_symlink()
or parent_info.st_uid != os.geteuid() or parent_info.st_mode & 0o022):
raise PermissionError("recorder must own a socket directory with no group/other write")
if socket_path.exists() or socket_path.is_symlink():
raise ValueError("socket path already exists; verify an orphan before removal")
ledger = Ledger(directory, policy)
os.umask(0o077)
class Handler(socketserver.StreamRequestHandler):
def handle(self):
self.connection.settimeout(policy["io_timeout_seconds"])
_, uid, _ = struct.unpack("3i", self.connection.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12))
try:
raw = self.rfile.readline(policy["max_message_bytes"] + 1)
if len(raw) > policy["max_message_bytes"] or not raw.endswith(b"\n"):
raise ValueError("request exceeds signed transport bounds")
request = strict_json(raw)
if set(request) != {"operation", "arguments"} or not isinstance(request["arguments"], dict):
raise ValueError("request field set differs; caller UID/role is never accepted")
operation, args = request["operation"], request["arguments"]
if operation == "append" and set(args) == {"event_id", "kind", "payload"}:
result = ledger.append(uid, **args)
elif operation == "export" and set(args) == {"after_seq", "high_water"}:
result = ledger.export_page(uid, **args)
elif operation == "pending" and not args:
result = ledger.pending(uid)
elif operation == "acknowledge_export" and set(args) == {"event_id", "sha", "authenticated_receipt"}:
result = ledger.acknowledge_export(uid, **args)
else:
raise PermissionError("recorder operation is not admitted")
reply = {"ok": True, "result": result}
except Exception:
reply = {"ok": False, "result": None}
encoded = canonical(reply) + b"\n"
if len(encoded) > policy["max_message_bytes"]:
encoded = canonical({"ok": False, "result": None}) + b"\n"
self.wfile.write(encoded)
class Server(socketserver.ThreadingUnixStreamServer):
daemon_threads = True
block_on_close = True
with Server(str(socket_path), Handler) as server:
socket_path.chmod(0o660) # Socket directory/group grants are provisioned by the operator.
server.serve_forever()
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--state-directory", type=Path, required=True)
parser.add_argument("--socket", type=Path, required=True)
parser.add_argument("--policy", type=Path, required=True)
parser.add_argument("--policy-bundle", type=Path, required=True)
parser.add_argument("--policy-key", type=Path, required=True)
parser.add_argument("--verify-timeout-seconds", type=float, required=True)
parser.add_argument("--bootstrap-max-bytes", type=int, required=True)
args = parser.parse_args()
if args.verify_timeout_seconds <= 0 or not math.isfinite(args.verify_timeout_seconds) or args.bootstrap_max_bytes < 1:
raise ValueError("root-owned bootstrap deadlines/caps are required")
policy = validate_policy(strict_json(verify_file(args.policy, args.policy_bundle, args.policy_key,
args.verify_timeout_seconds, args.bootstrap_max_bytes)))
serve(args.state_directory, args.socket, policy)
if __name__ == "__main__":
main()
Step 4: start the recorder from root-controlled service configuration
set -euo pipefail
test -n "$AIDEFEND_COSIGN_VERIFY_TIMEOUT_SECONDS"
test -n "$AIDEFEND_BOOTSTRAP_MAX_BYTES"
sudo -u aidefend-route-recorder /usr/bin/python3 /opt/aidefend/gateway/route_ledger.py \
--state-directory /var/lib/aidefend-route-ledger \
--socket /run/aidefend-route-ledger/recorder.sock \
--policy /etc/aidefend/route-recorder-policy.json \
--policy-bundle /etc/aidefend/route-recorder-policy.bundle.json \
--policy-key /etc/aidefend/keys/route-recorder-policy.pub \
--verify-timeout-seconds "$AIDEFEND_COSIGN_VERIFY_TIMEOUT_SECONDS" \
--bootstrap-max-bytes "$AIDEFEND_BOOTSTRAP_MAX_BYTES"
On service restart, an existing socket is intentionally rejected. Confirm that its peer process is gone, inspect unresolved attempts, and let the privileged service manager remove only the verified orphan socket before starting. Never delete or recreate the ledger to bypass a capacity or recovery fault.
Step 5: bind intake, verified decisions, intent, and actual observations
The independent ingress service first appends INGRESS with a stable invocation_id, requested_model, and received_at. Each locally evaluated H-033.005 decision appends DECISION with a unique attempt_id, exact previous_attempt_id (null for the first attempt), decision_id, actual boolean allow, all eight effective route fields, policy_version, exact verified contract policy_sha256, decided_at, and reason. A gateway cannot create independent intake records. Record denied decisions too; they do not receive a dispatch intent.
The .005 gate immediately precedes an authenticated recorder append call for INTENT, whose payload binds the invocation, attempt, and returned DECISION record SHA. Pass that call to guarded_dispatch with the exact expected intent event ID. Its provider-send callback closes over the already checked, immutable final serialized route and payload; no route, credentials, parameters, retry, fallback, or default may change between evaluation and transport. Integrate the callback with the existing provider SDK or transport in the trusted adapter. The example does not invent a provider API. A stale, replayed, malformed, lost, or failed acknowledgment never permits a send.
The independent observer appends OBSERVED using actual egress and provider evidence: DISPATCHED requires the real effective route and provider request ID; NOT_DISPATCHED and UNKNOWN must leave both absent. It supplies an independently retained evidence reference. A process crash after durable intent is an unknown dispatch until external readback resolves it. Do not retry automatically or claim exactly-once transport: a crash may occur before send, during send, or after a provider accepted it. Only an independently resolved preceding attempt permits a new retry/fallback decision, with a new attempt ID and signed eligibility evaluation. Resolve UNKNOWN by appending a new observation, never overwriting its history. FINAL records summarize supported completion or unresolved state and cannot turn planned routing into observed routing.
Step 6: export asynchronously within the declared failure model
The exporter identity polls the bounded pending operation. Export each record's exact bytes and hash to the deployment's existing protected destination; authenticate its acknowledgment and independently read back the exact object before calling acknowledge_export with that receipt. The recorder checks the retained hash and preserves the receipt; destination authentication and retention remain the export adapter's responsibility. No generic sink acknowledgment string proves remote custody. Export failure leaves entries pending; the signed backlog limit blocks further intents while recovery drains verified pages. Local acknowledgment permits dispatch before remote export only under the explicitly declared local-storage survival model. It does not retroactively promise off-host durability or permit discarding unexported events.
Step 7: verify provenance continuity independently
The verifier uses collect_export with signed frame and snapshot ceilings. Paging pins one high-water sequence and checks complete sequence/hash continuity. project_decisions emits aidefend.route-decision.v3, including invocation/attempt/predecessor, eligibility result, exact route, policy digest, decision record digest, source subject, and canonical recorder-policy digest. Supply mandatory expected_subject and expected_recorder_policy_sha256 from the verified deployment/window policy; the projection checks its export header and every source event identity/hash before retaining these bindings in each decision row. The recorder-policy digest is SHA-256 of canonical JSON after verifying its signed bytes, distinct from the full signed routing-contract file digest. Select the signed closed window's population without dropping attempts; compare it to the independently signed ingress IDs and native source cursors. Reconcile all intake records, decisions (including deny), intents, independent observations, unknowns, finals, and authenticated export receipts. Validate role-denial and append-denial, a real crash before/after commit, disk-full, atomic outbox failure, lost acknowledgment, exporter outage/backlog, alias/fallback multi-attempt bindings, unsupported policy digests, and duplicate or missing IDs. Missing actual readback stays unknown. This control establishes durable, complete, protected provenance; .005 separately establishes route eligibility enforcement and D-004.006 separately establishes detection and alert effectiveness.
The v3 decision projection binds each attempt to its exact prior allowed decision and committed durable_intent_sha256; absent intent remains null. A witnessed dispatch with no intent or with a denied decision is retained as a VIOLATION, and cannot be finalized as COMPLETE. The independent detector reports this violation even when the actual provider identity matches the planned route. Migrate decision consumers together; a v2 projection cannot establish this claim. An intent proves committed dispatch permission, not provider execution. Network isolation and exclusive provider credentials remain necessary to force every sender through guarded_dispatch; these examples do not certify that deployment prerequisite.
Retain unbound dispatches and attest protected source history
The independent observer may append a witnessed DISPATCHED attempt even without a decision; preserve its actual route and provider identity. FINAL then records VIOLATION. Later independent evidence may supersede an earlier COMPLETE checkpoint, but a COMPLETE/VIOLATION invocation cannot authorize another dispatch. Start a new independently recorded ingress identity for new work.
The recorder-attestation worker must itself collect the exact protected export through collect_export, including the stable high-water cursor and verified complete sequence/hash chain. It signs those captured ledger.json bytes under the separate protected ledger.pub role. Never sign a ledger supplied by the gateway or an arbitrary caller path. Keep the attestation private key inaccessible to decisions, window-receipt, gateway and provider publishers. Cosign can sign the captured file with cosign sign-blob --key /etc/aidefend/recorder-attestation.key --bundle ledger.json.bundle.json ledger.json; the fixed service invokes this only after authenticated recorder capture and subject/policy/chain validation. This signing infrastructure and Linux custody require deployment verification.
Install cryptography for canonical role-key identity checks. The v3 window receipt binds an eighth input, ledger.json: its digest, event count, cursor range and completeness. The evaluator verifies its separate signature, re-derives the v3 decision projection for the independent window invocation IDs, and compares exact canonical JSONL bytes. A decisions/window publisher cannot turn a fabricated intent digest into a clean result.
The window evaluator also checks attested OBSERVED DISPATCHED and FINAL VIOLATION events for the independently enumerated invocation IDs. A dispatch missing from the gateway snapshot, conflicting observed route/provider identity, or retained violation remains a finding even when the decision projection matches. Events for other invocation populations do not taint this window; a later witness for this population still invalidates a clean claim.
Snapshot capacity: this v2 ledger exporter and verifier require the complete protected segment from sequence 1. Paging bounds each frame, not the total snapshot or verification cost. An export exceeding the signed snapshot ceiling produces a processing/capacity fault; never truncate history, raise the ceiling without reviewing resources, or treat that fault as a clean window. Automatic rotation and cross-segment checkpoint verification are not implemented here. Use an externally qualified segmented recorder/verifier with signed continuity anchors for sustained high-volume service, or bound and close this reference segment within the deployment profile. Keep every unresolved invocation and outbox record recoverable across a segment transition.
Reconcile the latest protected-ledger observation for each attempt and latest FINAL for each invocation in the independently enumerated window population. A ledger NOT_DISPATCHED conflicting with an actual gateway dispatch is a finding. Missing ledger observation, latest UNKNOWN, or latest UNRESOLVED FINAL is a visibility gap even when gateway/provider snapshots report dispatch. A later independently retained resolution replaces an earlier UNKNOWN or UNRESOLVED checkpoint; historical dispatched evidence and violations remain findings. Complete agreement is required for a clean local window observation, and confirmed findings survive other visibility gaps.
SQLite storage prerequisite: the database, WAL and shared-memory files must remain on a protected local filesystem, with all participating processes on the same host. This WAL path does not support NFS/SMB or independent hosts sharing a file. A distributed deployment needs an independently managed transactional store providing the same durable uniqueness/CAS and authenticated readback. See SQLite WAL deployment requirements.
Capacity response and segment closure: route PROCESSING_ERROR and capacity/visibility gaps to the independent production alert destination within the versioned operational deadline. The local recorder does not automatically stop dispatch because a separate detector becomes blind. The deployment must implement and test its required degradation or pause policy. Close a segment only after every independently enumerated invocation has a reconciled FINAL, pending exports have authenticated readback, and continuity anchors are retained. Preserve the old recorder for straggling observations and re-evaluate any affected windows; unresolved calls cannot be silently transferred or discarded. These conditions qualify an external segment lifecycle, not an automatic rotation implementation in this reference.
Open-source tools
- LiteLLM (trusted final-dispatch adapter; custom recorder integration required)
- Envoy (actual upstream observation and gateway integration; protected recorder required)
- OpenTelemetry (secondary telemetry export; not the durable acknowledgment or complete population authority)
- Cosign (signed policy and exported snapshot verification)
Commercial / hosted tools
- Cloudflare AI Gateway (secondary route telemetry; protected recorder and independent readback required)
- Prisma AIRS AI Gateway (formerly Portkey; gateway integration and secondary route telemetry; protected recorder required)
- Kong Konnect AI Gateway (gateway integration; protected recorder and independent readback required)
Threat mappings
MITRE ATLAS
- N/A
MAESTRO
- N/A
OWASP LLM Top 10 2026
- N/A
OWASP ML Top 10 2023
- N/A
OWASP Top 10 for Agentic Applications 2026
- N/A
NIST Adversarial Machine Learning 2025
- N/A
Cisco Integrated AI Security and Safety Framework
- N/A
Google Secure AI Framework 2.0 - Risks
- N/A
Databricks AI Security Framework 3.0
- Governance 4.1: Lack of traceability and transparency of model assets (technical protected requested-versus-effective model-route provenance across attempts; does not enforce route eligibility)