tegwick
64509d2c40
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a140c4-354a-7823-8f0d-8f717397f786
90 lines
4.6 KiB
Python
90 lines
4.6 KiB
Python
"""AUDIT-WP-0013-T02: wire compatibility, never live sender admission.
|
|
|
|
Fixture field mapping is pinned in docs/flex-auth-secrets-engine-admission.md.
|
|
All identities/tokens below exist only in the isolated test receiver.
|
|
"""
|
|
from datetime import datetime, timedelta, timezone
|
|
from urllib.parse import urlencode
|
|
|
|
import pytest
|
|
|
|
from audit_core.ingestion import IngestionApplication
|
|
from audit_core.senders import SenderIdentity, SenderRegistry
|
|
from audit_core.sqlite_backend import SQLiteAuditBackend
|
|
from audit_core.stream_findings import evaluate, NEVER_HEARTBEAT
|
|
from test_stream_findings import invoke
|
|
|
|
SOURCE = "flex-auth.secrets-engine"
|
|
RARE = tuple("flex-auth.decision." + name for name in ("deny", "redact", "not_applicable", "audit_only"))
|
|
CLASSES = RARE + ("flex-auth.decision.allow",)
|
|
|
|
|
|
@pytest.fixture
|
|
def receiver(tmp_path):
|
|
writer = SenderIdentity(name=SOURCE, tokens=("fixture-writer",), sources=frozenset({SOURCE}),
|
|
tenants=frozenset({"tenant:platform"}), may_read=False,
|
|
evidence_kind="load-bearing", heartbeat_classes=tuple((c, 86400) for c in RARE))
|
|
reader = SenderIdentity(name="fixture-independent-reader", tokens=("fixture-reader",),
|
|
sources=frozenset({SOURCE}), tenants=frozenset({"tenant:platform"}),
|
|
may_write=False, may_read=True)
|
|
return IngestionApplication(SQLiteAuditBackend(str(tmp_path / "events.db")), SenderRegistry([writer, reader])), writer
|
|
|
|
|
|
def envelope(kind, event_id="fixture-event"):
|
|
# outbox.go Event + event(): no receiver-derived fields or invented sequence.
|
|
return dict(id=event_id, type=kind, source=SOURCE, subject="openbao/metadata/fixture",
|
|
tenant="tenant:platform", correlation_id="fixture-decision",
|
|
occurred_at="2026-10-16T00:00:00.123456789Z",
|
|
data={"id": "fixture-decision", "effect": kind.rsplit(".", 1)[-1]})
|
|
|
|
|
|
def post(app, event):
|
|
return invoke(app, "/v1/events", token="fixture-writer", method="POST", payload=event, key=event["id"])
|
|
|
|
|
|
def count(app, since, until, source=SOURCE, tenant="tenant:platform"):
|
|
return invoke(app, "/v1/reconciliation?" + urlencode(dict(source=source, tenant=tenant, since=since, until=until)), token="fixture-writer")
|
|
|
|
|
|
@pytest.mark.parametrize("kind", CLASSES)
|
|
def test_exact_pin_wire_duplicate_and_independent_readback(receiver, kind):
|
|
app, _ = receiver
|
|
event = envelope(kind)
|
|
assert post(app, event)[0].startswith("202")
|
|
assert post(app, event)[0].startswith("200")
|
|
assert post(app, dict(event, subject="changed"))[0].startswith("409")
|
|
assert invoke(app, "/v1/events/fixture-event", token="fixture-writer")[0].startswith("403")
|
|
status, retained = invoke(app, "/v1/events/fixture-event", token="fixture-reader")
|
|
assert status.startswith("200")
|
|
assert (retained["source"], retained["tenant"], retained["action"]) == (SOURCE, "tenant:platform", kind)
|
|
assert retained["details"]["data"] == event["data"]
|
|
accepted = datetime.fromisoformat(retained["accepted_at"])
|
|
before, after = ((accepted + timedelta(seconds=n)).isoformat() for n in (-1, 1))
|
|
status, counts = count(app, before, after)
|
|
assert status.startswith("200")
|
|
assert counts["counts"] == [{"class": kind, "count": 1}]
|
|
# Acceptance-time [since, until): source occurred_at is deliberately older.
|
|
assert count(app, before, accepted.isoformat())[1]["counts"] == []
|
|
assert count(app, accepted.isoformat(), after)[1]["counts"] == [{"class": kind, "count": 1}]
|
|
|
|
|
|
def test_wrong_scope_refusals_are_not_empty_counts(receiver):
|
|
app, _ = receiver
|
|
assert post(app, dict(envelope(CLASSES[0]), source="flex-auth.other"))[0].startswith("400")
|
|
assert post(app, dict(envelope(CLASSES[0]), tenant="tenant:other"))[0].startswith("400")
|
|
window = ("2026-10-15T00:00:00Z", "2027-01-01T00:00:00Z")
|
|
assert count(app, *window, source="flex-auth.other")[0].startswith("403")
|
|
assert count(app, *window, tenant="tenant:other")[0].startswith("403")
|
|
|
|
|
|
def test_each_rare_class_must_have_its_own_positive_claim(receiver):
|
|
app, writer = receiver
|
|
now = datetime.now(timezone.utc)
|
|
covered = RARE[0]
|
|
event = dict(envelope("audit-core.heartbeat"), subject=SOURCE,
|
|
data={"class": covered, "assertion": "nothing-to-report", "committed": 0, "released_uncommitted": 0})
|
|
assert post(app, event)[0].startswith("202")
|
|
# Source/class finding evaluator; a heartbeat for deny cannot cover redact.
|
|
findings = evaluate([writer], {(SOURCE, covered): now.isoformat()}, now=now)
|
|
assert {f.event_class for f in findings} == set(RARE) - {covered}
|
|
assert all(f.kind == NEVER_HEARTBEAT for f in findings)
|