From b9349782f44bbe4c2644be6795f4c175f42f951a Mon Sep 17 00:00:00 2001 From: tegwick Date: Sat, 5 Sep 2026 01:39:48 +0200 Subject: [PATCH] feat(audit): publish sequenced heartbeat and reconciliation evidence Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a06ec5-7e2b-7743-ac08-719e1b0f42e2 --- SCOPE.md | 2 +- WORK-RECORDS.md | 12 +- deploy/k8s/qonto-assistant/configmap.yaml | 1 + docs/SecurityPractice.md | 42 ++++--- docs/operator-runbook.md | 27 ++++ specs/ArchitectureBlueprint.md | 11 ++ specs/audit-emission-cadence.yaml | 66 ++++++++++ specs/security-genome.yaml | 13 +- src/qonto_assistant/app.py | 67 +++++++--- src/qonto_assistant/audit.py | 99 ++++++++++++++- src/qonto_assistant/config.py | 19 ++- src/qonto_assistant/contracts.py | 3 + src/qonto_assistant/credentials.py | 4 +- src/qonto_assistant/key_cape_auth.py | 1 + src/qonto_assistant/main.py | 1 - src/qonto_assistant/mcp_server.py | 6 +- src/qonto_assistant/policy.py | 25 +++- src/qonto_assistant/qonto_client.py | 19 ++- src/qonto_assistant/service.py | 58 +++++++-- tests/test_api.py | 60 ++++++++- tests/test_audit.py | 118 +++++++++++++++++- tests/test_audit_parity.py | 31 ++++- tests/test_deny_escalation.py | 36 ++++-- tests/test_flex_auth_client.py | 15 ++- tests/test_key_cape_auth.py | 11 +- tests/test_live_authorization_gate.py | 35 ++++-- tests/test_mcp_server.py | 8 +- tests/test_policy.py | 20 ++- tests/test_snapshot_semantics.py | 4 +- tests/test_tenant_engine_client.py | 4 +- workplans/ADHOC-2026-09-05.md | 25 ++++ ...-WP-0005-audit-deny-stream-completeness.md | 107 ++++++++++++++++ 32 files changed, 839 insertions(+), 111 deletions(-) create mode 100644 specs/audit-emission-cadence.yaml create mode 100644 workplans/ADHOC-2026-09-05.md create mode 100644 workplans/QONTO-WP-0005-audit-deny-stream-completeness.md diff --git a/SCOPE.md b/SCOPE.md index b9affb8..6f169d6 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -43,7 +43,7 @@ client. | INTENT / SCOPE / AGENTS | present | | Architecture blueprint | `specs/ArchitectureBlueprint.md` | | Research notes | `research/2026-07-21-…` | -| Runtime implementation | not started (see workplans) | +| Runtime implementation | REST + MCP, policy/auth gates, audit reconciliation implemented (see workplans) | | Upstream custody | live in platform (`binky-qonto-api`, CCR-2026-0008) | ## Getting oriented diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index 6cbe11b..9d365d6 100644 --- a/WORK-RECORDS.md +++ b/WORK-RECORDS.md @@ -8,13 +8,16 @@ | Kind | ID | Status | Lane | Source | | --- | --- | --- | --- | --- | -| workplan | ADHOC-2026-07-21 | finished | — | workplans/ADHOC-2026-07-21.md | +| workplan | QONTO-WP-ADHOC-2026-07-21 | finished | — | workplans/ADHOC-2026-07-21.md | +| workplan | ADHOC-2026-09-05 | finished | — | workplans/ADHOC-2026-09-05.md | | workplan | QONTO-WP-0001 | finished | — | workplans/QONTO-WP-0001-statehub-bootstrap.md | | workplan | QONTO-WP-0002 | finished | — | workplans/QONTO-WP-0002-policy-kernel-and-rest.md | | workplan | QONTO-WP-0003 | finished | — | workplans/QONTO-WP-0003-mcp-surface.md | | workplan | QONTO-WP-0004 | finished | — | workplans/QONTO-WP-0004-security-hardening-and-scale-to-zero.md | -| task | ADHOC-2026-07-21-T01 | done | — | workplans/ADHOC-2026-07-21.md | -| task | ADHOC-2026-07-21-T02 | done | — | workplans/ADHOC-2026-07-21.md | +| workplan | QONTO-WP-0005 | finished | — | workplans/QONTO-WP-0005-audit-deny-stream-completeness.md | +| task | QONTO-WP-ADHOC-2026-07-21-T01 | done | — | workplans/ADHOC-2026-07-21.md | +| task | QONTO-WP-ADHOC-2026-07-21-T02 | done | — | workplans/ADHOC-2026-07-21.md | +| task | ADHOC-2026-09-05-T01 | done | — | workplans/ADHOC-2026-09-05.md | | task | QONTO-WP-0001-T01 | done | — | workplans/QONTO-WP-0001-statehub-bootstrap.md | | task | QONTO-WP-0001-T02 | done | — | workplans/QONTO-WP-0001-statehub-bootstrap.md | | task | QONTO-WP-0001-T03 | done | — | workplans/QONTO-WP-0001-statehub-bootstrap.md | @@ -40,3 +43,6 @@ | task | QONTO-WP-0004-T06 | done | — | workplans/QONTO-WP-0004-security-hardening-and-scale-to-zero.md | | task | QONTO-WP-0004-T07 | done | — | workplans/QONTO-WP-0004-security-hardening-and-scale-to-zero.md | | task | QONTO-WP-0004-T08 | done | — | workplans/QONTO-WP-0004-security-hardening-and-scale-to-zero.md | +| task | QONTO-WP-0005-T01 | done | — | workplans/QONTO-WP-0005-audit-deny-stream-completeness.md | +| task | QONTO-WP-0005-T02 | done | — | workplans/QONTO-WP-0005-audit-deny-stream-completeness.md | +| task | QONTO-WP-0005-T03 | done | — | workplans/QONTO-WP-0005-audit-deny-stream-completeness.md | diff --git a/deploy/k8s/qonto-assistant/configmap.yaml b/deploy/k8s/qonto-assistant/configmap.yaml index 41b85c9..9aeb446 100644 --- a/deploy/k8s/qonto-assistant/configmap.yaml +++ b/deploy/k8s/qonto-assistant/configmap.yaml @@ -24,6 +24,7 @@ data: QONTO_RATE_LIMIT_WINDOW_SECONDS: "60" QONTO_MAX_CONCURRENCY: "4" QONTO_DENY_ESCALATION_ENABLED: "true" + QONTO_AUDIT_HEARTBEAT_INTERVAL_SECONDS: "86400" # QONTO_KEY_CAPE_JWKS_URL: TODO once key-cape has a confirmed in-cluster address # QONTO_FLEX_AUTH_URL: TODO once flex-auth has a confirmed in-cluster address # QONTO_TENANT_ENGINE_URL: TODO once tenant-engine is deployed anywhere reachable diff --git a/docs/SecurityPractice.md b/docs/SecurityPractice.md index 62d9f9e..9e27ee4 100644 --- a/docs/SecurityPractice.md +++ b/docs/SecurityPractice.md @@ -195,11 +195,11 @@ effect of how it was built, not extra work: recovery procedure with no data-loss risk (`maximum_data_loss: PT0S` applies trivially since there is no persisted data). -## 9. Kings Guard mapping (prep now, cheap; enforcement later) +## 9. King's Guard observation mapping -Nothing below requires `kings-guard` to exist yet. It is preparation so -`qonto-assistant` needs zero rework once a sentinel-mesh or decision plane -does exist. +King's Guard has observed the structured allow/deny stream. The source keeps +ownership of its genome and emission claims; the observer independently checks +the records and completeness evidence it receives. ### 9.1 Security Genome record @@ -210,21 +210,29 @@ nothing else), data classification, and recovery expectations. ### 9.2 Audit stream is already observation-shaped -`AuditLogger`'s existing event shape (actor, capability, decision, -deny_reason, latency, upstream HTTP status, policy version — see -`tests/test_audit.py`, `tests/test_audit_parity.py`) already matches Kings -Guard's Immune Observation contract closely enough that no schema rework -should be needed later — just a new consumer pointed at the same stream. +`AuditLogger`'s event shape includes actor, capability, decision, deny reason, +latency, upstream HTTP status, policy version, identity binding, and egress +destination (see `tests/test_audit.py`, `tests/test_audit_parity.py`). Every +record also carries a process-instance id and monotonic stream sequence. + +The source-owned declaration in `specs/audit-emission-cadence.yaml` classifies +`audit.deny` as load-bearing and low-volume. A startup heartbeat, periodic +heartbeat (24 hours by default), and best-effort shutdown heartbeat publish +cumulative and window transition counts. Consumers compare those counts and +sequence continuity with observed records. `/v1/audit/reconciliation` exposes +the same non-secret current source counts for an authenticated diagnostic read. ### 9.3 A concrete, actionable-today signal This does not require any Kings Guard component: repeated `arg_constraint` or `credential_exfil` deny reasons from the same actor within a short window is a real, current signal. It should trip a -tightened rate limit or a temporary lockout for that actor now, using the -audit stream and rate limiter that already exist — this is a Fast Local -Loop (`NetKingdomImmuneArchitecture.md` §14.1) response that does not need -to wait for any future component. +tightened rate limit or a temporary lockout for that actor now. The current +`DenyEscalationTracker` branches directly on the same decision path that emits +the audit event; it does not consume the audit stream. This is a Fast Local Loop +(`NetKingdomImmuneArchitecture.md` §14.1) response that does not need to wait +for any future component, while the stream remains load-bearing for estate +observation rather than for the local lockout. --- @@ -235,12 +243,12 @@ to wait for any future component. | Security Genome record | `qonto-assistant` | can ship now | | Audit-stream shape review against Immune Observation contract | `qonto-assistant` | can ship now | | Actor lockout on repeated deny signals | `qonto-assistant` | can ship now | -| `key-cape` token verification in place of bearer token | `qonto-assistant` + `key-cape` | needs `key-cape` client integration support | -| `finance.qonto.read` resource + live decision call | `qonto-assistant` + `flex-auth` | needs the resource registered in `flex-auth` | -| Live tenant-role gate | `qonto-assistant` + `tenant-engine` | needs a `tenant-engine` lookup/cache API call wired in | +| `key-cape` token verification in place of bearer token | `qonto-assistant` + `key-cape` | verifier shipped; live enforcement is runtime-configured | +| `finance.qonto.read` resource + live decision call | `qonto-assistant` + `flex-auth` | client and policy shipped; live use is runtime-configured | +| Live tenant-role gate | `qonto-assistant` + `tenant-engine` | client shipped; live use requires a reachable deployment URL | | Facade / scale-to-zero activator | new component (home TBD — Railiance or a dedicated repo) | design only so far | | I1/I2 isolation placement on `railiance01` | Railiance | needs a placement decision | -| Sentinel-mesh / decision-plane consumption of the audit stream | `kings-guard` | does not exist yet — see intake | +| Audit-stream observation and completeness acceptance | `kings-guard` | consuming events; cadence/reconciliation acceptance pending | Tracked as `QONTO-WP-0004` in this repo, with the `kings-guard`-owned portion tracked as an intake against `KG-WP-0002` (pilot-lane selection). diff --git a/docs/operator-runbook.md b/docs/operator-runbook.md index 90f1275..195250c 100644 --- a/docs/operator-runbook.md +++ b/docs/operator-runbook.md @@ -9,6 +9,7 @@ Phase 1 ships: - policy-gated `GET /v1/transactions` - policy-gated `GET /v1/snapshot` - structured audit events without secrets +- sequenced audit heartbeat and reconciliation evidence - env-backed or OpenBao-CLI-backed credential loading Spend, transfer, card, invoicing, payment-link, and other volume-cost actions @@ -207,6 +208,32 @@ That consumer-side write remains outside this repo. the current dogfood path remains `legacy_api_key` because that is the proven BINKY-WP-0005 header mode. +## Audit deny-stream reconciliation + +`audit.deny` is a low-volume, load-bearing observation class. Do not infer +completeness from a minimum event rate. The service emits `audit.heartbeat` at +startup, every `QONTO_AUDIT_HEARTBEAT_INTERVAL_SECONDS` while the process is +active (default `86400`), and best-effort at shutdown. Each request event and +heartbeat carries `stream_instance_id` plus a monotonic `stream_sequence`. + +For each instance, compare received `audit.allow` / `audit.deny` totals with +the heartbeat's `source_transition_counts` and reject sequence gaps. Window +counts describe transitions since the prior heartbeat. A quiet deny window has +`assertion: nothing-to-report`; a non-quiet one has +`assertion: transitions-reported`. Counters are process-local and reset only +when `stream_instance_id` changes. + +The authenticated diagnostic view is: + +```text +GET /v1/audit/reconciliation +``` + +It returns only stream identity, timestamps, sequence, and counts. It does not +write State Hub, query an observer, include actor/bank data, or increment the +stream it describes. The exact contract is declared in +`specs/audit-emission-cadence.yaml`. + ## Live authorization gate (flex-auth + tenant-engine) Off by default (no `QONTO_FLEX_AUTH_URL` set). When configured, every diff --git a/specs/ArchitectureBlueprint.md b/specs/ArchitectureBlueprint.md index 45396f3..c36647d 100644 --- a/specs/ArchitectureBlueprint.md +++ b/specs/ArchitectureBlueprint.md @@ -416,8 +416,13 @@ Suggested audit envelope: ```yaml request_id: req-... timestamp: ... +event_class: audit.allow | audit.deny +stream_id: qonto-assistant.audit +stream_instance_id: ... +stream_sequence: ... actor: agt-... / workload id tenant_id: binky +identity_binding: self_asserted | key_cape_jwt capability: list_transactions protocol: rest | mcp decision: allow | deny @@ -426,9 +431,15 @@ policy_version: 1 latency_ms: ... qonto_http_status: ... result_count: ... +egress_destination: qonto-thirdparty-api # never: Authorization header, API_KEY, OpenBao token, full IBAN by default ``` +The load-bearing `audit.deny` class uses the source-owned cadence and +reconciliation contract in `specs/audit-emission-cadence.yaml`. Heartbeats +carry cumulative and per-window source transition counts; observers compare +those counts and per-instance sequence numbers with received records. + ### 4.11 Operational guardrails (v1, not "later if we remember") Basic safety and reliability controls belong in Phase 1 because one noisy diff --git a/specs/audit-emission-cadence.yaml b/specs/audit-emission-cadence.yaml new file mode 100644 index 0000000..e358dd3 --- /dev/null +++ b/specs/audit-emission-cadence.yaml @@ -0,0 +1,66 @@ +# Source-owned declaration for the qonto-assistant structured audit stream. +# The semantic shape follows the Taxonomy draft in +# kings-guard/specs/EmissionCadenceDeclaration.md. This repository owns the +# concrete emission claim even while the fleet-wide schema remains a draft. + +audit_emission_cadence: + schema_version: "0.1" + source: qonto-assistant + stream_id: qonto-assistant.audit + transport: structured-stdout + instance_boundary: process + + ordering: + instance_field: stream_instance_id + sequence_field: stream_sequence + sequence_starts_at: 1 + gap: finding + reset: permitted-only-when-stream_instance_id-changes + + event_classes: + audit.allow: + evidence_class: attributive + completeness_claimed: false + audit.deny: + evidence_class: load-bearing + form: heartbeat-or-reconciliation + rate_monitoring: forbidden + completeness_claimed: true + audit.heartbeat: + evidence_class: load-bearing + role: positive-liveness-and-reconciliation-claim + + heartbeat: + event_class: audit.heartbeat + interval: PT24H + interval_config: QONTO_AUDIT_HEARTBEAT_INTERVAL_SECONDS + default_interval_seconds: 86400 + lifecycle: + startup: required + periodic_while_process_active: required + shutdown: best-effort + quiet_assertion: nothing-to-report + active_instance_missing: finding + scale_to_zero_semantics: > + No periodic heartbeat is promised while no process instance exists. + Observers track each stream_instance_id from its startup heartbeat until + a shutdown heartbeat or expiry after a missing active-instance cadence. + + reconciliation: + source_counts_field: source_transition_counts + window_counts_field: window_transition_counts + counted_classes: + - audit.allow + - audit.deny + snapshot_endpoint: /v1/audit/reconciliation + compare_observed: > + Count received request events by event_class and stream_instance_id, then + compare them with the heartbeat source counts and stream sequence. + divergence: finding + undrained_local: lag-not-divergence + persistence: process-local + + event_context: + identity_binding_field: identity_binding + egress_destination_field: egress_destination + qonto_egress_destination: qonto-thirdparty-api diff --git a/specs/security-genome.yaml b/specs/security-genome.yaml index 37c31c9..f50a077 100644 --- a/specs/security-genome.yaml +++ b/specs/security-genome.yaml @@ -1,9 +1,9 @@ # Security Genome record for qonto-assistant. # # Schema per kings-guard/specs/NetKingdomImmuneArchitecture.md §9.1 -# ("Minimum Genome Record"). Kings Guard does not exist as a running system -# yet — this record is written now so no rework is needed once an admission -# or posture-assessment consumer does exist. See docs/SecurityPractice.md. +# ("Minimum Genome Record"). King's Guard now consumes this source's structured +# audit events; the adjacent cadence declaration supplies stream-completeness +# semantics without making the consumer an authority for source intent. security_genome_record: id: kg:genome:qonto-assistant @@ -24,7 +24,7 @@ security_genome_record: criticality: high identities: - workload_identity: "TBD — pending key-cape/workload-identity integration (QONTO-WP-0004)" + workload_identity: "key-cape JWT verification supported; runtime enforcement is configuration-bound" deployment_identity: "TBD — pending Railiance placement decision" capabilities: @@ -84,3 +84,8 @@ security_genome_record: bound to verified identity. Tolerated until key-cape/flex-auth integration (QONTO-WP-0004) closes this. expires_at: null + +# Source emission claim for the load-bearing deny observation class. The full +# machine-readable declaration is kept separate so consumers can ingest it +# without interpreting the genome schema. +audit_emission_cadence: audit-emission-cadence.yaml diff --git a/src/qonto_assistant/app.py b/src/qonto_assistant/app.py index 6f95bfb..ef741d7 100644 --- a/src/qonto_assistant/app.py +++ b/src/qonto_assistant/app.py @@ -1,8 +1,9 @@ from __future__ import annotations -from collections.abc import AsyncIterator, Callable -from contextlib import AsyncExitStack, asynccontextmanager, suppress +import asyncio import logging +from collections.abc import AsyncIterator +from contextlib import AsyncExitStack, asynccontextmanager, suppress from uuid import uuid4 import uvicorn @@ -39,13 +40,18 @@ def create_app( key_cape_verifier: KeyCapeTokenVerifier | None = None, ) -> FastAPI: settings = settings or Settings.from_env() - audit_logger = audit_logger or AuditLogger() - service = service or _build_service( - settings=settings, - audit_logger=audit_logger, - rate_limiter=rate_limiter, - concurrency_limiter=concurrency_limiter, - ) + if service is not None: + if audit_logger is not None and audit_logger is not service.audit_logger: + raise ValueError("service and app must share one audit_logger") + audit_logger = service.audit_logger + else: + audit_logger = audit_logger or AuditLogger() + service = _build_service( + settings=settings, + audit_logger=audit_logger, + rate_limiter=rate_limiter, + concurrency_limiter=concurrency_limiter, + ) if key_cape_verifier is None and settings.key_cape_jwks_url: key_cape_verifier = KeyCapeTokenVerifier( jwks_url=settings.key_cape_jwks_url, @@ -57,7 +63,9 @@ def create_app( cache_seconds=settings.key_cape_cache_seconds, ) - mcp_server = create_mcp_server(settings=settings, service=service, key_cape_verifier=key_cape_verifier) + mcp_server = create_mcp_server( + settings=settings, service=service, key_cape_verifier=key_cape_verifier + ) mcp_app = mcp_server.streamable_http_app() if settings.mcp_auth_token: mcp_app.add_middleware(BearerTokenAuthMiddleware, token=settings.mcp_auth_token) @@ -66,9 +74,24 @@ def create_app( async def lifespan(_: FastAPI) -> AsyncIterator[None]: async with AsyncExitStack() as stack: await stack.enter_async_context(mcp_app.router.lifespan_context(mcp_app)) - yield - with suppress(Exception): - service.client.close() + audit_logger.emit_heartbeat(reason="startup") + heartbeat_task = asyncio.create_task( + _emit_audit_heartbeats( + audit_logger=audit_logger, + interval_seconds=settings.audit_heartbeat_interval_seconds, + ) + ) + try: + yield + finally: + try: + heartbeat_task.cancel() + with suppress(asyncio.CancelledError): + await heartbeat_task + audit_logger.emit_heartbeat(reason="shutdown") + finally: + with suppress(Exception): + service.client.close() app = FastAPI(title="qonto-assistant", version=__version__, lifespan=lifespan) app.state.settings = settings @@ -91,11 +114,21 @@ def create_app( "policy_file": str(settings.policy_file), } + @app.get("/v1/audit/reconciliation") + async def audit_reconciliation(request: Request) -> dict[str, object]: + # Apply the same deployment identity boundary as finance calls. The + # view contains counts only and deliberately does not create another + # transition in the stream it is describing. + actor_claims_from_request(request, settings, key_cape_verifier=key_cape_verifier) + return audit_logger.reconciliation_snapshot() + @app.get("/v1/accounts") async def get_accounts(request: Request) -> JSONResponse: claims = actor_claims_from_request(request, settings, key_cape_verifier=key_cape_verifier) request_id = _request_id(request) - payload = await run_in_threadpool(service.get_accounts, claims=claims, request_id=request_id) + payload = await run_in_threadpool( + service.get_accounts, claims=claims, request_id=request_id + ) return JSONResponse(content=payload, headers={"X-Request-ID": request_id}) @app.get("/v1/transactions") @@ -143,6 +176,12 @@ def create_app( return app +async def _emit_audit_heartbeats(*, audit_logger: AuditLogger, interval_seconds: int) -> None: + while True: + await asyncio.sleep(interval_seconds) + audit_logger.emit_heartbeat(reason="periodic") + + def _request_id(request: Request) -> str: existing = getattr(request.state, "request_id", None) if existing: diff --git a/src/qonto_assistant/audit.py b/src/qonto_assistant/audit.py index 9ab8a6d..4ffc184 100644 --- a/src/qonto_assistant/audit.py +++ b/src/qonto_assistant/audit.py @@ -3,9 +3,12 @@ from __future__ import annotations import json import logging from collections.abc import Callable, Mapping +from copy import deepcopy from dataclasses import asdict, is_dataclass from datetime import UTC, datetime +from threading import RLock from typing import Any +from uuid import uuid4 from qonto_assistant.contracts import AuditEvent @@ -18,6 +21,8 @@ SECRET_KEYS = { "secret", "token", } +AUDIT_STREAM_ID = "qonto-assistant.audit" +AUDIT_EVENT_CLASSES = ("audit.allow", "audit.deny") def utc_now_iso() -> str: @@ -40,19 +45,107 @@ def _sanitize(value: Any) -> Any: class AuditLogger: + """Emit sequenced audit records and source-side reconciliation evidence. + + Counters represent source transitions, not observer delivery receipts. An + observer compares these counters with received ``audit.allow`` and + ``audit.deny`` records and also checks ``stream_sequence`` for gaps. + """ + def __init__( self, *, logger: logging.Logger | None = None, sink: Callable[[dict[str, Any]], None] | None = None, + instance_id: str | None = None, + clock: Callable[[], str] = utc_now_iso, ) -> None: self.logger = logger or logging.getLogger("qonto_assistant.audit") self.logger.setLevel(logging.INFO) self.sink = sink + self.instance_id = instance_id or str(uuid4()) + self.clock = clock + self.started_at = clock() + self._lock = RLock() + self._sequence = 0 + self._transition_counts = {event_class: 0 for event_class in AUDIT_EVENT_CLASSES} + self._heartbeat_counts = dict(self._transition_counts) + self._heartbeat_at = self.started_at def emit(self, event: AuditEvent | Mapping[str, Any]) -> dict[str, Any]: payload = _sanitize(asdict(event) if is_dataclass(event) else dict(event)) - if self.sink is not None: - self.sink(payload) + decision = payload.get("decision") + event_class = f"audit.{decision}" if decision in {"allow", "deny"} else "audit.event" + + with self._lock: + self._sequence += 1 + if event_class in self._transition_counts: + self._transition_counts[event_class] += 1 + payload.update(self._stream_fields(event_class=event_class, sequence=self._sequence)) + self._publish(payload) + + return deepcopy(payload) + + def emit_heartbeat(self, *, reason: str = "periodic") -> dict[str, Any]: + """Emit a positive liveness claim plus counts for observer reconciliation.""" + with self._lock: + now = self.clock() + self._sequence += 1 + window_counts = { + event_class: self._transition_counts[event_class] + - self._heartbeat_counts[event_class] + for event_class in AUDIT_EVENT_CLASSES + } + payload: dict[str, Any] = { + **self._stream_fields(event_class="audit.heartbeat", sequence=self._sequence), + "timestamp": now, + "reason": reason, + "assertion": ( + "nothing-to-report" + if window_counts["audit.deny"] == 0 + else "transitions-reported" + ), + "window_started_at": self._heartbeat_at, + "window_ended_at": now, + "source_transition_counts": dict(self._transition_counts), + "window_transition_counts": window_counts, + } + self._heartbeat_counts = dict(self._transition_counts) + self._heartbeat_at = now + self._publish(payload) + + return deepcopy(payload) + + def reconciliation_snapshot(self) -> dict[str, Any]: + """Return non-secret source state without adding an audit-stream record.""" + with self._lock: + now = self.clock() + return { + "stream_id": AUDIT_STREAM_ID, + "stream_instance_id": self.instance_id, + "instance_started_at": self.started_at, + "snapshot_at": now, + "last_stream_sequence": self._sequence, + "last_heartbeat_at": self._heartbeat_at, + "source_transition_counts": dict(self._transition_counts), + "since_last_heartbeat_counts": { + event_class: self._transition_counts[event_class] + - self._heartbeat_counts[event_class] + for event_class in AUDIT_EVENT_CLASSES + }, + } + + def _stream_fields(self, *, event_class: str, sequence: int) -> dict[str, Any]: + return { + "event_class": event_class, + "stream_id": AUDIT_STREAM_ID, + "stream_instance_id": self.instance_id, + "stream_sequence": sequence, + } + + def _publish(self, payload: dict[str, Any]) -> None: + # Structured logging is the production hot-path sink. Publish it first + # so an optional secondary sink cannot suppress the primary record. self.logger.info(json.dumps(payload, sort_keys=True)) - return payload + if self.sink is not None: + self.sink(deepcopy(payload)) diff --git a/src/qonto_assistant/config.py b/src/qonto_assistant/config.py index bbda29a..3e70b8f 100644 --- a/src/qonto_assistant/config.py +++ b/src/qonto_assistant/config.py @@ -32,6 +32,7 @@ class Settings: deny_escalation_threshold: int deny_escalation_window_seconds: int deny_escalation_lockout_seconds: int + audit_heartbeat_interval_seconds: int key_cape_jwks_url: str | None key_cape_issuer: str key_cape_audience: str @@ -81,10 +82,18 @@ class Settings: rate_limit_requests=int(os.getenv("QONTO_RATE_LIMIT_REQUESTS", "20")), rate_limit_window_seconds=int(os.getenv("QONTO_RATE_LIMIT_WINDOW_SECONDS", "60")), max_concurrency=int(os.getenv("QONTO_MAX_CONCURRENCY", "4")), - deny_escalation_enabled=os.getenv("QONTO_DENY_ESCALATION_ENABLED", "true").lower() == "true", + deny_escalation_enabled=os.getenv("QONTO_DENY_ESCALATION_ENABLED", "true").lower() + == "true", deny_escalation_threshold=int(os.getenv("QONTO_DENY_ESCALATION_THRESHOLD", "3")), - deny_escalation_window_seconds=int(os.getenv("QONTO_DENY_ESCALATION_WINDOW_SECONDS", "60")), - deny_escalation_lockout_seconds=int(os.getenv("QONTO_DENY_ESCALATION_LOCKOUT_SECONDS", "300")), + deny_escalation_window_seconds=int( + os.getenv("QONTO_DENY_ESCALATION_WINDOW_SECONDS", "60") + ), + deny_escalation_lockout_seconds=int( + os.getenv("QONTO_DENY_ESCALATION_LOCKOUT_SECONDS", "300") + ), + audit_heartbeat_interval_seconds=max( + 1, int(os.getenv("QONTO_AUDIT_HEARTBEAT_INTERVAL_SECONDS", "86400")) + ), key_cape_jwks_url=os.getenv("QONTO_KEY_CAPE_JWKS_URL") or None, key_cape_issuer=os.getenv("QONTO_KEY_CAPE_ISSUER", "https://key-cape.netkingdom"), key_cape_audience=os.getenv("QONTO_KEY_CAPE_AUDIENCE", "qonto-assistant"), @@ -94,7 +103,9 @@ class Settings: flex_auth_base_url=os.getenv("QONTO_FLEX_AUTH_URL") or None, flex_auth_timeout_seconds=float(os.getenv("QONTO_FLEX_AUTH_TIMEOUT_SECONDS", "3")), tenant_engine_base_url=os.getenv("QONTO_TENANT_ENGINE_URL") or None, - tenant_engine_timeout_seconds=float(os.getenv("QONTO_TENANT_ENGINE_TIMEOUT_SECONDS", "3")), + tenant_engine_timeout_seconds=float( + os.getenv("QONTO_TENANT_ENGINE_TIMEOUT_SECONDS", "3") + ), tenant_engine_required_roles=frozenset( role.strip() for role in os.getenv("QONTO_TENANT_ENGINE_REQUIRED_ROLES", "VEN,CUS").split(",") diff --git a/src/qonto_assistant/contracts.py b/src/qonto_assistant/contracts.py index 074fe67..1664607 100644 --- a/src/qonto_assistant/contracts.py +++ b/src/qonto_assistant/contracts.py @@ -12,6 +12,7 @@ class ActorClaims: tenant_id: str lane: str = "green" scopes: frozenset[str] = field(default_factory=frozenset) + identity_binding: str = "self_asserted" @dataclass(frozen=True, slots=True) @@ -52,5 +53,7 @@ class AuditEvent: deny_reason: str | None policy_version: int latency_ms: int + identity_binding: str + egress_destination: str qonto_http_status: int | None = None result_count: int | None = None diff --git a/src/qonto_assistant/credentials.py b/src/qonto_assistant/credentials.py index e2fb6c0..0d10f38 100644 --- a/src/qonto_assistant/credentials.py +++ b/src/qonto_assistant/credentials.py @@ -74,7 +74,9 @@ class OpenBaoCliCredentialProvider: return value -def build_credential_provider(settings: Settings) -> EnvironmentCredentialProvider | OpenBaoCliCredentialProvider: +def build_credential_provider( + settings: Settings, +) -> EnvironmentCredentialProvider | OpenBaoCliCredentialProvider: if settings.credential_source == "bao-cli": return OpenBaoCliCredentialProvider( command=settings.openbao_command, diff --git a/src/qonto_assistant/key_cape_auth.py b/src/qonto_assistant/key_cape_auth.py index 1bb81d1..52841cd 100644 --- a/src/qonto_assistant/key_cape_auth.py +++ b/src/qonto_assistant/key_cape_auth.py @@ -102,6 +102,7 @@ class KeyCapeTokenVerifier: tenant_id=str(claims["tenant"]), lane=_lane_from_roles(claims.get("roles"), default=self.default_lane), scopes=frozenset(_coerce_scopes(claims.get("scope") or claims.get("scp"))), + identity_binding="key_cape_jwt", ) def _ensure_keys(self) -> None: diff --git a/src/qonto_assistant/main.py b/src/qonto_assistant/main.py index 0de158b..3bf1961 100644 --- a/src/qonto_assistant/main.py +++ b/src/qonto_assistant/main.py @@ -1,5 +1,4 @@ from qonto_assistant.app import main - if __name__ == "__main__": main() diff --git a/src/qonto_assistant/mcp_server.py b/src/qonto_assistant/mcp_server.py index b7ed9e1..5698dba 100644 --- a/src/qonto_assistant/mcp_server.py +++ b/src/qonto_assistant/mcp_server.py @@ -86,7 +86,7 @@ def create_mcp_server( window_days: int = 90, page_size: int = 50, ) -> dict[str, Any]: - """Normalized recurring-cost hints, not a raw export (cost_run_rate_hints capability).""" + """Normalized recurring-cost hints for the cost_run_rate_hints capability.""" claims = _claims(ctx, settings, key_cape_verifier) return service.get_cost_run_rate_hints( claims=claims, @@ -103,7 +103,9 @@ def mcp_asgi_app(*, settings: Settings, service: CapabilityService | None = None return create_mcp_server(settings=settings, service=service).streamable_http_app() -def _claims(ctx: Context, settings: Settings, key_cape_verifier: KeyCapeTokenVerifier | None = None): +def _claims( + ctx: Context, settings: Settings, key_cape_verifier: KeyCapeTokenVerifier | None = None +): headers = _headers_from_context(ctx) return actor_claims_from_headers(headers, settings, key_cape_verifier=key_cape_verifier) diff --git a/src/qonto_assistant/policy.py b/src/qonto_assistant/policy.py index 6efc0ba..cd7a3e2 100644 --- a/src/qonto_assistant/policy.py +++ b/src/qonto_assistant/policy.py @@ -10,7 +10,9 @@ from qonto_assistant.contracts import CapabilityRequest, PolicyDecision class PolicyEngine: - def __init__(self, config: Mapping[str, Any], *, required_scope: str, enforce_scope: bool) -> None: + def __init__( + self, config: Mapping[str, Any], *, required_scope: str, enforce_scope: bool + ) -> None: self.config = dict(config) self.version = int(self.config.get("version", 1)) self.required_scope = required_scope @@ -54,7 +56,9 @@ class PolicyEngine: request_args=dict(request.request_args), ) - def _constraints_ok(self, capability: Mapping[str, Any], request_args: Mapping[str, Any]) -> bool: + def _constraints_ok( + self, capability: Mapping[str, Any], request_args: Mapping[str, Any] + ) -> bool: constraints = dict(capability.get("constraints", {})) checks = { "page_size": ("max_per_page", 1), @@ -78,9 +82,14 @@ class PolicyEngine: def _match_deny_classes(self, request: CapabilityRequest) -> str | None: spend_prefixes = tuple(self.deny_classes.get("spend", {}).get("match_prefixes", [])) volume_tags = tuple(self.deny_classes.get("volume_cost", {}).get("match_tags", [])) - credential_fields = set(self.deny_classes.get("credential_exfil", {}).get("response_fields", [])) + credential_fields = set( + self.deny_classes.get("credential_exfil", {}).get("response_fields", []) + ) - lowered_tokens = {token.lower() for token in self._flatten_strings(request.capability_id, request.request_args)} + lowered_tokens = { + token.lower() + for token in self._flatten_strings(request.capability_id, request.request_args) + } for prefix in spend_prefixes: lowered_prefix = prefix.lower() if any(token.startswith(lowered_prefix) for token in lowered_tokens): @@ -100,11 +109,15 @@ class PolicyEngine: if any(str(field).lower() in credential_fields for field in requested_fields): return "credential_exfil" - if request.request_args.get("include_full_iban") or request.request_args.get("include_api_key"): + if request.request_args.get("include_full_iban") or request.request_args.get( + "include_api_key" + ): return "credential_exfil" amount_keys = {"amount", "amount_cents", "amount_eur"} - if any(key in request.request_args for key in amount_keys) and request.request_args.get("execute"): + if any(key in request.request_args for key in amount_keys) and request.request_args.get( + "execute" + ): return "spend" return None diff --git a/src/qonto_assistant/qonto_client.py b/src/qonto_assistant/qonto_client.py index 6d510c3..b00b617 100644 --- a/src/qonto_assistant/qonto_client.py +++ b/src/qonto_assistant/qonto_client.py @@ -4,8 +4,7 @@ import json from collections.abc import Mapping from datetime import datetime from pathlib import Path -from typing import Any -from typing import Protocol +from typing import Any, Protocol import httpx @@ -101,7 +100,9 @@ class QontoClient: if attempt < self.max_retries: attempt += 1 continue - raise UpstreamError("Qonto request timed out", error_code="qonto_timeout", status_code=504) from exc + raise UpstreamError( + "Qonto request timed out", error_code="qonto_timeout", status_code=504 + ) from exc except httpx.TransportError as exc: if attempt < self.max_retries: attempt += 1 @@ -179,7 +180,9 @@ class FixtureQontoClient: status_code=500, ) - filtered: list[Mapping[str, Any]] = [item for item in transactions if isinstance(item, Mapping)] + filtered: list[Mapping[str, Any]] = [ + item for item in transactions if isinstance(item, Mapping) + ] if iban: filtered = [item for item in filtered if item.get("iban") in {None, "", iban}] if status: @@ -221,14 +224,18 @@ class FixtureQontoClient: ) return payload - def _filter_window(self, transactions: list[Mapping[str, Any]], window_days: int) -> list[Mapping[str, Any]]: + def _filter_window( + self, transactions: list[Mapping[str, Any]], window_days: int + ) -> list[Mapping[str, Any]]: dated = [] for transaction in transactions: settled_at = transaction.get("settled_at") or transaction.get("settledAt") if not settled_at: dated.append((None, transaction)) continue - dated.append((datetime.fromisoformat(str(settled_at).replace("Z", "+00:00")), transaction)) + dated.append( + (datetime.fromisoformat(str(settled_at).replace("Z", "+00:00")), transaction) + ) dates = [item[0] for item in dated if item[0] is not None] if not dates: diff --git a/src/qonto_assistant/service.py b/src/qonto_assistant/service.py index 92dc07b..b9425f2 100644 --- a/src/qonto_assistant/service.py +++ b/src/qonto_assistant/service.py @@ -7,7 +7,13 @@ from datetime import UTC, datetime from typing import Any from qonto_assistant.audit import AuditLogger, utc_now_iso -from qonto_assistant.contracts import ActorClaims, AuditEvent, CapabilityRequest, PolicyDecision, ProtocolName +from qonto_assistant.contracts import ( + ActorClaims, + AuditEvent, + CapabilityRequest, + PolicyDecision, + ProtocolName, +) from qonto_assistant.errors import InvalidRequestError, PolicyDeniedError, UpstreamError from qonto_assistant.live_authorization import LiveAuthorizationGate from qonto_assistant.policy import PolicyEngine @@ -102,7 +108,9 @@ class CapabilityService: resource_scope="snapshot", request_id=request_id, protocol=protocol, - operation=lambda _: self._build_snapshot_payload(window_days=window_days, page_size=page_size), + operation=lambda _: self._build_snapshot_payload( + window_days=window_days, page_size=page_size + ), ) def get_cost_run_rate_hints( @@ -227,13 +235,17 @@ class CapabilityService: def _build_accounts_payload(self, _: Mapping[str, Any]) -> dict[str, Any]: organization_payload = self.client.get_organization() organization = _extract_organization(organization_payload) - accounts = [_normalize_account(account) for account in _extract_accounts(organization_payload)] + accounts = [ + _normalize_account(account) for account in _extract_accounts(organization_payload) + ] return { "organization": _normalize_organization(organization), "accounts": accounts, "totals": { "balance": round(sum(account["balance"] for account in accounts), 2), - "authorized_balance": round(sum(account["authorized_balance"] for account in accounts), 2), + "authorized_balance": round( + sum(account["authorized_balance"] for account in accounts), 2 + ), }, } @@ -259,7 +271,9 @@ class CapabilityService: status=status, side=side, ) - transactions = [_normalize_transaction(item) for item in _extract_transactions(raw_transactions)] + transactions = [ + _normalize_transaction(item) for item in _extract_transactions(raw_transactions) + ] return { "organization": _normalize_organization(organization), "account": _normalize_account(selected_account), @@ -272,7 +286,9 @@ class CapabilityService: def _build_snapshot_payload(self, *, window_days: int, page_size: int) -> dict[str, Any]: accounts_payload = self._build_accounts_payload({}) accounts = accounts_payload["accounts"] - main_account = next((account for account in accounts if account["main"]), accounts[0] if accounts else None) + main_account = next( + (account for account in accounts if account["main"]), accounts[0] if accounts else None + ) recent_transactions = [] if main_account is not None: transactions_payload = self._build_transactions_payload( @@ -297,10 +313,14 @@ class CapabilityService: "recent_transactions": recent_transactions[:10], } - def _build_cost_run_rate_hints_payload(self, *, window_days: int, page_size: int) -> dict[str, Any]: + def _build_cost_run_rate_hints_payload( + self, *, window_days: int, page_size: int + ) -> dict[str, Any]: accounts_payload = self._build_accounts_payload({}) accounts = accounts_payload["accounts"] - main_account = next((account for account in accounts if account["main"]), accounts[0] if accounts else None) + main_account = next( + (account for account in accounts if account["main"]), accounts[0] if accounts else None + ) recent_transactions: list[dict[str, Any]] = [] if main_account is not None: transactions_payload = self._build_transactions_payload( @@ -343,6 +363,8 @@ class CapabilityService: deny_reason=deny_reason, policy_version=self.policy.version, latency_ms=latency_ms, + identity_binding=claims.identity_binding, + egress_destination="qonto-thirdparty-api", qonto_http_status=qonto_http_status, result_count=result_count, ) @@ -429,7 +451,9 @@ def _amount_value(raw: Mapping[str, Any], key: str = "balance") -> float: return round(float(amount), 2) -def _select_account(accounts: list[Mapping[str, Any]], account_slug: str | None) -> Mapping[str, Any]: +def _select_account( + accounts: list[Mapping[str, Any]], account_slug: str | None +) -> Mapping[str, Any]: if not accounts: raise UpstreamError( "No accounts available in organization payload", @@ -440,7 +464,9 @@ def _select_account(accounts: list[Mapping[str, Any]], account_slug: str | None) for account in accounts: if account.get("slug") == account_slug: return account - raise InvalidRequestError(f"Unknown account slug: {account_slug}", error_code="resource_scope") + raise InvalidRequestError( + f"Unknown account slug: {account_slug}", error_code="resource_scope" + ) for account in accounts: if account.get("main"): return account @@ -485,11 +511,19 @@ def _build_cost_run_rate_hints(transactions: list[dict[str, Any]]) -> dict[str, recurring_debits.sort(key=lambda item: (-item["amount"], item["label"])) total_debits = round( - sum(float(transaction["amount"]) for transaction in transactions if transaction.get("side") == "debit"), + sum( + float(transaction["amount"]) + for transaction in transactions + if transaction.get("side") == "debit" + ), 2, ) total_credits = round( - sum(float(transaction["amount"]) for transaction in transactions if transaction.get("side") == "credit"), + sum( + float(transaction["amount"]) + for transaction in transactions + if transaction.get("side") == "credit" + ), 2, ) return { diff --git a/tests/test_api.py b/tests/test_api.py index 4d83d63..78ee7f4 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -1,7 +1,9 @@ from pathlib import Path import httpx +import pytest +from qonto_assistant.app import create_app from qonto_assistant.audit import AuditLogger from qonto_assistant.config import Settings from qonto_assistant.contracts import ActorClaims @@ -12,7 +14,9 @@ from qonto_assistant.qonto_client import QontoClient from qonto_assistant.rate_limits import ConcurrencyLimiter, RateLimiter from qonto_assistant.service import CapabilityService -POLICY_FILE = Path(__file__).resolve().parents[1] / "src" / "qonto_assistant" / "policy" / "qonto-v1.yaml" +POLICY_FILE = ( + Path(__file__).resolve().parents[1] / "src" / "qonto_assistant" / "policy" / "qonto-v1.yaml" +) def _settings() -> Settings: @@ -38,6 +42,7 @@ def _settings() -> Settings: deny_escalation_threshold=3, deny_escalation_window_seconds=60, deny_escalation_lockout_seconds=300, + audit_heartbeat_interval_seconds=86400, key_cape_jwks_url=None, key_cape_issuer="https://key-cape.netkingdom", key_cape_audience="qonto-assistant", @@ -227,3 +232,56 @@ def test_snapshot_contract_returns_cost_run_rate_hints_for_90_day_window(monkeyp assert payload["summary"]["total_balance"] == 2185.94 assert payload["cost_run_rate_hints"]["recurring_debits"][0]["label"] == "HUB31" assert any(event["capability"] == "snapshot_bundle" for event in events) + + +async def test_app_lifecycle_and_reconciliation_use_the_request_audit_stream(monkeypatch) -> None: + service, events = _service(monkeypatch) + app = create_app(settings=_settings(), service=service) + + async with app.router.lifespan_context(app): + assert events[-1]["event_class"] == "audit.heartbeat" + assert events[-1]["reason"] == "startup" + + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://test" + ) as client: + response = await client.get( + "/v1/audit/reconciliation", + headers={"X-Actor-ID": "observer", "X-Tenant-ID": "binky"}, + ) + + assert response.status_code == 200 + assert response.json()["last_stream_sequence"] == 1 + assert response.json()["source_transition_counts"] == { + "audit.allow": 0, + "audit.deny": 0, + } + assert len(events) == 1 + + assert events[-1]["event_class"] == "audit.heartbeat" + assert events[-1]["reason"] == "shutdown" + assert events[-1]["stream_sequence"] == 2 + + +async def test_client_closes_when_shutdown_heartbeat_fails(monkeypatch) -> None: + service, _ = _service(monkeypatch) + closed = [] + monkeypatch.setattr(service.client, "close", lambda: closed.append(True)) + + def failing_shutdown(payload): + if payload.get("reason") == "shutdown": + raise RuntimeError("sink unavailable") + + service.audit_logger.sink = failing_shutdown + app = create_app(settings=_settings(), service=service) + with pytest.raises(ExceptionGroup) as caught: + async with app.router.lifespan_context(app): + pass + assert caught.group_contains(RuntimeError, match="sink unavailable") + assert closed == [True] + + +def test_app_rejects_split_audit_streams(monkeypatch) -> None: + service, _ = _service(monkeypatch) + with pytest.raises(ValueError, match="share one audit_logger"): + create_app(settings=_settings(), service=service, audit_logger=AuditLogger()) diff --git a/tests/test_audit.py b/tests/test_audit.py index 928079a..419e50b 100644 --- a/tests/test_audit.py +++ b/tests/test_audit.py @@ -1,6 +1,10 @@ +import asyncio import logging +from concurrent.futures import ThreadPoolExecutor +from contextlib import suppress -from qonto_assistant.audit import AuditLogger, REDACTED +from qonto_assistant.app import _emit_audit_heartbeats +from qonto_assistant.audit import AUDIT_STREAM_ID, REDACTED, AuditLogger def test_audit_logger_redacts_secret_fields() -> None: @@ -23,3 +27,115 @@ def test_audit_logger_redacts_secret_fields() -> None: assert payload["nested"]["token"] == REDACTED assert "super-secret" not in str(events[0]) assert "top-secret" not in str(events[0]) + + +def test_audit_logger_sequences_events_and_counts_source_transitions() -> None: + events: list[dict[str, object]] = [] + audit = AuditLogger(sink=events.append, instance_id="instance-1") + + allowed = audit.emit({"decision": "allow", "request_id": "allow-1"}) + denied = audit.emit({"decision": "deny", "request_id": "deny-1"}) + snapshot = audit.reconciliation_snapshot() + + assert allowed["event_class"] == "audit.allow" + assert denied["event_class"] == "audit.deny" + assert [event["stream_sequence"] for event in events] == [1, 2] + assert all(event["stream_id"] == AUDIT_STREAM_ID for event in events) + assert all(event["stream_instance_id"] == "instance-1" for event in events) + assert snapshot["last_stream_sequence"] == 2 + assert snapshot["source_transition_counts"] == {"audit.allow": 1, "audit.deny": 1} + assert snapshot["since_last_heartbeat_counts"] == {"audit.allow": 1, "audit.deny": 1} + + +def test_heartbeat_reconciles_window_and_resets_only_window_counts() -> None: + timestamps = iter( + [ + "2026-09-04T00:00:00+00:00", + "2026-09-04T00:01:00+00:00", + "2026-09-04T00:02:00+00:00", + "2026-09-04T00:03:00+00:00", + ] + ) + events: list[dict[str, object]] = [] + audit = AuditLogger( + sink=events.append, instance_id="instance-1", clock=lambda: next(timestamps) + ) + + first = audit.emit_heartbeat(reason="startup") + audit.emit({"decision": "deny", "request_id": "deny-1"}) + second = audit.emit_heartbeat(reason="periodic") + snapshot = audit.reconciliation_snapshot() + + assert first["assertion"] == "nothing-to-report" + assert first["window_transition_counts"] == {"audit.allow": 0, "audit.deny": 0} + assert second["assertion"] == "transitions-reported" + assert second["source_transition_counts"] == {"audit.allow": 0, "audit.deny": 1} + assert second["window_transition_counts"] == {"audit.allow": 0, "audit.deny": 1} + assert second["stream_sequence"] == 3 + assert snapshot["source_transition_counts"] == {"audit.allow": 0, "audit.deny": 1} + assert snapshot["since_last_heartbeat_counts"] == {"audit.allow": 0, "audit.deny": 0} + + +def test_primary_log_is_written_before_optional_sink_failure(caplog) -> None: + def broken_sink(_: dict[str, object]) -> None: + raise RuntimeError("secondary sink unavailable") + + audit = AuditLogger(sink=broken_sink, instance_id="instance-1") + + with caplog.at_level(logging.INFO, logger="qonto_assistant.audit"): + try: + audit.emit({"decision": "deny", "request_id": "deny-1"}) + except RuntimeError: + pass + else: + raise AssertionError("Expected the secondary sink error") + + assert '"event_class": "audit.deny"' in caplog.text + assert '"stream_sequence": 1' in caplog.text + + +async def test_periodic_heartbeat_loop_emits_until_cancelled() -> None: + events: list[dict[str, object]] = [] + audit = AuditLogger(sink=events.append, instance_id="instance-1") + ready = asyncio.Event() + + def collect(payload): + events.append(payload) + if len(events) >= 2: + ready.set() + + audit.sink = collect + task = asyncio.create_task(_emit_audit_heartbeats(audit_logger=audit, interval_seconds=0.01)) + try: + await asyncio.wait_for(ready.wait(), timeout=2) + finally: + task.cancel() + with suppress(asyncio.CancelledError): + await task + + assert len(events) >= 2 + assert all(event["event_class"] == "audit.heartbeat" for event in events) + assert all(event["reason"] == "periodic" for event in events) + + +def test_concurrent_transitions_and_heartbeats_preserve_publication_order() -> None: + events = [] + audit = AuditLogger(sink=events.append) + + def publish(index): + if index % 3 == 0: + audit.emit_heartbeat() + else: + audit.emit({"decision": "deny"}) + + with ThreadPoolExecutor(max_workers=8) as executor: + list(executor.map(publish, range(120))) + + assert [event["stream_sequence"] for event in events] == list(range(1, 121)) + denies = 0 + for event in events: + if event["event_class"] == "audit.deny": + denies += 1 + else: + assert event["source_transition_counts"]["audit.deny"] == denies + assert audit.reconciliation_snapshot()["source_transition_counts"]["audit.deny"] == 80 diff --git a/tests/test_audit_parity.py b/tests/test_audit_parity.py index 9341eba..d3a157a 100644 --- a/tests/test_audit_parity.py +++ b/tests/test_audit_parity.py @@ -18,18 +18,35 @@ from qonto_assistant.rate_limits import ConcurrencyLimiter, RateLimiter from qonto_assistant.service import CapabilityService FIXTURE_DIR = Path(__file__).resolve().parent / "fixtures" / "qonto" -POLICY_FILE = Path(__file__).resolve().parents[1] / "src" / "qonto_assistant" / "policy" / "qonto-v1.yaml" +POLICY_FILE = ( + Path(__file__).resolve().parents[1] / "src" / "qonto_assistant" / "policy" / "qonto-v1.yaml" +) # Fields whose values are expected to vary per call (timing/identifiers) or # by design (protocol). Everything else must match exactly between REST and # MCP for the same logical call. -NON_COMPARABLE_FIELDS = {"request_id", "timestamp", "latency_ms", "protocol"} -SECRET_FIELD_NAMES = {"api_key", "authorization", "authorization_header", "openbao_token", "secret", "token"} +NON_COMPARABLE_FIELDS = { + "request_id", + "timestamp", + "latency_ms", + "protocol", + "stream_sequence", +} +SECRET_FIELD_NAMES = { + "api_key", + "authorization", + "authorization_header", + "openbao_token", + "secret", + "token", +} def _service() -> tuple[CapabilityService, list[dict[str, object]]]: events: list[dict[str, object]] = [] - policy = PolicyEngine.from_file(POLICY_FILE, required_scope="finance.qonto.read", enforce_scope=False) + policy = PolicyEngine.from_file( + POLICY_FILE, required_scope="finance.qonto.read", enforce_scope=False + ) service = CapabilityService( client=FixtureQontoClient(fixture_dir=FIXTURE_DIR), policy=policy, @@ -56,10 +73,14 @@ def test_allow_path_audit_schema_identical_across_protocols() -> None: assert rest_event["protocol"] == "rest" assert mcp_event["protocol"] == "mcp" for field in set(rest_event) - NON_COMPARABLE_FIELDS: - assert rest_event[field] == mcp_event[field], f"field {field!r} diverged: {rest_event[field]!r} != {mcp_event[field]!r}" + assert rest_event[field] == mcp_event[field], ( + f"field {field!r} diverged: {rest_event[field]!r} != {mcp_event[field]!r}" + ) assert rest_event["decision"] == "allow" assert rest_event["capability"] == "org_summary" + assert rest_event["identity_binding"] == "self_asserted" + assert rest_event["egress_destination"] == "qonto-thirdparty-api" def test_deny_path_audit_schema_identical_across_protocols() -> None: diff --git a/tests/test_deny_escalation.py b/tests/test_deny_escalation.py index 3cbeed4..1ba6d9d 100644 --- a/tests/test_deny_escalation.py +++ b/tests/test_deny_escalation.py @@ -14,7 +14,9 @@ from qonto_assistant.security_watch import ( ) from qonto_assistant.service import CapabilityService -POLICY_FILE = Path(__file__).resolve().parents[1] / "src" / "qonto_assistant" / "policy" / "qonto-v1.yaml" +POLICY_FILE = ( + Path(__file__).resolve().parents[1] / "src" / "qonto_assistant" / "policy" / "qonto-v1.yaml" +) FIXTURE_DIR = Path(__file__).resolve().parent / "fixtures" / "qonto" @@ -53,7 +55,9 @@ def _service(tracker: DenyEscalationTracker, events: list[dict[str, object]]) -> def test_tracker_allows_denies_below_threshold() -> None: clock = _FakeClock() - tracker = DenyEscalationTracker(threshold=3, window_seconds=60, lockout_seconds=300, clock=clock) + tracker = DenyEscalationTracker( + threshold=3, window_seconds=60, lockout_seconds=300, clock=clock + ) tracker.record_deny("binky:prober", "arg_constraint") tracker.record_deny("binky:prober", "arg_constraint") @@ -63,7 +67,9 @@ def test_tracker_allows_denies_below_threshold() -> None: def test_tracker_locks_out_after_threshold_within_window() -> None: clock = _FakeClock() - tracker = DenyEscalationTracker(threshold=3, window_seconds=60, lockout_seconds=300, clock=clock) + tracker = DenyEscalationTracker( + threshold=3, window_seconds=60, lockout_seconds=300, clock=clock + ) for _ in range(3): tracker.record_deny("binky:prober", "arg_constraint") @@ -74,7 +80,9 @@ def test_tracker_locks_out_after_threshold_within_window() -> None: def test_tracker_ignores_non_escalating_deny_reasons() -> None: clock = _FakeClock() - tracker = DenyEscalationTracker(threshold=3, window_seconds=60, lockout_seconds=300, clock=clock) + tracker = DenyEscalationTracker( + threshold=3, window_seconds=60, lockout_seconds=300, clock=clock + ) for _ in range(10): tracker.record_deny("binky:prober", "tenant_scope") @@ -85,7 +93,9 @@ def test_tracker_ignores_non_escalating_deny_reasons() -> None: def test_tracker_denies_outside_window_do_not_accumulate() -> None: clock = _FakeClock() - tracker = DenyEscalationTracker(threshold=3, window_seconds=60, lockout_seconds=300, clock=clock) + tracker = DenyEscalationTracker( + threshold=3, window_seconds=60, lockout_seconds=300, clock=clock + ) tracker.record_deny("binky:prober", "arg_constraint") clock.advance(61) @@ -98,7 +108,9 @@ def test_tracker_denies_outside_window_do_not_accumulate() -> None: def test_tracker_lockout_expires_after_lockout_window() -> None: clock = _FakeClock() - tracker = DenyEscalationTracker(threshold=3, window_seconds=60, lockout_seconds=300, clock=clock) + tracker = DenyEscalationTracker( + threshold=3, window_seconds=60, lockout_seconds=300, clock=clock + ) for _ in range(3): tracker.record_deny("binky:prober", "arg_constraint") @@ -118,7 +130,9 @@ def test_tracker_lockout_expires_after_lockout_window() -> None: def test_tracker_is_scoped_per_actor() -> None: clock = _FakeClock() - tracker = DenyEscalationTracker(threshold=3, window_seconds=60, lockout_seconds=300, clock=clock) + tracker = DenyEscalationTracker( + threshold=3, window_seconds=60, lockout_seconds=300, clock=clock + ) for _ in range(3): tracker.record_deny("binky:prober", "arg_constraint") @@ -131,7 +145,9 @@ def test_tracker_is_scoped_per_actor() -> None: def test_service_locks_out_actor_after_repeated_arg_constraint_denies() -> None: clock = _FakeClock() - tracker = DenyEscalationTracker(threshold=2, window_seconds=60, lockout_seconds=300, clock=clock) + tracker = DenyEscalationTracker( + threshold=2, window_seconds=60, lockout_seconds=300, clock=clock + ) events: list[dict[str, object]] = [] service = _service(tracker, events) @@ -164,7 +180,9 @@ def test_service_locks_out_actor_after_repeated_arg_constraint_denies() -> None: def test_service_does_not_lock_out_for_ordinary_denies() -> None: clock = _FakeClock() - tracker = DenyEscalationTracker(threshold=2, window_seconds=60, lockout_seconds=300, clock=clock) + tracker = DenyEscalationTracker( + threshold=2, window_seconds=60, lockout_seconds=300, clock=clock + ) events: list[dict[str, object]] = [] service = _service(tracker, events) diff --git a/tests/test_flex_auth_client.py b/tests/test_flex_auth_client.py index 0b293cb..8bb993c 100644 --- a/tests/test_flex_auth_client.py +++ b/tests/test_flex_auth_client.py @@ -25,7 +25,10 @@ def _client(handler) -> FlexAuthCheckClient: def test_allow_effect_authorizes() -> None: def handler(request: httpx.Request) -> httpx.Response: - return httpx.Response(200, json={"id": "d-1", "effect": "allow", "resource": {}, "subject": {}, "provenance": {}}) + return httpx.Response( + 200, + json={"id": "d-1", "effect": "allow", "resource": {}, "subject": {}, "provenance": {}}, + ) assert _client(handler).is_allowed(_request()) is True @@ -33,7 +36,10 @@ def test_allow_effect_authorizes() -> None: @pytest.mark.parametrize("effect", ["deny", "redact", "audit_only", "not_applicable"]) def test_non_allow_effects_deny(effect: str) -> None: def handler(request: httpx.Request) -> httpx.Response: - return httpx.Response(200, json={"id": "d-1", "effect": effect, "resource": {}, "subject": {}, "provenance": {}}) + return httpx.Response( + 200, + json={"id": "d-1", "effect": effect, "resource": {}, "subject": {}, "provenance": {}}, + ) assert _client(handler).is_allowed(_request()) is False @@ -78,7 +84,10 @@ def test_request_body_matches_schema_shape() -> None: def handler(request: httpx.Request) -> httpx.Response: seen.update(json.loads(request.content)) - return httpx.Response(200, json={"id": "d-1", "effect": "allow", "resource": {}, "subject": {}, "provenance": {}}) + return httpx.Response( + 200, + json={"id": "d-1", "effect": "allow", "resource": {}, "subject": {}, "provenance": {}}, + ) _client(handler).is_allowed(_request()) diff --git a/tests/test_key_cape_auth.py b/tests/test_key_cape_auth.py index 2f6f1b0..9278a32 100644 --- a/tests/test_key_cape_auth.py +++ b/tests/test_key_cape_auth.py @@ -53,7 +53,9 @@ def _token(private_key: rsa.RSAPrivateKey, *, kid: str = KID, **claim_overrides) return jwt.encode(claims, private_key, algorithm="RS256", headers={"kid": kid}) -def _verifier(jwk: dict, *, required: bool = False, calls: list[int] | None = None) -> KeyCapeTokenVerifier: +def _verifier( + jwk: dict, *, required: bool = False, calls: list[int] | None = None +) -> KeyCapeTokenVerifier: return KeyCapeTokenVerifier( jwks_url="https://key-cape.netkingdom.test/jwks", issuer=ISSUER, @@ -71,6 +73,7 @@ def test_verify_accepts_valid_token() -> None: assert claims.actor_id == "agent-harness-binky" assert claims.tenant_id == "tenant:friendly:binky" + assert claims.identity_binding == "key_cape_jwt" assert claims.lane == "blue" assert "finance.qonto.read" in claims.scopes @@ -194,6 +197,7 @@ def _settings() -> Settings: deny_escalation_threshold=3, deny_escalation_window_seconds=60, deny_escalation_lockout_seconds=300, + audit_heartbeat_interval_seconds=86400, key_cape_jwks_url=None, key_cape_issuer=ISSUER, key_cape_audience=AUDIENCE, @@ -236,7 +240,9 @@ def test_required_verifier_rejects_missing_bearer_token() -> None: verifier = _verifier(jwk, required=True) with pytest.raises(KeyCapeAuthError): - actor_claims_from_headers({"x-actor-id": "someone"}, _settings(), key_cape_verifier=verifier) + actor_claims_from_headers( + {"x-actor-id": "someone"}, _settings(), key_cape_verifier=verifier + ) def test_optional_verifier_falls_back_to_self_asserted_headers_when_absent() -> None: @@ -249,6 +255,7 @@ def test_optional_verifier_falls_back_to_self_asserted_headers_when_absent() -> assert claims.actor_id == "someone" assert claims.tenant_id == "binky" + assert claims.identity_binding == "self_asserted" def test_invalid_bearer_token_is_rejected_even_when_not_required() -> None: diff --git a/tests/test_live_authorization_gate.py b/tests/test_live_authorization_gate.py index cace6db..a161cce 100644 --- a/tests/test_live_authorization_gate.py +++ b/tests/test_live_authorization_gate.py @@ -7,14 +7,20 @@ from qonto_assistant.audit import AuditLogger from qonto_assistant.contracts import ActorClaims from qonto_assistant.errors import PolicyDeniedError from qonto_assistant.flex_auth_client import FlexAuthCheckClient -from qonto_assistant.live_authorization import DENY_LIVE_AUTHZ, DENY_TENANT_ROLE, LiveAuthorizationGate +from qonto_assistant.live_authorization import ( + DENY_LIVE_AUTHZ, + DENY_TENANT_ROLE, + LiveAuthorizationGate, +) from qonto_assistant.policy import PolicyEngine from qonto_assistant.qonto_client import FixtureQontoClient from qonto_assistant.rate_limits import ConcurrencyLimiter, RateLimiter from qonto_assistant.service import CapabilityService from qonto_assistant.tenant_engine_client import TenantEngineClient -POLICY_FILE = Path(__file__).resolve().parents[1] / "src" / "qonto_assistant" / "policy" / "qonto-v1.yaml" +POLICY_FILE = ( + Path(__file__).resolve().parents[1] / "src" / "qonto_assistant" / "policy" / "qonto-v1.yaml" +) FIXTURE_DIR = Path(__file__).resolve().parent / "fixtures" / "qonto" @@ -24,16 +30,23 @@ def _claims(tenant_id: str = "tenant:friendly:binky") -> ActorClaims: def _flex_auth_client(effect: str) -> FlexAuthCheckClient: def handler(request: httpx.Request) -> httpx.Response: - return httpx.Response(200, json={"id": "d-1", "effect": effect, "resource": {}, "subject": {}, "provenance": {}}) + return httpx.Response( + 200, + json={"id": "d-1", "effect": effect, "resource": {}, "subject": {}, "provenance": {}}, + ) - return FlexAuthCheckClient(base_url="https://flex-auth.test", transport=httpx.MockTransport(handler)) + return FlexAuthCheckClient( + base_url="https://flex-auth.test", transport=httpx.MockTransport(handler) + ) def _tenant_engine_client(roles: list[str]) -> TenantEngineClient: def handler(request: httpx.Request) -> httpx.Response: return httpx.Response(200, json={"tenant_id": "tenant:friendly:binky", "roles": roles}) - return TenantEngineClient(base_url="https://tenant-engine.test", transport=httpx.MockTransport(handler)) + return TenantEngineClient( + base_url="https://tenant-engine.test", transport=httpx.MockTransport(handler) + ) # --- Gate unit tests ------------------------------------------------------------------- @@ -95,7 +108,9 @@ def test_gate_denies_when_tenant_engine_unreachable() -> None: gate = LiveAuthorizationGate( flex_auth_client=_flex_auth_client("allow"), - tenant_engine_client=TenantEngineClient(base_url="https://tenant-engine.test", transport=httpx.MockTransport(handler)), + tenant_engine_client=TenantEngineClient( + base_url="https://tenant-engine.test", transport=httpx.MockTransport(handler) + ), required_tenant_roles=frozenset({"VEN"}), ) @@ -108,7 +123,9 @@ def test_gate_denies_when_tenant_engine_unreachable() -> None: def _service(gate: LiveAuthorizationGate, events: list[dict[str, object]]) -> CapabilityService: return CapabilityService( client=FixtureQontoClient(fixture_dir=FIXTURE_DIR), - policy=PolicyEngine.from_file(POLICY_FILE, required_scope="finance.qonto.read", enforce_scope=False), + policy=PolicyEngine.from_file( + POLICY_FILE, required_scope="finance.qonto.read", enforce_scope=False + ), audit_logger=AuditLogger(sink=events.append), rate_limiter=RateLimiter(limit=20, window_seconds=60), concurrency_limiter=ConcurrencyLimiter(limit=4), @@ -151,7 +168,9 @@ def test_service_without_gate_configured_behaves_as_before() -> None: events: list[dict[str, object]] = [] service = CapabilityService( client=FixtureQontoClient(fixture_dir=FIXTURE_DIR), - policy=PolicyEngine.from_file(POLICY_FILE, required_scope="finance.qonto.read", enforce_scope=False), + policy=PolicyEngine.from_file( + POLICY_FILE, required_scope="finance.qonto.read", enforce_scope=False + ), audit_logger=AuditLogger(sink=events.append), rate_limiter=RateLimiter(limit=20, window_seconds=60), concurrency_limiter=ConcurrencyLimiter(limit=4), diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 8485f88..b29a66c 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -12,7 +12,9 @@ from qonto_assistant.rate_limits import ConcurrencyLimiter, RateLimiter from qonto_assistant.service import CapabilityService FIXTURE_DIR = Path(__file__).resolve().parent / "fixtures" / "qonto" -POLICY_FILE = Path(__file__).resolve().parents[1] / "src" / "qonto_assistant" / "policy" / "qonto-v1.yaml" +POLICY_FILE = ( + Path(__file__).resolve().parents[1] / "src" / "qonto_assistant" / "policy" / "qonto-v1.yaml" +) def _settings() -> Settings: @@ -21,7 +23,9 @@ def _settings() -> Settings: def _service() -> tuple[CapabilityService, list[dict[str, object]]]: events: list[dict[str, object]] = [] - policy = PolicyEngine.from_file(POLICY_FILE, required_scope="finance.qonto.read", enforce_scope=False) + policy = PolicyEngine.from_file( + POLICY_FILE, required_scope="finance.qonto.read", enforce_scope=False + ) service = CapabilityService( client=FixtureQontoClient(fixture_dir=FIXTURE_DIR), policy=policy, diff --git a/tests/test_policy.py b/tests/test_policy.py index d8068b9..84a8b3b 100644 --- a/tests/test_policy.py +++ b/tests/test_policy.py @@ -5,7 +5,9 @@ import pytest from qonto_assistant.contracts import ActorClaims, CapabilityRequest from qonto_assistant.policy import PolicyEngine -POLICY_FILE = Path(__file__).resolve().parents[1] / "src" / "qonto_assistant" / "policy" / "qonto-v1.yaml" +POLICY_FILE = ( + Path(__file__).resolve().parents[1] / "src" / "qonto_assistant" / "policy" / "qonto-v1.yaml" +) def _policy(*, enforce_scope: bool = False) -> PolicyEngine: @@ -55,21 +57,31 @@ def test_policy_denies_cross_tenant_requests() -> None: def test_policy_denies_excessive_page_size() -> None: - decision = _policy().decide(_request("list_transactions", page=1, page_size=101, window_days=31)) + decision = _policy().decide( + _request("list_transactions", page=1, page_size=101, window_days=31) + ) assert decision.allowed is False assert decision.reason == "arg_constraint" def test_policy_denies_volume_cost_shaped_requests() -> None: decision = _policy().decide( - _request("list_transactions", page=1, page_size=50, window_days=31, operation_type="card_operation") + _request( + "list_transactions", + page=1, + page_size=50, + window_days=31, + operation_type="card_operation", + ) ) assert decision.allowed is False assert decision.reason == "volume_cost" def test_policy_denies_credential_exfiltration_flags() -> None: - decision = _policy().decide(_request("list_transactions", page=1, page_size=50, window_days=31, include_full_iban=True)) + decision = _policy().decide( + _request("list_transactions", page=1, page_size=50, window_days=31, include_full_iban=True) + ) assert decision.allowed is False assert decision.reason == "credential_exfil" diff --git a/tests/test_snapshot_semantics.py b/tests/test_snapshot_semantics.py index 063bdb3..3f66a6c 100644 --- a/tests/test_snapshot_semantics.py +++ b/tests/test_snapshot_semantics.py @@ -7,7 +7,9 @@ from qonto_assistant.qonto_client import FixtureQontoClient from qonto_assistant.rate_limits import ConcurrencyLimiter, RateLimiter from qonto_assistant.service import CapabilityService -POLICY_FILE = Path(__file__).resolve().parents[1] / "src" / "qonto_assistant" / "policy" / "qonto-v1.yaml" +POLICY_FILE = ( + Path(__file__).resolve().parents[1] / "src" / "qonto_assistant" / "policy" / "qonto-v1.yaml" +) FIXTURE_DIR = Path(__file__).resolve().parent / "fixtures" / "qonto" diff --git a/tests/test_tenant_engine_client.py b/tests/test_tenant_engine_client.py index 8811f39..6268a80 100644 --- a/tests/test_tenant_engine_client.py +++ b/tests/test_tenant_engine_client.py @@ -14,7 +14,9 @@ def _client(handler) -> TenantEngineClient: def test_active_roles_returns_roles_on_200() -> None: def handler(request: httpx.Request) -> httpx.Response: assert request.url.path == "/tenants/tenant:friendly:binky/roles/live" - return httpx.Response(200, json={"tenant_id": "tenant:friendly:binky", "roles": ["VEN", "CUS"]}) + return httpx.Response( + 200, json={"tenant_id": "tenant:friendly:binky", "roles": ["VEN", "CUS"]} + ) roles = _client(handler).active_roles("tenant:friendly:binky") diff --git a/workplans/ADHOC-2026-09-05.md b/workplans/ADHOC-2026-09-05.md new file mode 100644 index 0000000..17a8d81 --- /dev/null +++ b/workplans/ADHOC-2026-09-05.md @@ -0,0 +1,25 @@ +--- +id: ADHOC-2026-09-05 +type: workplan +title: "Clear verification lint failures and stale scope status" +domain: infotech +repo: qonto-assistant +status: finished +owner: codex +created: "2026-09-05" +updated: "2026-09-05" +--- + +## Restore clean static checks + +```task +id: ADHOC-2026-09-05-T01 +status: done +priority: low +``` + +Applied Ruff import and formatting corrections to existing source/tests and +shortened one docstring to clear the 44 repository lint failures. Corrected +SCOPE.md's stale runtime-not-started entry. Full Ruff checks and the 88-test +suite pass using the existing venv directly. Make bootstrap remains dependent +on pip, which this workstation's existing venv does not contain. diff --git a/workplans/QONTO-WP-0005-audit-deny-stream-completeness.md b/workplans/QONTO-WP-0005-audit-deny-stream-completeness.md new file mode 100644 index 0000000..1803258 --- /dev/null +++ b/workplans/QONTO-WP-0005-audit-deny-stream-completeness.md @@ -0,0 +1,107 @@ +--- +id: QONTO-WP-0005 +type: workplan +title: "Establish audit.deny stream completeness evidence" +domain: infotech +repo: qonto-assistant +status: finished +owner: codex +topic_slug: qonto-audit-deny-stream-completeness +created: "2026-09-04" +updated: "2026-09-05" +state_hub_workstream_id: "8a1bc02c-00fc-5400-af1f-10af562b9782" +--- + +# Establish `audit.deny` stream completeness evidence + +Closes the qonto-assistant-owned remediation from `RISK-F-0011` and the +King's Guard message received on 2026-09-01. A received deny event is already +useful, but observers cannot currently distinguish a quiet stream from an +incomplete one. The source therefore needs to publish its cadence, emit a +positive heartbeat carrying source-side counts, and expose a reconciliation +snapshot that makes missing events and sequence gaps visible. + +The implementation stays local to the structured audit stream. State Hub is +not placed on the request hot path, and the existing in-process deny escalation +loop remains independent from estate observation. + +## Task: Declare cadence and reconciliation semantics + +```task +id: QONTO-WP-0005-T01 +status: done +priority: high +state_hub_task_id: "e846d42b-d9e0-587e-874b-0ad8c1d5c471" +``` + +Publish the `audit.deny` load-bearing classification next to the security +genome. Declare heartbeat behavior, active-instance lifecycle semantics, +source-side counters, sequence-gap detection, and what an observer must compare. + +Done when the declaration is machine-readable and the operator documentation +does not imply that a received event alone proves stream completeness. + +**Done 2026-09-04:** Added `specs/audit-emission-cadence.yaml` and linked it +from the security genome. It declares the per-process sequence boundary, +`audit.deny` as low-volume/load-bearing, the 24-hour active-process cadence, +scale-to-zero lifecycle semantics, cumulative/window counters, and observer +comparison rules. The architecture, security practice, and operator runbook +now carry the same semantics. + +## Task: Emit heartbeat and reconciliation evidence + +```task +id: QONTO-WP-0005-T02 +status: done +priority: high +state_hub_task_id: "cdb6576a-f606-57c9-94b2-923f23c2044b" +``` + +Add process-instance identity, monotonic stream sequence, per-class transition +counts, periodic/startup/shutdown heartbeat events, and a read-only local +reconciliation view. Enrich request events with their identity binding and +declared egress destination so observers no longer substitute genome constants. + +Done when the source emits enough information to detect a missing deny or +heartbeat without introducing State Hub or another network dependency. + +**Done 2026-09-04:** `AuditLogger` now serializes publication under one lock, +assigns a UUID process instance and monotonic sequence, counts allow/deny +source transitions, and emits startup/periodic/best-effort-shutdown heartbeats +with reconciliation counters. `GET /v1/audit/reconciliation` returns the same +non-secret local state without incrementing it. REST/MCP request records carry +`identity_binding` (`self_asserted` or `key_cape_jwt`) and +`egress_destination: qonto-thirdparty-api`. The app rejects split injected +audit loggers so request events and heartbeat evidence cannot silently land in +different streams. + +## Task: Verify and hand back the finding + +```task +id: QONTO-WP-0005-T03 +status: done +priority: high +state_hub_task_id: "e5ed80bd-f0aa-5473-a185-ecb9f87ab21c" +``` + +Cover sequencing, counts, heartbeat windows, reconciliation output, REST/MCP +parity, identity binding, and secret redaction in tests. Run the full local +suite, sync the workplan to State Hub, and notify Risk Nexus and King's Guard +with the source-owned evidence and any remaining deployment observation gate. + +**Done 2026-09-04:** Full suite passed (`85 passed`); changed-file Ruff checks, +YAML parsing, `compileall`, `git diff --check`, REST smoke, and MCP smoke all +passed. Both smoke runs visibly emitted startup/shutdown heartbeats with +contiguous sequences and matching counts. Source implementation is complete; +deployment and independent observer acceptance remain with the runtime owner, +King's Guard, and Risk Nexus rather than as unfinished source work here. + + +**Review 2026-09-05:** Reviewed the pending implementation against its cadence +contract. Fixed client cleanup when shutdown audit publication fails, added +regressions for cleanup and split stream rejection, and verified concurrent +heartbeat/request publication ordering. Replaced timing-sensitive periodic test +sleep with an explicit event. Final verification: 88 tests passed, full-source +Ruff, compileall, YAML parsing, and REST/MCP fixture smoke checks passed. Runtime +deployment and independent observer acceptance remain external gates. No new +cross-repo messages were sent during this review.