tegwick
5227fb14e4
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a13eab-1e6a-7b33-9c37-372e90ada3a2
137 lines
7.2 KiB
Python
137 lines
7.2 KiB
Python
"""Credential-preserving Approval Engine registry metadata update.
|
|
|
|
No credential is created, returned, logged or written to a local file. The
|
|
caller owns the attended custody envelope and persists only ``receipt``.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
from typing import Any
|
|
|
|
from audit_core.emission_cadence import parse_declaration
|
|
from audit_core.senders import SenderRegistry
|
|
|
|
CLASSES = {"issuance", "use", "supersession", "revocation"}
|
|
|
|
|
|
def strict_json(raw: bytes | str) -> Any:
|
|
def unique(pairs):
|
|
result = {}
|
|
for key, value in pairs:
|
|
if key in result:
|
|
raise ValueError("duplicate_json_key")
|
|
result[key] = value
|
|
return result
|
|
return json.loads(raw, object_pairs_hook=unique)
|
|
|
|
|
|
def prepare(body: dict, declaration_raw: bytes, proposal: dict) -> tuple[dict, dict]:
|
|
"""Return an in-memory KV data body and a nonsecret receipt.
|
|
|
|
Refuse absent/ambiguous identity, wrong scope, credential absence, stale
|
|
source pins, widened class inventory, or proposal/declaration disagreement.
|
|
Preserve every sibling sender, credential, expiry, policy and KV property.
|
|
"""
|
|
digest = hashlib.sha256(declaration_raw).hexdigest()
|
|
if proposal.get("emission_cadence_sha256") != digest:
|
|
raise ValueError("declaration_digest_mismatch")
|
|
if (proposal.get("sender"), proposal.get("source"), proposal.get("tenant"),
|
|
proposal.get("may_read")) != ("approval-engine", "approval-engine", "tenant:platform", False):
|
|
raise ValueError("unexpected_proposal_scope")
|
|
if proposal.get("preserve_existing_credentials_and_other_registry_fields") is not True:
|
|
raise ValueError("preservation_not_declared")
|
|
frame = proposal["coordination_frame"]
|
|
f = frame["frame_seconds"]
|
|
if type(f) is not int or f <= 0:
|
|
raise ValueError("invalid_frame")
|
|
waits = proposal["heartbeat_classes"]
|
|
if waits != dict.fromkeys(CLASSES, 2 * f):
|
|
raise ValueError("unexpected_class_waits")
|
|
if (proposal.get("receiver_grace_factor"), proposal.get("effective_deadline_seconds")) != (1.5, 3 * f):
|
|
raise ValueError("unexpected_deadline")
|
|
declaration = strict_json(declaration_raw)
|
|
parsed = parse_declaration(declaration)
|
|
if parsed.source != "approval-engine" or parsed.rates:
|
|
raise ValueError("unexpected_declaration_source_or_rate")
|
|
declared = {e["event_class"]: e for e in parsed.other_entries}
|
|
if len(declared) != len(parsed.other_entries) or set(declared) != CLASSES | {"heartbeat"}:
|
|
raise ValueError("unexpected_declaration_inventory")
|
|
if declaration.get("extensions", {}).get("netkingdom", {}).get("coordination_frame") != frame:
|
|
raise ValueError("frame_provenance_mismatch")
|
|
for name in CLASSES:
|
|
beat = declared[name].get("heartbeat", {})
|
|
ext = beat.get("extensions", {}).get("approval-engine", {})
|
|
if (beat.get("event_class"), beat.get("interval_seconds"), ext.get("asserted_class"),
|
|
ext.get("requested_observer_max_gap_seconds"), ext.get("effective_deadline_seconds")) != (
|
|
"audit-core.heartbeat", f, name, 2 * f, 3 * f):
|
|
raise ValueError("declaration_wait_mismatch")
|
|
rows = strict_json(body["senders.json"])
|
|
if not isinstance(rows, list) or any(not isinstance(row, dict) for row in rows):
|
|
raise ValueError("invalid_registry")
|
|
indices = [i for i, row in enumerate(rows) if row.get("name") == "approval-engine"]
|
|
if len(indices) != 1:
|
|
raise ValueError("sender_absent_or_ambiguous")
|
|
i = indices[0]
|
|
before = rows[i]
|
|
if (before.get("sources"), before.get("tenants"), before.get("may_write", True),
|
|
before.get("may_read", False)) != (["approval-engine"], ["tenant:platform"], True, False):
|
|
raise ValueError("existing_sender_scope_mismatch")
|
|
if not (before.get("tokens") or before.get("token")):
|
|
raise ValueError("existing_credentials_required")
|
|
existing_classes = set(before.get("heartbeat_classes", {}))
|
|
if existing_classes - CLASSES:
|
|
raise ValueError("would_drop_existing_heartbeat_class")
|
|
updated_rows = copy.deepcopy(rows)
|
|
updated_rows[i]["heartbeat_classes"] = copy.deepcopy(waits)
|
|
updated_rows[i]["emission_cadence"] = declaration
|
|
# Validate the entire candidate against the actual receiver parser. No
|
|
# private data appears in exceptions emitted by the attended wrapper.
|
|
SenderRegistry.from_env({"AUDIT_CORE_SENDERS": json.dumps(updated_rows)})
|
|
updated = copy.deepcopy(body)
|
|
changed = updated_rows != rows
|
|
if changed:
|
|
updated["senders.json"] = json.dumps(updated_rows, separators=(",", ":"))
|
|
receipt = {"sender": "approval-engine", "source": "approval-engine",
|
|
"tenant": "tenant:platform", "may_read": False,
|
|
"emission_cadence_sha256": digest, "heartbeat_classes": waits,
|
|
"effective_deadline_seconds": 3 * f, "changed": changed,
|
|
"credentials_preserved": True, "other_registry_fields_preserved": True}
|
|
return updated, receipt
|
|
|
|
|
|
def qonto_metadata(declaration: dict, declaration_sha256: str) -> dict:
|
|
"""Build a nonsecret sender proposal; custody owner adds tokens in memory.
|
|
|
|
The digest is of the owner's YAML bytes, which the custody caller verifies
|
|
before parsing; JSON serialization cannot stand in for that source digest.
|
|
"""
|
|
parsed = parse_declaration(declaration)
|
|
if parsed.source != "qonto-assistant" or parsed.stream_id != "qonto-assistant.audit" or parsed.rates:
|
|
raise ValueError("unexpected_qonto_declaration")
|
|
entries = {e["event_class"]: e for e in parsed.other_entries}
|
|
kinds = {"audit.allow": "attributive", "audit.deny": "load-bearing", "audit.heartbeat": "load-bearing"}
|
|
if set(entries) != set(kinds) or len(entries) != len(parsed.other_entries):
|
|
raise ValueError("unexpected_qonto_inventory")
|
|
frame = declaration.get("extensions", {}).get("netkingdom", {}).get("coordination_frame", {})
|
|
f = frame.get("frame_seconds")
|
|
if type(f) is not int or f <= 0:
|
|
raise ValueError("invalid_qonto_frame")
|
|
if len(declaration_sha256) != 64 or any(c not in "0123456789abcdef" for c in declaration_sha256):
|
|
raise ValueError("invalid_qonto_source_digest")
|
|
from audit_core.emission_cadence import duration_seconds
|
|
for name, kind in kinds.items():
|
|
entry = entries[name]
|
|
if entry.get("extensions", {}).get("netkingdom", {}).get("evidence_class") != kind:
|
|
raise ValueError("qonto_classification_mismatch")
|
|
beat = entry.get("heartbeat", {})
|
|
if beat.get("event_class") != "audit.heartbeat" or duration_seconds(beat.get("interval")) != f:
|
|
raise ValueError("qonto_cadence_mismatch")
|
|
if beat.get("extensions", {}).get("qonto-assistant", {}).get("assertion_scope") != "no-deny-transitions-in-window":
|
|
raise ValueError("qonto_assertion_scope_mismatch")
|
|
return {"name": "rapp-qonto-collector", "sources": ["qonto-assistant"],
|
|
"tenants": ["tenant:friendly:binky"], "may_write": True, "may_read": False,
|
|
"secret_policy": "redact", "evidence_kind": "load-bearing",
|
|
"heartbeat_classes": {"audit.deny": 2 * f},
|
|
"emission_cadence": copy.deepcopy(declaration)}
|