diff --git a/audit_core/ingestion.py b/audit_core/ingestion.py index b39d02a..578adc8 100644 --- a/audit_core/ingestion.py +++ b/audit_core/ingestion.py @@ -46,7 +46,7 @@ from audit_core.redaction import ( apply_policy, finding_from_path, ) -from audit_core.senders import SenderIdentity, SenderRegistry, development_registry +from audit_core.senders import WILDCARD, SenderIdentity, SenderRegistry, development_registry from audit_core.sqlite_backend import SQLiteAuditBackend MAX_BODY_BYTES = 256 * 1024 @@ -164,10 +164,21 @@ class IngestionApplication: "/v1/secret-findings", "/v1/stats", "/v1/integrity", + "/v1/stream-findings", ) ): return self._read(start_response, environ, path, identity) + # Reconciliation is deliberately routed BEFORE the may_read gate. + # A source asking how many of its own events audit-core holds is not + # reading the archive — it learns nothing it did not itself emit — and + # §9.6 makes that comparison the source's own detection obligation. A + # writer with may_read: false must therefore be able to ask, or the + # obligation audit-core argued for is undischargeable by every sender + # actually registered. Cross-source counts stay behind may_read. + if method == "GET" and path == "/v1/reconciliation": + return self._reconciliation(start_response, environ, identity) + if path != "/v1/events" or method != "POST": return self._json(start_response, HTTPStatus.NOT_FOUND, {"error": "not_found"}) @@ -264,6 +275,8 @@ class IngestionApplication: ) if path == "/v1/stats": return self._json(start_response, HTTPStatus.OK, self.counters.snapshot()) + if path == "/v1/stream-findings": + return self._stream_findings(start_response) if path == "/v1/secret-findings": return self._json( start_response, HTTPStatus.OK, @@ -309,6 +322,97 @@ class IngestionApplication: start_response, HTTPStatus.SERVICE_UNAVAILABLE, {"error": "backend_unavailable"} ) + def _reconciliation(self, start_response, environ, identity): + """Per-class counts of one source's own events (AUDIT-WP-0009-T06). + + Counts, never payloads, and never another source's. The comparison a + source makes against its own state transitions is the §9.6 detection + obligation; audit-core supplies the numerator and says nothing about + what the answer means. + """ + query = parse_qs(environ.get("QUERY_STRING", "")) + source = (query.get("source") or [""])[0] + if not source: + return self._json( + start_response, HTTPStatus.BAD_REQUEST, {"error": "source_required"} + ) + if not identity.permits_source(source): + # 403 rather than an empty count. A zero here would read as "we + # hold none of yours", which is a materially different and false + # answer to a question about completeness. + return self._json( + start_response, HTTPStatus.FORBIDDEN, {"error": "source_not_allowed"} + ) + window = _window(query) + if window is None: + return self._json( + start_response, HTTPStatus.BAD_REQUEST, + {"error": "since_and_until_required"}, + ) + since, until = window + tenant = (query.get("tenant") or [""])[0] or None + if tenant is not None and not identity.permits_tenant(tenant): + return self._json( + start_response, HTTPStatus.FORBIDDEN, {"error": "tenant_not_allowed"} + ) + if tenant is None and not identity.has_full_tenant_scope(): + # A scoped credential must name the tenant it is counting, so the + # answer cannot silently aggregate across a boundary it may not see. + return self._json( + start_response, HTTPStatus.BAD_REQUEST, {"error": "tenant_required"} + ) + counter = getattr(self.backend, "event_counts", None) + if not callable(counter): + return self._json( + start_response, HTTPStatus.NOT_FOUND, + {"error": "reconciliation_not_supported"}, + ) + try: + counts = counter(source, since, until, tenant) + except BackendUnavailableError as exc: + log.error("reconciliation failed: %s", exc) + return self._json( + start_response, HTTPStatus.SERVICE_UNAVAILABLE, + {"error": "backend_unavailable"}, + ) + return self._json(start_response, HTTPStatus.OK, { + "source": source, + "tenant": tenant, + "since": since, + "until": until, + "counts": counts, + # Said in the response because this is the number most likely to + # be quoted out of context in someone else's conformance argument. + "means": ( + "the count of events audit-core accepted and stored in this " + "window. Divergence from the source's own count is a finding " + "for the source; agreement proves neither completeness nor " + "that any event occurred." + ), + }) + + def _stream_findings(self, start_response): + """Missing-heartbeat findings (AUDIT-WP-0009-T04, T07).""" + from audit_core.stream_findings import evaluate + + reader = getattr(self.backend, "last_heartbeats", None) + if not callable(reader): + return self._json( + start_response, HTTPStatus.NOT_FOUND, + {"error": "stream_findings_not_supported"}, + ) + last: dict[tuple[str, str], str | None] = {} + for identity in self.senders.identities: + for source in identity.sources: + if source == WILDCARD: + continue + for event_class, at in reader(source).items(): + last[(source, event_class)] = at + findings = evaluate(self.senders.identities, last) + return self._json(start_response, HTTPStatus.OK, { + "stream_findings": [finding.as_dict() for finding in findings], + }) + def _count_secrets(self, payload: Any, identity, outcome: str, findings) -> None: """Count secret-shaped fields by path, sender, source and action. @@ -472,9 +576,32 @@ _UNSCOPED_READ_PATHS = frozenset({ "/v1/stats", "/v1/secret-findings", "/v1/integrity", + # Findings span every registered sender and carry no tenant key, so there + # is nothing to filter on. A scoped reader is refused rather than served + # instance-wide facts — the same rule as the other three. + "/v1/stream-findings", }) +def _window(query: dict[str, list[str]]) -> tuple[str, str] | None: + """Require an explicit bounded window for a reconciliation query. + + No default window. A count whose bounds the caller did not choose is not + comparable against anything the caller computed, and would be quoted as + though it were. + """ + since = (query.get("since") or [""])[0] + until = (query.get("until") or [""])[0] + if not since or not until: + return None + try: + start = _normalize_timestamp(since) + end = _normalize_timestamp(until) + except ValueError: + return None + return (start, end) if start < end else None + + def _readable_by(identity, record: dict) -> bool: """Whether ``identity`` may read ``record``. diff --git a/audit_core/postgres_backend.py b/audit_core/postgres_backend.py index eb79162..03e25ea 100644 --- a/audit_core/postgres_backend.py +++ b/audit_core/postgres_backend.py @@ -24,6 +24,7 @@ import time from datetime import datetime, timezone from typing import Any +from audit_core.stream_findings import HEARTBEAT_ACTION from audit_core.interface import ( AcceptResult, AuditEvent, @@ -408,6 +409,37 @@ class PostgresAuditBackend: ) return [{"accepted_at": _iso(at), **rec} for rec, at in rows] + def event_counts( + self, source: str, since: str, until: str, tenant: str | None = None + ) -> list[dict]: + """Per-class counts for one source over a bounded window. + + AUDIT-WP-0009-T06. Counts only — never payloads. A reconciliation + answer that carried records would turn a completeness check into a read + surface, and a source does not gain the right to read the archive by + emitting into it. + """ + sql = ( + f"SELECT action, count(*) FROM {self._events} " + "WHERE source = %s AND accepted_at >= %s AND accepted_at < %s" + ) + params: tuple = (source, since, until) + if tenant is not None: + sql += " AND tenant = %s" + params += (tenant,) + sql += " GROUP BY action ORDER BY action" + return [{"class": action, "count": count} for action, count in self._query(sql, params)] + + def last_heartbeats(self, source: str) -> dict[str, str]: + """Most recent heartbeat per class for one source.""" + rows = self._query( + f"SELECT record->'details'->'data'->>'class' AS cls, max(accepted_at) " + f"FROM {self._events} WHERE source = %s AND action = %s " + "GROUP BY cls", + (source, HEARTBEAT_ACTION), + ) + return {cls: _iso(at) for cls, at in rows if cls} + def replay(self, event_id: str) -> AcceptResult: """Re-submit a stored event through :meth:`accept`. diff --git a/audit_core/senders.py b/audit_core/senders.py index 9bae94f..d856c3c 100644 --- a/audit_core/senders.py +++ b/audit_core/senders.py @@ -68,6 +68,20 @@ class SenderIdentity: # §9.6 requires the trade be declared where the trail is documented, so it # travels with the identity rather than living only in prose. completeness_trade: str | None = None + # §9.6 heartbeat declaration (AUDIT-WP-0009-T04): event class -> the + # longest gap, in seconds, that is not yet a finding. + # + # This is deliberately NOT the §17 emission-cadence schema that T05 waits + # on. Cadence describes a stream's expected *rate* and belongs to Taxonomy; + # this is a registration property saying how often a source promises to say + # "nothing to report" for a class that may legitimately be silent. The two + # are complementary and audit-core is not inventing a competing rate shape. + # + # Keyed per class, not per source, on purpose. A per-source heartbeat from + # a mixed-volume emitter is satisfied by its chattiest class and says + # nothing about the quiet, security-relevant one — which is the only case + # heartbeats exist for. + heartbeat_classes: tuple[tuple[str, int], ...] = () def __post_init__(self) -> None: if not self.name: @@ -102,6 +116,17 @@ class SenderIdentity: "completeness_trade — §9.6 requires emission atomicity of it" ) + for action, interval in self.heartbeat_classes: + if not action or not str(action).strip(): + raise ValueError( + f"sender {self.name!r}: a heartbeat class needs an event class" + ) + if not isinstance(interval, int) or isinstance(interval, bool) or interval <= 0: + raise ValueError( + f"sender {self.name!r}: heartbeat interval for {action!r} must " + f"be a positive number of seconds, got {interval!r}" + ) + @property def is_load_bearing(self) -> bool: return self.evidence_kind == EVIDENCE_LOAD_BEARING @@ -118,6 +143,7 @@ class SenderIdentity: "evidence_kind": self.evidence_kind, "completeness_claimed": False, "completeness_trade": self.completeness_trade, + "heartbeat_classes": dict(self.heartbeat_classes), "detection_surface": None, } @@ -288,6 +314,7 @@ def _apply_scope_overlay( if "completeness_trade" in extra else identity.completeness_trade ), + heartbeat_classes=_overlay_heartbeats(identity, extra), ) ) return merged @@ -313,6 +340,36 @@ def _overlay_evidence_kind(identity: SenderIdentity, extra: dict[str, Any]) -> s return declared +def _overlay_heartbeats( + identity: SenderIdentity, extra: dict[str, Any] +) -> tuple[tuple[str, int], ...]: + """The overlay may add a heartbeat class or shorten an interval, never + lengthen or remove one. + + Same asymmetry as ``evidence_kind`` and for the same reason: a ConfigMap + refresh must not be able to widen the window in which a suppressed class + goes unnoticed, or drop the obligation entirely, without anyone deciding + to. Tightening detection is safe in the direction a mistake would take it. + """ + if "heartbeat_classes" not in extra: + return identity.heartbeat_classes + declared = dict(identity.heartbeat_classes) + proposed = dict(_heartbeat_classes(extra["heartbeat_classes"])) + for action, interval in declared.items(): + if action not in proposed: + raise ValueError( + f"sender {identity.name!r}: the scope overlay may not remove " + f"heartbeat class {action!r}" + ) + if proposed[action] > interval: + raise ValueError( + f"sender {identity.name!r}: the scope overlay may not lengthen " + f"the heartbeat interval for {action!r} " + f"({interval}s -> {proposed[action]}s)" + ) + return tuple(sorted(proposed.items())) + + def _parse_identities(raw: str) -> list[SenderIdentity]: try: entries = json.loads(raw) @@ -338,9 +395,18 @@ def _identity_from(entry: Any) -> SenderIdentity: expires_at=_parse_expiry(entry.get("expires_at")), evidence_kind=str(entry.get("evidence_kind", EVIDENCE_ATTRIBUTIVE)), completeness_trade=_clean_trade(entry.get("completeness_trade")), + heartbeat_classes=_heartbeat_classes(entry.get("heartbeat_classes")), ) +def _heartbeat_classes(value: Any) -> tuple[tuple[str, int], ...]: + if not value: + return () + if not isinstance(value, dict): + raise ValueError("heartbeat_classes must be an object of class -> seconds") + return tuple(sorted((str(k), v) for k, v in value.items())) + + def _clean_trade(value: Any) -> str | None: if value in (None, ""): return None diff --git a/audit_core/sqlite_backend.py b/audit_core/sqlite_backend.py index 6811f97..cb8fb15 100644 --- a/audit_core/sqlite_backend.py +++ b/audit_core/sqlite_backend.py @@ -13,6 +13,7 @@ import sqlite3 import threading from datetime import datetime, timezone +from audit_core.stream_findings import HEARTBEAT_ACTION from audit_core.interface import ( AcceptResult, AuditEvent, @@ -218,6 +219,40 @@ class SQLiteAuditBackend: ) return [{"accepted_at": at, **json.loads(rec)} for rec, at in rows] + def event_counts( + self, source: str, since: str, until: str, tenant: str | None = None + ) -> list[dict]: + """Per-class counts for one source over a bounded window. + + AUDIT-WP-0009-T06. Counts only — never payloads. A reconciliation + answer that carried records would turn a completeness check into a read + surface, and a source does not gain the right to read the archive by + emitting into it. + """ + sql = ( + "SELECT json_extract(record, '$.action') AS action, count(*) " + "FROM events WHERE json_extract(record, '$.source') = ? " + "AND accepted_at >= ? AND accepted_at < ?" + ) + params: tuple = (source, since, until) + if tenant is not None: + sql += " AND tenant = ?" + params += (tenant,) + sql += " GROUP BY action ORDER BY action" + return [{"class": action, "count": count} for action, count in self._query(sql, params)] + + def last_heartbeats(self, source: str) -> dict[str, str]: + """Most recent heartbeat per class for one source.""" + rows = self._query( + "SELECT json_extract(record, '$.details.data.class') AS cls, " + "max(accepted_at) FROM events " + "WHERE json_extract(record, '$.source') = ? " + "AND json_extract(record, '$.action') = ? " + "GROUP BY cls", + (source, HEARTBEAT_ACTION), + ) + return {cls: at for cls, at in rows if cls} + def record_rejection( self, *, diff --git a/audit_core/stream_findings.py b/audit_core/stream_findings.py new file mode 100644 index 0000000..1e16fc0 --- /dev/null +++ b/audit_core/stream_findings.py @@ -0,0 +1,131 @@ +"""Stream-completeness findings: when the trail itself is the defect. + +`AUDIT-WP-0009-T04` and `T07`. `INTENT.md` principle 10 already says a degraded +audit stream is itself an audit and operations event. The principle was in +place; the mechanism was not. + +**What this can and cannot see.** The hash chain proves records held were not +altered or truncated. It says nothing about a record that never arrived, and +§9.6 is explicit that omission is the acute risk for exactly the rare negative +classes — revocation, denial, containment — where suppression is most valuable +and least visible. Rate monitoring cannot help there either: a class that is +legitimately silent for a month is indistinguishable from one being suppressed. + +A heartbeat fixes that by inverting the burden. Instead of inferring health +from events that may never come, the source makes a positive claim — +*nothing to report for this class* — on a declared interval. The claim can +itself go missing, and a missing claim is a finding. That is the whole idea, +and it is the only control here that covers adversarial omission. + +**The bound, stated because a finding surface invites over-reading.** A +compromised source emits a truthful-looking heartbeat while suppressing the +event it is supposed to be vouching for. Heartbeats cover loss, outage, drain +failure and accident — most of what actually goes wrong — and do not cover a +source lying about itself. Nothing an archive holds can close that; it needs an +observer independent of the emitter. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from typing import Any, Iterable + +# The event class a source uses to say "nothing to report". A heartbeat is an +# ordinary event: same envelope, same append-only custody, same chain. It gets +# no special table, because a heartbeat that lived outside the chain would be +# the one record in this store that could be back-dated. +HEARTBEAT_ACTION = "audit-core.heartbeat" + +MISSING_HEARTBEAT = "missing_heartbeat" +NEVER_HEARTBEAT = "no_heartbeat_since_registration" + + +@dataclass(frozen=True) +class StreamFinding: + """One reason to believe a stream is not telling the whole truth.""" + + kind: str + sender: str + source: str + event_class: str + expected_within_seconds: int + last_seen: str | None + overdue_seconds: int + + def as_dict(self) -> dict[str, Any]: + return { + "kind": self.kind, + "sender": self.sender, + "source": self.source, + "class": self.event_class, + "expected_within_seconds": self.expected_within_seconds, + "last_seen": self.last_seen, + "overdue_seconds": self.overdue_seconds, + # Said on every finding rather than in a document nobody opens + # alongside it: this is the absence of a positive claim, not proof + # that an event was suppressed. + "means": ( + "a declared heartbeat did not arrive; the class may be healthy " + "and the emitter silent. Absence is not proof of suppression." + ), + } + + +def _parse(value: str | None) -> datetime | None: + if not value: + return None + try: + parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00")) + except ValueError: + return None + return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc) + + +def evaluate( + identities: Iterable[Any], + last_heartbeats: dict[tuple[str, str], str | None], + *, + now: datetime | None = None, + grace_factor: float = 1.5, +) -> list[StreamFinding]: + """Findings for every declared heartbeat class that is overdue. + + ``last_heartbeats`` is keyed ``(source, class)``. A class with no heartbeat + ever recorded produces a finding of its own kind rather than being skipped: + a source that declared a heartbeat and never sent one is the case most + likely to be a broken integration, and it is precisely the one a + "compare against last seen" implementation silently drops. + + ``grace_factor`` exists so one late run does not flap a finding on and off, + matching the reasoning behind the attestation freshness window. It widens + the window; it never removes the finding. + """ + now = now or datetime.now(timezone.utc) + findings: list[StreamFinding] = [] + for identity in identities: + for event_class, interval in getattr(identity, "heartbeat_classes", ()): + for source in sorted(identity.sources): + if source == "*": + # A wildcard source cannot be held to a heartbeat: there is + # no determinate set of streams to expect one from. Refused + # rather than guessed at. + continue + deadline = timedelta(seconds=interval * grace_factor) + last = _parse(last_heartbeats.get((source, event_class))) + if last is None: + findings.append(StreamFinding( + kind=NEVER_HEARTBEAT, sender=identity.name, source=source, + event_class=event_class, expected_within_seconds=interval, + last_seen=None, overdue_seconds=-1, + )) + continue + overdue = (now - last) - deadline + if overdue.total_seconds() > 0: + findings.append(StreamFinding( + kind=MISSING_HEARTBEAT, sender=identity.name, source=source, + event_class=event_class, expected_within_seconds=interval, + last_seen=last.isoformat(), + overdue_seconds=int(overdue.total_seconds()), + )) + return findings diff --git a/docs/stream-completeness.md b/docs/stream-completeness.md new file mode 100644 index 0000000..e80680a --- /dev/null +++ b/docs/stream-completeness.md @@ -0,0 +1,116 @@ +# Stream completeness: heartbeats, reconciliation, and findings + +`AUDIT-WP-0009` T04, T06, T07. Statute §9.6. + +The chain proves records held were not altered or truncated. It says nothing +about a record that never arrived — and §9.6 is explicit that **omission** is +the acute risk for exactly the rare negative classes (revocation, denial, +containment) where suppression is most valuable and least visible. + +audit-core argued that obligation up to a MUST. A source cannot declare a +cadence to a system with nowhere to put it, so this is the surface owed. + +## Three surfaces, and what each actually covers + +| Surface | Covers | Does not cover | +| --- | --- | --- | +| Heartbeat (`POST /v1/events`, class `audit-core.heartbeat`) | A class going silent — including one that is *legitimately* silent, which rate monitoring can never distinguish | A compromised source emitting a truthful-looking heartbeat while suppressing the event | +| Reconciliation (`GET /v1/reconciliation`) | Loss, outage, drain failure — divergence between what a source emitted and what arrived | A compromised source suppressing the event and its own count together | +| Findings (`GET /v1/stream-findings`) | Surfacing the above where an operator sees them | Anything the two above do not detect | + +**The bound is the same in both rows and it is not a footnote.** Where the +emitter itself is compromised, both controls agree with it. They cover loss, +outage, drain failure and accident — most of what actually goes wrong — and not +a source lying about itself. Closing that needs an observer independent of the +emitter, which §16 placed outside audit-core's scope. No conformance claim may +read these surfaces as covering adversarial omission by the source. + +## Heartbeats + +A heartbeat is an **ordinary event**: same envelope, same append-only custody, +same chain, no special table. That is deliberate — a heartbeat stored outside +the chain would be the one record in this store that could be back-dated. + +```json +{ + "id": "...", "type": "audit-core.heartbeat", + "source": "approval-engine", "subject": "approval-engine", + "tenant": "tenant:platform", "correlation_id": "...", + "occurred_at": "2026-09-10T03:00:00+00:00", + "data": {"class": "approval.revocation", "assertion": "nothing-to-report"} +} +``` + +### Declared per class, never per source + +`heartbeat_classes` on the sender registration maps event class → the longest +gap in seconds that is not yet a finding. + +Per class is the whole point. A per-source heartbeat from a mixed-volume +emitter is satisfied by its chattiest class and says nothing about the quiet, +security-relevant one — and the quiet class is the only reason heartbeats +exist. `informed-decision` raised this shape first for presentations versus +dispositions; it generalises. + +### Not the §17 cadence schema + +`AUDIT-WP-0009-T05` waits on the emission-cadence declaration `kings-guard` is +drafting for Taxonomy, and this does not pre-empt it. Cadence describes a +stream's expected **rate**; a heartbeat is a registration property saying how +often a source promises to say *nothing to report* for a class that may +legitimately be silent. Complementary, not alternatives, and audit-core is not +inventing a competing rate shape while the real one is being written. + +### Findings + +`GET /v1/stream-findings` requires `may_read` **and** full tenant scope: +findings span every registered sender and carry no tenant key, so there is +nothing to filter on and a scoped reader is refused rather than served +instance-wide facts. Same rule as dead letters, secret findings and integrity. + +Two kinds: + +- `no_heartbeat_since_registration` — declared a heartbeat, never sent one. + Its own kind rather than skipped, because it is the case most likely to be a + broken integration and the one a naive "compare against last seen" + implementation silently drops. +- `missing_heartbeat` — one arrived once and is now overdue, with how late. + +A grace factor of 1.5 widens the window so a single late run does not flap a +finding on and off — the same reasoning as the attestation freshness window. It +widens; it never removes. + +Every finding carries a `means` field saying that absence of a heartbeat is not +proof of suppression. That travels on the finding rather than in a document +nobody opens alongside it. + +## Reconciliation + +`GET /v1/reconciliation?source=&tenant=&since=&until=` returns per-class counts +of a source's own events. **Counts, never payloads.** + +### Why a writer with `may_read: false` may call it + +Every registered sender holds `may_read: false` — a source does not gain a read +surface by emitting. Taken literally that would make the §9.6 reconciliation +obligation undischargeable by every source actually registered, which is a +strange place for a rule audit-core argued for to end up. + +The resolution is that **a source asking how many of its own events audit-core +holds is not reading the archive**. It learns nothing it did not itself emit. +So the surface is scoped to the caller's own permitted sources and tenants, and +returns no payloads. Anything wider — another source's counts, or an unscoped +aggregate — stays behind `may_read` and full tenant scope. + +Two refusals worth knowing: + +- Another source's counts return **403, not an empty count**. A zero would read + as "we hold none of yours", which is a materially different and false answer + to a question about completeness. +- There is **no default window**. A count whose bounds the caller did not choose + is not comparable against anything the caller computed, and would be quoted as + though it were. + +The response carries a `means` field stating that agreement proves neither +completeness nor that any event occurred — because this is the number most +likely to be quoted out of context in someone else's conformance argument. diff --git a/tests/test_stream_findings.py b/tests/test_stream_findings.py new file mode 100644 index 0000000..f3b83ca --- /dev/null +++ b/tests/test_stream_findings.py @@ -0,0 +1,272 @@ +"""AUDIT-WP-0009 T04/T06/T07 — heartbeats, reconciliation, findings.""" + +import io +import json +from datetime import datetime, timedelta, timezone + +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 ( + HEARTBEAT_ACTION, + MISSING_HEARTBEAT, + NEVER_HEARTBEAT, + evaluate, +) + + +def invoke(app, path, *, token="approval", method="GET", payload=None, key=None): + raw = json.dumps(payload).encode() if payload is not None else b"" + environ = { + "PATH_INFO": path.split("?")[0], + "QUERY_STRING": path.split("?")[1] if "?" in path else "", + "REQUEST_METHOD": method, + "CONTENT_LENGTH": str(len(raw)), + "wsgi.input": io.BytesIO(raw), + "HTTP_AUTHORIZATION": f"Bearer {token}", + } + if key: + environ["HTTP_IDEMPOTENCY_KEY"] = key + result = {} + out = b"".join(app(environ, lambda status, headers: result.update(status=status))) + return result["status"], (json.loads(out) if out else {}) + + +SENDER = SenderIdentity( + name="approval-engine", + tokens=("approval",), + sources=frozenset({"approval-engine"}), + tenants=frozenset({"tenant:platform"}), + may_write=True, + may_read=False, + evidence_kind="load-bearing", + heartbeat_classes=(("approval.revocation", 86400),), +) +OPERATOR = SenderIdentity( + name="operator", tokens=("operator",), sources=frozenset({"approval-engine"}), + tenants=frozenset({"*"}), may_write=False, may_read=True, +) + + +@pytest.fixture +def app(tmp_path): + backend = SQLiteAuditBackend(str(tmp_path / "s.db")) + return IngestionApplication(backend, SenderRegistry([SENDER, OPERATOR])) + + +def emit(app, event_id, action, *, data=None, tenant="tenant:platform"): + payload = { + "id": event_id, "type": action, "source": "approval-engine", + "subject": "approval:1", "tenant": tenant, "correlation_id": "c-1", + "occurred_at": "2026-09-10T00:00:00+00:00", "data": data or {"k": "v"}, + } + return invoke(app, "/v1/events", method="POST", payload=payload, key=event_id) + + +# --- T04: heartbeats ------------------------------------------------------- + + +def test_a_heartbeat_is_an_ordinary_event_and_is_chained(app): + """No special table. A heartbeat outside the chain could be back-dated.""" + status, _ = emit( + app, "hb-1", HEARTBEAT_ACTION, + data={"class": "approval.revocation", "assertion": "nothing-to-report"}, + ) + assert status.startswith("202") + status, body = invoke(app, "/v1/integrity", token="operator") + assert status.startswith("200") + assert body["intact"] is True + assert body["events"] == 1 + + +def test_a_declared_class_with_no_heartbeat_ever_is_its_own_finding(): + """The case a naive 'compare against last seen' silently drops.""" + findings = evaluate([SENDER], {}) + assert [f.kind for f in findings] == [NEVER_HEARTBEAT] + assert findings[0].event_class == "approval.revocation" + assert findings[0].source == "approval-engine" + + +def test_a_fresh_heartbeat_produces_no_finding(): + now = datetime(2026, 9, 10, 12, tzinfo=timezone.utc) + last = (now - timedelta(hours=2)).isoformat() + assert evaluate([SENDER], {("approval-engine", "approval.revocation"): last}, now=now) == [] + + +def test_an_overdue_heartbeat_is_a_finding_with_its_lateness(): + now = datetime(2026, 9, 10, 12, tzinfo=timezone.utc) + # 86400s interval, 1.5 grace -> overdue past 36h. + last = (now - timedelta(hours=40)).isoformat() + findings = evaluate([SENDER], {("approval-engine", "approval.revocation"): last}, now=now) + assert [f.kind for f in findings] == [MISSING_HEARTBEAT] + assert findings[0].overdue_seconds == int(timedelta(hours=4).total_seconds()) + + +def test_grace_widens_the_window_but_never_removes_the_finding(): + now = datetime(2026, 9, 10, 12, tzinfo=timezone.utc) + last = (now - timedelta(days=30)).isoformat() + findings = evaluate( + [SENDER], {("approval-engine", "approval.revocation"): last}, + now=now, grace_factor=100.0, + ) + # 86400 * 100 is ~100 days, so this one is inside the window... + assert findings == [] + # ...and no grace factor makes a year-old heartbeat acceptable. + older = (now - timedelta(days=400)).isoformat() + assert evaluate( + [SENDER], {("approval-engine", "approval.revocation"): older}, + now=now, grace_factor=100.0, + ) + + +def test_a_wildcard_source_is_not_held_to_a_heartbeat(): + """No determinate set of streams to expect one from. Refused, not guessed.""" + wild = SenderIdentity( + name="w", tokens=("t",), sources=frozenset({"*"}), + heartbeat_classes=(("anything", 60),), + ) + assert evaluate([wild], {}) == [] + + +def test_every_finding_says_absence_is_not_proof_of_suppression(): + """The surface invites over-reading; the bound travels on each finding.""" + finding = evaluate([SENDER], {})[0].as_dict() + assert "not proof of suppression" in finding["means"].lower() + + +# --- T07: the findings surface --------------------------------------------- + + +def test_stream_findings_needs_read_and_full_tenant_scope(app): + assert invoke(app, "/v1/stream-findings")[0].startswith("403") + status, body = invoke(app, "/v1/stream-findings", token="operator") + assert status.startswith("200") + assert body["stream_findings"][0]["kind"] == NEVER_HEARTBEAT + + +def test_a_delivered_heartbeat_clears_the_finding(app): + emit( + app, "hb-2", HEARTBEAT_ACTION, + data={"class": "approval.revocation", "assertion": "nothing-to-report"}, + ) + _, body = invoke(app, "/v1/stream-findings", token="operator") + # Stored today against a 24h interval: no longer a finding. + assert body["stream_findings"] == [] + + +# --- T06: reconciliation --------------------------------------------------- + + +def test_a_writer_may_count_its_own_events_without_may_read(app): + """The §9.6 obligation would be undischargeable otherwise.""" + assert SENDER.may_read is False + emit(app, "e-1", "approval.issuance") + emit(app, "e-2", "approval.issuance") + emit(app, "e-3", "approval.revocation") + status, body = invoke( + app, + "/v1/reconciliation?source=approval-engine&tenant=tenant:platform" + "&since=2026-09-01T00:00:00%2B00:00&until=2027-01-01T00:00:00%2B00:00", + ) + assert status.startswith("200") + assert body["counts"] == [ + {"class": "approval.issuance", "count": 2}, + {"class": "approval.revocation", "count": 1}, + ] + + +def test_reconciliation_returns_counts_and_never_payloads(app): + emit(app, "e-9", "approval.issuance", data={"secret_shaped": "x", "binding": "b"}) + _, body = invoke( + app, + "/v1/reconciliation?source=approval-engine&tenant=tenant:platform" + "&since=2026-09-01T00:00:00%2B00:00&until=2027-01-01T00:00:00%2B00:00", + ) + assert set(body) == {"source", "tenant", "since", "until", "counts", "means"} + assert "binding" not in json.dumps(body) + + +def test_a_source_may_not_count_another_source(app): + status, body = invoke( + app, + "/v1/reconciliation?source=user-engine&tenant=tenant:platform" + "&since=2026-09-01T00:00:00%2B00:00&until=2027-01-01T00:00:00%2B00:00", + ) + # 403, not an empty count: a zero would read as "we hold none of yours", + # which is a false answer to a question about completeness. + assert status.startswith("403") + assert body["error"] == "source_not_allowed" + + +def test_reconciliation_requires_an_explicit_bounded_window(app): + status, body = invoke(app, "/v1/reconciliation?source=approval-engine&tenant=tenant:platform") + assert status.startswith("400") + assert body["error"] == "since_and_until_required" + + +def test_a_scoped_credential_must_name_its_tenant(app): + status, body = invoke( + app, + "/v1/reconciliation?source=approval-engine" + "&since=2026-09-01T00:00:00%2B00:00&until=2027-01-01T00:00:00%2B00:00", + ) + assert status.startswith("400") + assert body["error"] == "tenant_required" + + +def test_a_scoped_credential_may_not_count_another_tenant(app): + status, body = invoke( + app, + "/v1/reconciliation?source=approval-engine&tenant=tenant:coulomb" + "&since=2026-09-01T00:00:00%2B00:00&until=2027-01-01T00:00:00%2B00:00", + ) + assert status.startswith("403") + assert body["error"] == "tenant_not_allowed" + + +def test_the_count_response_states_what_it_does_not_prove(app): + _, body = invoke( + app, + "/v1/reconciliation?source=approval-engine&tenant=tenant:platform" + "&since=2026-09-01T00:00:00%2B00:00&until=2027-01-01T00:00:00%2B00:00", + ) + means = body["means"].lower() + assert "proves neither completeness" in means + + +# --- the overlay asymmetry ------------------------------------------------- + + +def test_the_overlay_may_shorten_a_heartbeat_interval_but_not_lengthen_it(tmp_path): + scope = tmp_path / "scope.json" + base = [{ + "name": "approval-engine", "tokens": ["t"], "sources": ["approval-engine"], + "heartbeat_classes": {"approval.revocation": 3600}, + }] + + scope.write_text(json.dumps( + [{"name": "approval-engine", "heartbeat_classes": {"approval.revocation": 600}}] + )) + registry = SenderRegistry.from_env({ + "AUDIT_CORE_SENDERS": json.dumps(base), + "AUDIT_CORE_SENDERS_SCOPE_PATH": str(scope), + }) + assert dict(registry.identities[0].heartbeat_classes)["approval.revocation"] == 600 + + scope.write_text(json.dumps( + [{"name": "approval-engine", "heartbeat_classes": {"approval.revocation": 99999}}] + )) + with pytest.raises(ValueError, match="lengthen"): + SenderRegistry.from_env({ + "AUDIT_CORE_SENDERS": json.dumps(base), + "AUDIT_CORE_SENDERS_SCOPE_PATH": str(scope), + }) + + scope.write_text(json.dumps([{"name": "approval-engine", "heartbeat_classes": {}}])) + with pytest.raises(ValueError, match="remove"): + SenderRegistry.from_env({ + "AUDIT_CORE_SENDERS": json.dumps(base), + "AUDIT_CORE_SENDERS_SCOPE_PATH": str(scope), + })