tegwick
5227fb14e4
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a13eab-1e6a-7b33-9c37-372e90ada3a2
98 lines
5.6 KiB
Python
98 lines
5.6 KiB
Python
"""Guarded registration never changes custody or unrelated registry state."""
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from audit_core.cadence_registration import prepare, strict_json
|
|
|
|
|
|
def fixture():
|
|
frame = {"frame_seconds": 3600}
|
|
entries = [{"source_id": n, "event_class": n, "form": "heartbeat-or-reconciliation",
|
|
"heartbeat": {"event_class": "audit-core.heartbeat", "interval_seconds": 3600,
|
|
"assertion": "nothing-to-report", "missing": "finding",
|
|
"extensions": {"approval-engine": {"asserted_class": n,
|
|
"requested_observer_max_gap_seconds": 7200,
|
|
"effective_deadline_seconds": 10800}}}}
|
|
for n in ("issuance", "use", "supersession", "revocation")]
|
|
entries.append({"source_id": "heartbeat", "event_class": "heartbeat",
|
|
"form": "heartbeat-or-reconciliation", "reconciliation": {
|
|
"compare_local": "local", "compare_observed": "receiver", "divergence": "finding"}})
|
|
doc = {"schema_version": "0.1", "declaration_id": "test", "source": "approval-engine",
|
|
"sources": entries, "extensions": {"netkingdom": {"coordination_frame": frame}}}
|
|
raw = json.dumps(doc).encode()
|
|
proposal = {"sender": "approval-engine", "source": "approval-engine", "tenant": "tenant:platform",
|
|
"may_read": False, "preserve_existing_credentials_and_other_registry_fields": True,
|
|
"coordination_frame": frame, "emission_cadence_sha256": hashlib.sha256(raw).hexdigest(),
|
|
"heartbeat_classes": dict.fromkeys(("issuance", "use", "supersession", "revocation"),7200),
|
|
"receiver_grace_factor": 1.5, "effective_deadline_seconds": 10800}
|
|
rows = [{"name": "sibling", "sources": ["sibling"], "tokens": ["sibling-fixture"]},
|
|
{"name": "approval-engine", "sources": ["approval-engine"], "tenants": ["tenant:platform"],
|
|
"tokens": ["current-fixture", "overlap-fixture"], "may_read": False,
|
|
"evidence_kind": "load-bearing", "secret_policy": "redact", "owner_property": {"keep": True}}]
|
|
return {"senders.json": json.dumps(rows), "unrelated": "preserve"}, raw, proposal
|
|
|
|
|
|
def test_preserves_every_non_metadata_field_and_inputs_and_is_idempotent():
|
|
body, raw, proposal = fixture()
|
|
original = copy.deepcopy(body)
|
|
updated, receipt = prepare(body, raw, proposal)
|
|
before, after = json.loads(body["senders.json"]), json.loads(updated["senders.json"])
|
|
assert body == original and updated["unrelated"] == "preserve" and before[0] == after[0]
|
|
assert {k:v for k,v in after[1].items() if k not in ("heartbeat_classes", "emission_cadence")} == before[1]
|
|
assert "fixture" not in json.dumps(receipt)
|
|
assert prepare(updated, raw, proposal)[0] == updated
|
|
assert prepare(updated, raw, proposal)[1]["changed"] is False
|
|
|
|
|
|
@pytest.mark.parametrize("mutation", ["digest", "read", "tenant", "class", "credentials", "duplicate", "absent"])
|
|
def test_fail_closed(mutation):
|
|
body, raw, proposal = fixture()
|
|
rows = json.loads(body["senders.json"])
|
|
if mutation == "digest": proposal["emission_cadence_sha256"] = "bad"
|
|
if mutation == "read": rows[1]["may_read"] = True
|
|
if mutation == "tenant": rows[1]["tenants"] = ["*"]
|
|
if mutation == "class": rows[1]["heartbeat_classes"] = {"must-not-drop": 3600}
|
|
if mutation == "credentials": rows[1]["tokens"] = []
|
|
if mutation == "duplicate": rows.append(copy.deepcopy(rows[1]))
|
|
if mutation == "absent": rows.pop()
|
|
body["senders.json"] = json.dumps(rows)
|
|
with pytest.raises(ValueError): prepare(body, raw, proposal)
|
|
|
|
|
|
def test_duplicate_json_refused():
|
|
with pytest.raises(ValueError): strict_json('{"name":1,"name":2}')
|
|
|
|
|
|
def test_real_owner_proposal_when_checkout_available():
|
|
root = Path(__file__).resolve().parents[2]
|
|
proposal = root / "net-kingdom/generated/coordination-frame/approval-engine-receiver.json"
|
|
declaration = root / "approval-engine/cadence.yaml"
|
|
if not proposal.exists() or not declaration.exists(): pytest.skip("owner checkouts unavailable")
|
|
body, _, _ = fixture()
|
|
_, receipt = prepare(body, declaration.read_bytes(), strict_json(proposal.read_bytes()))
|
|
assert receipt["effective_deadline_seconds"] == 10800
|
|
|
|
|
|
def test_qonto_proposal_retains_mixed_classification():
|
|
from audit_core.cadence_registration import qonto_metadata
|
|
body, raw, proposal = fixture()
|
|
doc = strict_json(raw)
|
|
doc["source"] = "qonto-assistant"
|
|
doc["stream_id"] = "qonto-assistant.audit"
|
|
kinds = {"audit.allow": "attributive", "audit.deny": "load-bearing", "audit.heartbeat": "load-bearing"}
|
|
doc["sources"] = [{"source_id": name, "event_class": name, "form": "heartbeat-or-reconciliation",
|
|
"extensions": {"netkingdom": {"evidence_class": kind}},
|
|
"heartbeat": {"event_class": "audit.heartbeat", "interval": "PT3600S",
|
|
"assertion": "nothing-to-report", "missing": "finding",
|
|
"extensions": {"qonto-assistant": {"assertion_scope": "no-deny-transitions-in-window"}}}}
|
|
for name, kind in kinds.items()]
|
|
entry = qonto_metadata(doc, "a" * 64)
|
|
assert "tokens" not in entry and entry["may_read"] is False
|
|
assert entry["heartbeat_classes"] == {"audit.deny": 7200}
|
|
assert entry["emission_cadence"] == doc
|
|
doc["sources"][0]["extensions"]["netkingdom"]["evidence_class"] = "load-bearing"
|
|
with pytest.raises(ValueError, match="classification"): qonto_metadata(doc, "a" * 64)
|