diff --git a/SCOPE.md b/SCOPE.md index 6ea4982..38ee8b7 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -103,10 +103,12 @@ effect of its own. freshness window. Until `AUDIT-WP-0009-T02` schedules attestation, the honest answer in production is `false` — the overclaim is gone, the precondition is not yet met. -- No cadence, heartbeat, or reconciliation surface exists. The §9.6 detection - obligations Audit Core argued for are not yet supportable by Audit Core, so - a suppressed load-bearing event still produces silence - (`AUDIT-WP-0009-T04`/`T05`/`T06`). +- ~~No cadence, heartbeat, or reconciliation surface exists.~~ **Closed.** + Heartbeats and missing-heartbeat findings (`AUDIT-WP-0009-T04`) and the + reconciliation surface (`T06`) landed 2026-09-10; declared emission cadence + against info-tech-canon wire schema 0.1 (`T05`) landed 2026-09-22 — + `docs/stream-completeness.md`. None of them covers a compromised source + suppressing an event and its own count together. - Load-bearing classification **does** now exist: `evidence_kind` on `SenderIdentity` and the registration schema, defaulting to `attributive` (`AUDIT-WP-0009-T03`, closed 2026-09-06). diff --git a/audit_core/emission_cadence.py b/audit_core/emission_cadence.py new file mode 100644 index 0000000..d469417 --- /dev/null +++ b/audit_core/emission_cadence.py @@ -0,0 +1,290 @@ +"""Declared emission cadence, evaluated by the observer (AUDIT-WP-0009-T05). + +The contract is info-tech-canon's, not ours: the EmissionCadenceDeclaration +wire schema 0.1 (``infospace/schemas/emission-cadence.schema.yaml``), read at +contract digest ``b08b4d95fc4b0bd3``. The source owns and publishes the +declaration; audit-core accepts it on the sender registration and evaluates it +against the stream it actually holds. The standard keeps those roles apart and +so does this module: a declaration is a statement of intended cadence, never +evidence that the emission happens. + +**What is evaluated here.** Only the ``expected-rate`` form: at least +``expected_min`` events of a class in each trailing window, below which a +finding is raised. That is the high-volume case, and the one question an +archive can answer on its own — how many it holds. + +**What is accepted but evaluated elsewhere.** ``heartbeat-or-reconciliation`` +entries are validated and kept, not re-evaluated. Their heartbeat half is the +registration's ``heartbeat_classes`` (T04) and their reconciliation half is +``GET /v1/reconciliation`` (T06), whose count is the ``compare_observed`` side +of the comparison. The ``compare_local`` side belongs to the source; audit-core +cannot see it and does not pretend to. + +**The bound.** A rate held says the stream was not silent, not that it was +complete. A rate missed says the source may have stopped, not that it did. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from typing import Any, Callable, Iterable + +WIRE_SCHEMA_VERSION = "0.1" +CONTRACT_DIGEST = "b08b4d95fc4b0bd3" + +EXPECTED_RATE = "expected-rate" +HEARTBEAT_OR_RECONCILIATION = "heartbeat-or-reconciliation" +FORMS = (EXPECTED_RATE, HEARTBEAT_OR_RECONCILIATION) + +BELOW_CADENCE = "below_declared_cadence" + +_ISO = re.compile( + r"^P(?=\d|T\d)(?:(\d+)D)?(?:T(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?)?$" +) +_SHORT = re.compile(r"^([1-9][0-9]*)(s|m|h|d)$") +_UNIT = {"s": 1, "m": 60, "h": 3600, "d": 86400} + +_TOP_KEYS = {"schema_version", "declaration_id", "source", "stream_id", "sources", "extensions"} +_ENTRY_KEYS = { + "source_id", "source_system", "event_class", "form", "window", "window_seconds", + "expected_min", "drop_below", "heartbeat", "reconciliation", "extensions", +} +_HEARTBEAT_KEYS = {"event_class", "interval", "interval_seconds", "assertion", "missing", "extensions"} +_RECONCILIATION_KEYS = {"compare_local", "compare_observed", "divergence", "undrained_local", "extensions"} + + +def duration_seconds(value: Any) -> int: + """Seconds in a schema ``duration``: ISO-8601 ``PnDTnHnMnS`` or ``90s``/``5m``/``1h``/``1d``.""" + text = str(value) + short = _SHORT.match(text) + if short: + return int(short.group(1)) * _UNIT[short.group(2)] + iso = _ISO.match(text) + if iso and text not in ("P", "PT"): + days, hours, minutes, seconds = (int(g or 0) for g in iso.groups()) + total = days * 86400 + hours * 3600 + minutes * 60 + seconds + if total > 0: + return total + raise ValueError(f"not a positive schema duration: {value!r}") + + +@dataclass(frozen=True) +class RateEntry: + """One ``expected-rate`` entry: at least ``expected_min`` per ``window_seconds``.""" + + source_id: str + event_class: str + window_seconds: int + expected_min: int + drop_below: str + + +@dataclass(frozen=True) +class CadenceDeclaration: + """A source-owned declaration, validated against wire schema 0.1.""" + + declaration_id: str + source: str + stream_id: str | None + rates: tuple[RateEntry, ...] + # heartbeat-or-reconciliation entries, kept verbatim: accepted, and + # evaluated by T04/T06 rather than here. + other_entries: tuple[dict, ...] + document: dict + + def summary(self) -> dict[str, Any]: + return { + "declaration_id": self.declaration_id, + "source": self.source, + "stream_id": self.stream_id, + "contract_digest": CONTRACT_DIGEST, + "expected_rate_classes": { + r.event_class: {"window_seconds": r.window_seconds, "expected_min": r.expected_min} + for r in self.rates + }, + "heartbeat_or_reconciliation_classes": sorted( + e["event_class"] for e in self.other_entries + ), + } + + +def _text(obj: dict, key: str, where: str, *, required: bool = True) -> str | None: + if key not in obj: + if required: + raise ValueError(f"{where}: {key} is required") + return None + value = obj[key] + if not isinstance(value, str) or not value: + raise ValueError(f"{where}: {key} must be a non-empty string") + return value + + +def _closed(obj: Any, allowed: set[str], where: str) -> dict: + if not isinstance(obj, dict): + raise ValueError(f"{where} must be an object") + extra = set(obj) - allowed + if extra: + raise ValueError(f"{where}: unknown properties {sorted(extra)}") + return obj + + +def _one_of(obj: dict, a: str, b: str, where: str) -> int: + if (a in obj) == (b in obj): + raise ValueError(f"{where}: exactly one of {a} or {b} is required") + if a in obj: + return duration_seconds(obj[a]) + value = obj[b] + if not isinstance(value, int) or isinstance(value, bool) or value < 1: + raise ValueError(f"{where}: {b} must be a positive integer") + return value + + +def parse_declaration(document: Any) -> CadenceDeclaration: + """Validate a declaration against wire schema 0.1 and index it. + + Hand-checked rather than via a JSON Schema library, to keep the receiver + stdlib-only. Refuses what the schema refuses; it does not relax a rule to + admit a declaration, since a relaxed reading would be a competing shape. + """ + doc = _closed(document, _TOP_KEYS, "emission cadence declaration") + if doc.get("schema_version") != WIRE_SCHEMA_VERSION: + raise ValueError( + f"emission cadence declaration: schema_version must be " + f"{WIRE_SCHEMA_VERSION!r}, got {doc.get('schema_version')!r}" + ) + declaration_id = _text(doc, "declaration_id", "declaration") + source = _text(doc, "source", "declaration") + stream_id = _text(doc, "stream_id", "declaration", required=False) + entries = doc.get("sources") + if not isinstance(entries, list) or not entries: + raise ValueError("declaration: sources must be a non-empty list") + + rates: list[RateEntry] = [] + others: list[dict] = [] + for index, raw in enumerate(entries): + where = f"declaration sources[{index}]" + entry = _closed(raw, _ENTRY_KEYS, where) + source_id = _text(entry, "source_id", where) + event_class = _text(entry, "event_class", where) + _text(entry, "source_system", where, required=False) + form = entry.get("form") + if form not in FORMS: + raise ValueError(f"{where}: form must be one of {FORMS}, got {form!r}") + if form == EXPECTED_RATE: + if "heartbeat" in entry or "reconciliation" in entry: + raise ValueError(f"{where}: expected-rate carries no heartbeat or reconciliation") + window = _one_of(entry, "window", "window_seconds", where) + expected_min = entry.get("expected_min") + if not isinstance(expected_min, int) or isinstance(expected_min, bool) or expected_min < 0: + raise ValueError(f"{where}: expected_min must be a non-negative integer") + drop_below = _text(entry, "drop_below", where) + rates.append(RateEntry(source_id, event_class, window, expected_min, drop_below)) + continue + forbidden = {"window", "window_seconds", "expected_min", "drop_below"} & set(entry) + if forbidden: + raise ValueError(f"{where}: heartbeat-or-reconciliation may not carry {sorted(forbidden)}") + if "heartbeat" not in entry and "reconciliation" not in entry: + raise ValueError(f"{where}: heartbeat-or-reconciliation needs a heartbeat or a reconciliation") + if "heartbeat" in entry: + beat = _closed(entry["heartbeat"], _HEARTBEAT_KEYS, f"{where}.heartbeat") + for key in ("event_class", "assertion", "missing"): + _text(beat, key, f"{where}.heartbeat") + _one_of(beat, "interval", "interval_seconds", f"{where}.heartbeat") + if "reconciliation" in entry: + rec = _closed(entry["reconciliation"], _RECONCILIATION_KEYS, f"{where}.reconciliation") + for key in ("compare_local", "compare_observed", "divergence"): + _text(rec, key, f"{where}.reconciliation") + _text(rec, "undrained_local", f"{where}.reconciliation", required=False) + others.append(dict(entry)) + + classes = [r.event_class for r in rates] + if len(classes) != len(set(classes)): + # Two rates for one class would each be satisfied by the other's + # events; which one is meant is not ours to guess. + raise ValueError("declaration: an event class has more than one expected-rate entry") + return CadenceDeclaration( + declaration_id=declaration_id, source=source, stream_id=stream_id, + rates=tuple(rates), other_entries=tuple(others), document=dict(doc), + ) + + +@dataclass(frozen=True) +class CadenceFinding: + """A stream observed below the rate its source declared.""" + + sender: str + source: str + event_class: str + declaration_id: str + window_seconds: int + expected_min: int + observed: int + since: str + until: str + + def as_dict(self) -> dict[str, Any]: + return { + "kind": BELOW_CADENCE, + "sender": self.sender, + "source": self.source, + "class": self.event_class, + "declaration_id": self.declaration_id, + "contract_digest": CONTRACT_DIGEST, + "window_seconds": self.window_seconds, + "expected_min": self.expected_min, + "observed": self.observed, + "since": self.since, + "until": self.until, + "means": ( + "audit-core holds fewer events of this class in the window than " + "the source declared it would emit. The source may have stopped, " + "or the declaration may be wrong; this is not proof of " + "suppression, and a rate held would not prove completeness." + ), + } + + +def evaluate( + identities: Iterable[Any], + counter: Callable[[str, str, str], list[dict]], + *, + now: datetime | None = None, +) -> list[CadenceFinding]: + """Findings for every declared expected-rate class observed below its rate. + + ``counter(source, since, until)`` is the backend's per-class count — the + same numbers ``/v1/reconciliation`` returns, so a finding and a count + cannot disagree. Each declared class is evaluated over the trailing window + ending ``now``, per permitted source. A wildcard source is refused, as for + heartbeats: there is no determinate stream to hold to a rate. + """ + now = (now or datetime.now(timezone.utc)).astimezone(timezone.utc) + until = now.isoformat() + findings: list[CadenceFinding] = [] + for identity in identities: + declaration = getattr(identity, "emission_cadence", None) + if declaration is None or not declaration.rates: + continue + for source in sorted(identity.sources): + if source == "*": + continue + cache: dict[int, dict[str, int]] = {} + for rate in declaration.rates: + if rate.window_seconds not in cache: + since = (now - timedelta(seconds=rate.window_seconds)).isoformat() + cache[rate.window_seconds] = { + row["class"]: int(row["count"]) for row in counter(source, since, until) + } + observed = cache[rate.window_seconds].get(rate.event_class, 0) + if observed < rate.expected_min: + findings.append(CadenceFinding( + sender=identity.name, source=source, event_class=rate.event_class, + declaration_id=declaration.declaration_id, + window_seconds=rate.window_seconds, expected_min=rate.expected_min, + observed=observed, + since=(now - timedelta(seconds=rate.window_seconds)).isoformat(), + until=until, + )) + return findings diff --git a/audit_core/ingestion.py b/audit_core/ingestion.py index 809f5cc..b796eb9 100644 --- a/audit_core/ingestion.py +++ b/audit_core/ingestion.py @@ -401,7 +401,8 @@ class IngestionApplication: }) def _stream_findings(self, start_response): - """Missing-heartbeat findings (AUDIT-WP-0009-T04, T07).""" + """Missing-heartbeat and below-cadence findings (AUDIT-WP-0009-T04, T05, T07).""" + from audit_core.emission_cadence import evaluate as evaluate_cadence from audit_core.stream_findings import evaluate reader = getattr(self.backend, "last_heartbeats", None) @@ -417,9 +418,20 @@ class IngestionApplication: continue for event_class, at in reader(source).items(): last[(source, event_class)] = at - findings = evaluate(self.senders.identities, last) + findings = [f.as_dict() for f in evaluate(self.senders.identities, last)] + counter = getattr(self.backend, "event_counts", None) + if callable(counter): + # AUDIT-WP-0009-T05: a declared expected rate, read from the same + # counts /v1/reconciliation returns so the two cannot disagree. + findings += [ + f.as_dict() + for f in evaluate_cadence( + self.senders.identities, + lambda source, since, until: counter(source, since, until, None), + ) + ] return self._json(start_response, HTTPStatus.OK, { - "stream_findings": [finding.as_dict() for finding in findings], + "stream_findings": findings, }) def _count_secrets(self, payload: Any, identity, outcome: str, findings) -> None: diff --git a/audit_core/senders.py b/audit_core/senders.py index d856c3c..a400bbb 100644 --- a/audit_core/senders.py +++ b/audit_core/senders.py @@ -22,6 +22,7 @@ from pathlib import Path from typing import Any, Iterable from audit_core.redaction import POLICIES, POLICY_REDACT +from audit_core.emission_cadence import parse_declaration WILDCARD = "*" @@ -71,9 +72,9 @@ class SenderIdentity: # §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 + # This is deliberately NOT the emission-cadence declaration (T05, below). + # Cadence describes a stream's expected *rate* and its shape is + # info-tech-canon's; 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. # @@ -82,6 +83,10 @@ class SenderIdentity: # nothing about the quiet, security-relevant one — which is the only case # heartbeats exist for. heartbeat_classes: tuple[tuple[str, int], ...] = () + # The source-owned emission cadence declaration (AUDIT-WP-0009-T05), + # parsed from info-tech-canon wire schema 0.1 — the §17 shape the heartbeat + # field above deliberately did not invent. ``None`` means undeclared. + emission_cadence: Any = None def __post_init__(self) -> None: if not self.name: @@ -144,6 +149,9 @@ class SenderIdentity: "completeness_claimed": False, "completeness_trade": self.completeness_trade, "heartbeat_classes": dict(self.heartbeat_classes), + "emission_cadence": ( + self.emission_cadence.summary() if self.emission_cadence else None + ), "detection_surface": None, } @@ -315,6 +323,7 @@ def _apply_scope_overlay( else identity.completeness_trade ), heartbeat_classes=_overlay_heartbeats(identity, extra), + emission_cadence=_overlay_cadence(identity, extra), ) ) return merged @@ -370,6 +379,22 @@ def _overlay_heartbeats( return tuple(sorted(proposed.items())) +def _overlay_cadence(identity: SenderIdentity, extra: dict[str, Any]) -> Any: + """The overlay may not touch an emission cadence declaration. + + A declaration is the source's, pinned to a contract digest; a partial + overlay has no sound merge with it, and replacing it wholesale could widen + a window or drop a class without anyone deciding to. Changing it is a + change to the Secret-backed registration. + """ + if "emission_cadence" in extra: + raise ValueError( + f"sender {identity.name!r}: the scope overlay may not set or change " + "emission_cadence" + ) + return identity.emission_cadence + + def _parse_identities(raw: str) -> list[SenderIdentity]: try: entries = json.loads(raw) @@ -396,6 +421,11 @@ def _identity_from(entry: Any) -> SenderIdentity: 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")), + emission_cadence=( + parse_declaration(entry["emission_cadence"]) + if entry.get("emission_cadence") is not None + else None + ), ) diff --git a/docs/approval-engine-source-registration.md b/docs/approval-engine-source-registration.md index fa9b816..5120417 100644 --- a/docs/approval-engine-source-registration.md +++ b/docs/approval-engine-source-registration.md @@ -98,7 +98,7 @@ absence of one proves it did not. Completeness at the boundary is | `T02` attestation scheduling | **No** | Evidence *quality*, not custody. See below. | | `T04` heartbeat / missing-heartbeat findings | No | Detection of adversarial omission for rare classes — the one that matters most for revocation | | `T06` reconciliation counts | No | The source's own ability to detect divergence | -| `T05` declared cadence | No | Held deliberately on the §17 Taxonomy schema | +| `T05` declared cadence | No | Done 2026-09-22: `emission_cadence` on the registration, info-tech-canon wire schema 0.1 | **T02 is an evidence-quality gate, not an admission or deployment blocker.** Admission depends on identity, scope, ingress and a token. Attestation freshness diff --git a/docs/stream-completeness.md b/docs/stream-completeness.md index e80680a..8621c5c 100644 --- a/docs/stream-completeness.md +++ b/docs/stream-completeness.md @@ -1,6 +1,6 @@ # Stream completeness: heartbeats, reconciliation, and findings -`AUDIT-WP-0009` T04, T06, T07. Statute §9.6. +`AUDIT-WP-0009` T04, T05, 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 @@ -16,6 +16,7 @@ cadence to a system with nowhere to put it, so this is the surface owed. | --- | --- | --- | | 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 | +| Declared rate (`emission_cadence` on the registration) | A high-volume class falling below the rate its source declared | A class that may legitimately be silent — that is the heartbeat's job; and a rate held proves nothing about completeness | | 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 @@ -52,14 +53,12 @@ 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 +### Heartbeat and cadence are complementary -`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. +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. Cadence +describes a stream's expected **rate**. They are not alternatives, and the rate +shape is not audit-core's — see [Declared emission cadence](#declared-emission-cadence). ### Findings @@ -114,3 +113,57 @@ Two refusals worth knowing: 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. + +## Declared emission cadence + +`AUDIT-WP-0009-T05`. The contract is info-tech-canon's EmissionCadence +declaration, **wire schema 0.1**, read at contract digest `b08b4d95fc4b0bd3` +(standard document 0.2.0, candidate). audit-core did not invent the shape; it +waited for it. `audit_core/emission_cadence.py` validates a declaration against +that schema by hand — the receiver stays stdlib-only — and refuses what the +schema refuses rather than relaxing a rule to admit a declaration. + +The source owns and publishes its declaration. It reaches audit-core as +`emission_cadence` on the sender registration (the declaration object, as +JSON). An invalid declaration refuses the registration. The ConfigMap scope +overlay may not set or change it: a partial overlay has no sound merge with a +digest-pinned declaration, and a wholesale replacement could widen a window or +drop a class without anyone deciding to. + +What each declared form gets here: + +| Form | What audit-core does | +| --- | --- | +| `expected-rate` | Evaluated. For each permitted source, events of the class in the trailing `window` ending now are counted; fewer than `expected_min` is a `below_declared_cadence` finding on `GET /v1/stream-findings`. The count is the one `/v1/reconciliation` returns, so a finding and a count cannot disagree. | +| `heartbeat-or-reconciliation` | Validated and kept, not re-evaluated. The reconciliation half is `/v1/reconciliation`, whose count is the `compare_observed` side; `compare_local` belongs to the source. The heartbeat half is evaluated from `heartbeat_classes` (see below). | + +Refusals, same as for heartbeats: a wildcard source is not held to a rate, and +two `expected-rate` entries for one class are refused — each would be satisfied +by the other's events. + +Every cadence finding carries a `means` field: fewer events than declared says +the source may have stopped **or the declaration is wrong**, and is not proof of +suppression; a rate held would not prove completeness. + +### Observer evaluation, 2026-09-22 + +info-tech-canon asked whether audit-core would evaluate the one published +source-owned declaration, net-kingdom `local-identity/emission-cadence.yaml` +(commit `116643f`, pinned to the earlier digest `972c0b6701d1693f`; the wire +schema is identical). The result, stated at no more than it is: + +- **Structural: clean.** Both entries (`serve/token.token_issued`, + `revoke-token`) validate as `heartbeat-or-reconciliation` with a + reconciliation block. Asserted in `tests/test_emission_cadence.py`. +- **Operational: not evaluated.** audit-core registers no `local-identity` + sender and holds none of its events, so there is no observed count behind + `compare_observed` here either. This is not an observer result on the stream, + and nobody should count it as one. +- **One incompatibility, recorded.** The contract's `heartbeat.event_class` + names the heartbeat event's own class (e.g. `flex-auth.decision.heartbeat`). + audit-core's heartbeat is a single class, `audit-core.heartbeat`, carrying + the vouched-for class in `data.class`, declared per class in + `heartbeat_classes`. A declaration's heartbeat block therefore does not map + onto audit-core's evaluation by itself: a source needs `heartbeat_classes` on + its registration too, and the two can drift apart. That is fixable on + either side and has not been fixed on either. diff --git a/tests/test_emission_cadence.py b/tests/test_emission_cadence.py new file mode 100644 index 0000000..8a152bd --- /dev/null +++ b/tests/test_emission_cadence.py @@ -0,0 +1,219 @@ +"""AUDIT-WP-0009-T05 — declared emission cadence, evaluated by the observer.""" + +import copy +import json + +import pytest + +from audit_core.emission_cadence import ( + BELOW_CADENCE, + CONTRACT_DIGEST, + duration_seconds, + evaluate, + parse_declaration, +) +from audit_core.ingestion import IngestionApplication +from audit_core.senders import SenderIdentity, SenderRegistry, _identity_from +from audit_core.sqlite_backend import SQLiteAuditBackend +from tests.test_stream_findings import emit, invoke + +RATE = { + "schema_version": "0.1", + "declaration_id": "approval-engine.audit.v1", + "source": "approval-engine", + "stream_id": "approval-engine.audit", + "sources": [ + { + "source_id": "approval-engine.issued", + "event_class": "approval.issued", + "form": "expected-rate", + "window": "PT1H", + "expected_min": 2, + "drop_below": "finding", + }, + { + "source_id": "approval-engine.revocation", + "event_class": "approval.revocation", + "form": "heartbeat-or-reconciliation", + "heartbeat": { + "event_class": "approval-engine.heartbeat", + "interval": "24h", + "assertion": "nothing-to-report", + "missing": "finding", + }, + }, + ], +} + +# net-kingdom commit 116643f, local-identity/emission-cadence.yaml, as JSON. +# The one real source-owned declaration published so far. +NET_KINGDOM_LOCAL_IDENTITY = { + "schema_version": "0.1", + "declaration_id": "net-kingdom.local-identity.audit.v1", + "source": "net-kingdom", + "stream_id": "net-kingdom.local-identity.audit", + "extensions": {"netkingdom": {"contract_digest": "972c0b6701d1693f", + "contract_document_version": "0.2.0"}}, + "sources": [ + { + "source_id": f"net-kingdom.local-identity.audit.{name}", + "source_system": "local-identity", + "event_class": cls, + "form": "heartbeat-or-reconciliation", + "reconciliation": { + "compare_local": f"audit_log_counts.{cls}", + "compare_observed": f"evidence_counts.{cls}", + "divergence": "finding", + }, + "extensions": {"netkingdom": {"evidence_class": "load-bearing", + "rate_monitoring": "forbidden", + "completeness_claimed": False, + "heartbeat_emitted": False}}, + } + for name, cls in (("token-issued", "serve/token.token_issued"), + ("token-revoked", "revoke-token")) + ], +} + + +def _sender(declaration=RATE, sources=("approval-engine",)): + return SenderIdentity( + name="approval-engine", tokens=("approval",), sources=frozenset(sources), + tenants=frozenset({"tenant:platform"}), + emission_cadence=parse_declaration(declaration), + ) + + +@pytest.mark.parametrize("text,seconds", [ + ("PT1H", 3600), ("P1D", 86400), ("P1DT30M", 88200), ("90s", 90), ("5m", 300), ("1d", 86400), +]) +def test_durations_follow_the_schema_pattern(text, seconds): + assert duration_seconds(text) == seconds + + +@pytest.mark.parametrize("text", ["P", "PT", "PT0S", "0s", "1w", "3600", "P1H"]) +def test_durations_the_schema_refuses_are_refused(text): + with pytest.raises(ValueError): + duration_seconds(text) + + +def test_both_forms_parse_and_only_rates_are_indexed_for_evaluation(): + declaration = parse_declaration(RATE) + assert [(r.event_class, r.window_seconds, r.expected_min) for r in declaration.rates] == [ + ("approval.issued", 3600, 2) + ] + assert declaration.summary()["heartbeat_or_reconciliation_classes"] == ["approval.revocation"] + assert declaration.summary()["contract_digest"] == CONTRACT_DIGEST + + +def test_the_published_net_kingdom_declaration_is_accepted(): + """Observer evaluation of the one real declaration: it fits the contract.""" + declaration = parse_declaration(NET_KINGDOM_LOCAL_IDENTITY) + assert declaration.rates == () + assert len(declaration.other_entries) == 2 + + +def _broken(mutate): + doc = copy.deepcopy(RATE) + mutate(doc) + return doc + + +@pytest.mark.parametrize("mutate", [ + lambda d: d.update(schema_version="0.2"), + lambda d: d.pop("declaration_id"), + lambda d: d.update(sources=[]), + lambda d: d.update(unknown=1), + lambda d: d["sources"][0].pop("expected_min"), + lambda d: d["sources"][0].pop("drop_below"), + lambda d: d["sources"][0].update(window_seconds=3600), # both window forms + lambda d: d["sources"][0].pop("window"), # neither + lambda d: d["sources"][0].update(expected_min=-1), + lambda d: d["sources"][0].update(heartbeat=RATE["sources"][1]["heartbeat"]), + lambda d: d["sources"][0].update(form="rate"), + lambda d: d["sources"][1].pop("heartbeat"), # neither half + lambda d: d["sources"][1].update(window="PT1H"), + lambda d: d["sources"][1]["heartbeat"].pop("missing"), + lambda d: d["sources"][1]["heartbeat"].update(interval_seconds=60), + lambda d: d["sources"].append(copy.deepcopy(d["sources"][0])), # duplicate class +]) +def test_what_the_schema_refuses_is_refused(mutate): + with pytest.raises(ValueError): + parse_declaration(_broken(mutate)) + + +def _counter(counts): + return lambda source, since, until: [{"class": c, "count": n} for c, n in counts.items()] + + +def test_a_stream_below_its_declared_rate_is_a_finding(): + findings = evaluate([_sender()], _counter({"approval.issued": 1})) + assert len(findings) == 1 + finding = findings[0].as_dict() + assert finding["kind"] == BELOW_CADENCE + assert (finding["observed"], finding["expected_min"]) == (1, 2) + assert "not proof of suppression" in finding["means"] + + +def test_a_stream_at_its_declared_rate_is_not_a_finding(): + assert evaluate([_sender()], _counter({"approval.issued": 2})) == [] + + +def test_an_undeclared_class_is_never_a_finding(): + """Heartbeat-or-reconciliation classes are T04/T06's, not rate-monitored.""" + findings = evaluate([_sender()], _counter({"approval.issued": 5})) + assert all(f.event_class != "approval.revocation" for f in findings) + + +def test_a_wildcard_source_is_not_held_to_a_rate(): + assert evaluate([_sender(sources=("*",))], _counter({})) == [] + + +def test_the_registration_carries_the_declaration(): + identity = _identity_from({ + "name": "approval-engine", "token": "t", "sources": ["approval-engine"], + "emission_cadence": RATE, + }) + assert identity.evidence_declaration()["emission_cadence"]["declaration_id"] == ( + "approval-engine.audit.v1" + ) + + +def test_an_invalid_declaration_refuses_the_registration(): + with pytest.raises(ValueError): + _identity_from({ + "name": "x", "token": "t", "sources": ["x"], + "emission_cadence": _broken(lambda d: d.pop("sources")), + }) + + +def test_the_overlay_may_not_touch_the_declaration(tmp_path): + base = {"name": "approval-engine", "token": "t", "sources": ["approval-engine"], + "emission_cadence": RATE} + scope = tmp_path / "scope.json" + scope.write_text(json.dumps([{"name": "approval-engine", "emission_cadence": RATE}])) + with pytest.raises(ValueError, match="emission_cadence"): + SenderRegistry.from_env({ + "AUDIT_CORE_SENDERS": json.dumps([base]), + "AUDIT_CORE_SENDERS_SCOPE_PATH": str(scope), + }) + + +def test_the_findings_surface_reports_a_cadence_miss_and_clears_on_delivery(tmp_path): + operator = SenderIdentity( + name="operator", tokens=("operator",), sources=frozenset({"approval-engine"}), + tenants=frozenset({"*"}), may_write=False, may_read=True, + ) + app = IngestionApplication( + SQLiteAuditBackend(str(tmp_path / "s.db")), SenderRegistry([_sender(), operator]) + ) + + def misses(): + status, body = invoke(app, "/v1/stream-findings", token="operator") + assert status.startswith("200") + return [f for f in body["stream_findings"] if f["kind"] == BELOW_CADENCE] + + assert [f["observed"] for f in misses()] == [0] + for n in range(2): + assert emit(app, f"evt-{n}", "approval.issued")[0].startswith("202") + assert misses() == [] diff --git a/workplans/AUDIT-WP-0009-evidence-role-conformance.md b/workplans/AUDIT-WP-0009-evidence-role-conformance.md index ec365bc..9335304 100644 --- a/workplans/AUDIT-WP-0009-evidence-role-conformance.md +++ b/workplans/AUDIT-WP-0009-evidence-role-conformance.md @@ -4,12 +4,12 @@ type: workplan title: "Evidence-role conformance under Security Layer Model v0.7" domain: infotech repo: audit-core -status: active +status: finished flavor: implementation owner: claude topic_slug: railiance created: "2026-08-29" -updated: "2026-09-21" +updated: "2026-09-22" depends_on: - AUDIT-WP-0007 state_hub_workstream_id: "46a96b03-bc08-53b5-9c93-4071adabf734" @@ -201,10 +201,27 @@ itself go missing, which rate monitoring can never produce. ```task id: AUDIT-WP-0009-T05 -status: wait +status: done priority: medium state_hub_task_id: "382575ca-1801-5a32-a93d-90a8b9c8adbf" ``` +Done 2026-09-22. The wait ended when info-tech-canon published the contract +(message `368e54bf`): `emission-cadence.schema.yaml`, wire schema 0.1, read at +contract digest `b08b4d95fc4b0bd3`. No competing shape was invented. +`audit_core/emission_cadence.py` validates a source-owned declaration carried +as `emission_cadence` on the sender registration; `expected-rate` entries are +evaluated against the same per-class counts `/v1/reconciliation` returns and +surface as `below_declared_cadence` on `GET /v1/stream-findings`. +`heartbeat-or-reconciliation` entries are validated and left to T04/T06. The +scope overlay may not touch a declaration. Thirty-nine tests in +`tests/test_emission_cadence.py`. + +Observer evaluation of net-kingdom's `local-identity` declaration: structurally +clean, operationally not evaluated (no such sender is registered here), and one +incompatibility recorded — the contract's `heartbeat.event_class` names a +distinct heartbeat class, where audit-core's heartbeat is one class carrying +the vouched-for class in `data.class`. See `docs/stream-completeness.md`. + Accept and evaluate a declared emission cadence per source, and raise a finding when the stream falls below it. **Waiting on** the §17 emission-cadence declaration schema, which `kings-guard` is drafting and Taxonomy will own; do