diff --git a/credential-change-requests/CCR-2026-0047-rapp-qonto-audit-sender.yaml b/credential-change-requests/CCR-2026-0047-rapp-qonto-audit-sender.yaml new file mode 100644 index 0000000..13b53cd --- /dev/null +++ b/credential-change-requests/CCR-2026-0047-rapp-qonto-audit-sender.yaml @@ -0,0 +1,108 @@ +id: CCR-2026-0047 +kind: credential-change-request +schema_version: 1 +request_type: workload-kv-read +title: rapp-qonto collector exact audit sender custody and delivery +status: proposed +created: '2026-10-15' +updated: '2026-10-15' +requester: + agent: codex + reason: NK-WP-0035 operational receiver admission requires actual Qonto delivery. This proposed new + grant requires explicit owner approval of the concrete packet; prior timing rollout approval does + not approve it. +review: + required: true + required_approvers: + - platform-operator + - audit-core-owner + - rapp-qonto-owner + - qonto-assistant-owner + comments: [] +target: + domain: financials + tenant: friendly:binky + workload: rapp-qonto + environment: production + purpose: Sender rapp-qonto-collector, sole source qonto-assistant, sole tenant tenant:friendly:binky, + may_write=true, may_read=false, secret_policy=redact. Envelope admission load-bearing; preserve + the source declaration per-class distinction. One source heartbeat maps to audit-core.heartbeat + covering only audit.deny. Wait 2f=7200s, effective deadline3f=10800s. No bank calls or authority + changes. +openbao: + mount: platform + kv_path: platform/workloads/rapp-qonto/audit-sender + fields: + - AUDIT_TOKEN + - CUSTODY_REQUEST + policy_name: workload-kv-read-rapp-qonto-audit + policy_file: openbao/policies/workload-kv-read-rapp-qonto-audit.hcl + metadata_read: false + auth: + method: kubernetes + mount: kubernetes + role: external-secrets-rapp-qonto-audit + bound_claims: + service_account_names: + - external-secrets + service_account_namespaces: + - external-secrets + bound_claims_confirmed: true + policies: + - workload-kv-read-rapp-qonto-audit + ttl: 15m +access_frontdoor: + type: external-secrets + catalog_id: rapp-qonto-audit + readiness: pending-review + resolvable: false +delivery: + surface: external-secrets + target: ClusterSecretStore openbao-rapp-qonto-audit restricted to namespace rapp-qonto; ExternalSecret + and Secret rapp-qonto/rapp-qonto-audit-sender key bearer. Source manifests/rapp-qonto-audit-sender.yaml. + No namespace is created. +risk: + classification: high + notes: + - A bearer permits append only for exact collector sender/source/tenant. Own-source reconciliation + is available as receiver contract, cross-source and other read routes remain refused. + - The registry update reads existing sender credentials inside the attended platform process only. + No registry is delivered to a producer. + - OpenBao login TTL limits the ESO reader session, not the audit bearer. Revocation must remove the + sender token from the receiver registry and prove refusal. + - New paths extend the existing coding-agent deny boundary. No existing high-risk grant is widened. +verification: + positive: + - CAS=0 first provision; durable provenance-bound version1 resume without rotation; exact sender registry + CAS merge preserves all existing credentials/senders/other fields. + - ESO copies only AUDIT_TOKEN to the named bearer key, exact source/delivery readback, receiver registry + delivery and compatible reload. + - Actual capture, periodic real elapsed frame, per-instance reconciliation and independent observer + acceptance remain required after provisioning. + negative: + - Proposed or missing owner approval refuses before native calls; drifted custody, sender or source + hashes refuse. + - Reader cannot read registry, sibling custody, metadata or parent listing; wrong service account/namespace + refused; coding-agent boundary deny wins. + - Wrong sender source/tenant and non-own-source read routes refused. Audit bearer revocation is separate + from reader-session revoke. + activation_conditions: + - Explicit approval of this complete concrete packet from platform-operator, audit-core-owner, rapp-qonto-owner + and qonto-assistant-owner. + - Compatible immutable receiver release with cadence parser and per-instance reconciliation is deployed/ready + before credential reads. + - Use attended Warden platform-admin --exec envelope, no shell-exported credentials; keep nonsecret + independent receipt and self-revoke. + - Apply only reviewed exact policy/role/projection; retain custody on failure and resume explicitly; + no destructive rollback or blind rotation. + - No claim of operational completion until native positive/negative and independent observer receipts. + evidence: [] +lifecycle: + deactivate: Stop only collector; remove only its bearer from receiver via CAS and prove refusal; detach + exact reader role and ESO projection; retain custody versions and evidence. + rotate: Separate reviewed overlap-first change; this helper only provisions version1 and resumes without + rotation. + compromised: Stop collector and revoke its exact receiver bearer first; preserve evidence and perform + separate reviewed rotation. +state_hub: + related_workplan: NK-WP-0035 diff --git a/docs/cadence-receiver-custody.md b/docs/cadence-receiver-custody.md new file mode 100644 index 0000000..8450611 --- /dev/null +++ b/docs/cadence-receiver-custody.md @@ -0,0 +1,95 @@ +# Hourly cadence receiver admission and Qonto sender custody + +NK-WP-0035 / RPF-WP-0059 / CCR-2026-0047. This is a proposed production +packet. The founder's timing-rollout approval did not approve this new bearer +or ESO reader. Obtain explicit platform operator, Audit Core, rapp-qonto and +qonto-assistant owner review of the exact complete packet first. + +The prepared change preserves every existing sender credential, source/tenant +scope and other registry field. It updates only Approval Engine's canonical +cadence and four heartbeat waits (2f=7200s; receiver grace1.5 gives3f=10800s), +and adds one new exact sender: + +| Coordinate | Proposed value | +| --- | --- | +| Sender | rapp-qonto-collector | +| Source | qonto-assistant | +| Tenant | tenant:friendly:binky | +| Permission | write=true, read=false; receiver own-source reconciliation remains contract-defined | +| Evidence | envelope load-bearing/redact; complete source cadence retains each class's kind | +| Heartbeat | one source beat maps once to audit-core.heartbeat, data.class=audit.deny | +| Custody | platform/workloads/rapp-qonto/audit-sender: AUDIT_TOKEN,CUSTODY_REQUEST | +| Reader policy | workload-kv-read-rapp-qonto-audit; exact data path read only, no metadata/list | +| Kubernetes role | external-secrets-rapp-qonto-audit; external-secrets/external-secrets,15m | +| Delivery | namespace-restricted openbao-rapp-qonto-audit store; rapp-qonto/rapp-qonto-audit-sender:bearer | + +Plan without credentials or network: + +```bash +python3 scripts/credential-change.py validate CCR-2026-0047 +python3 scripts/cadence_receiver_custody.py plan +``` + +The plan hashes CCR, exact policy/projection, executable helpers and enforcing +receiver parser sources, raw source declarations and Approval metadata proposal. +Record owner approvals in the CCR, regenerate the plan, and review its resulting +SHA256. Any subsequent source change requires a newly reviewed packet. The +runner verifies the exact packet hash; a proposed CCR refuses before native +calls. The immutable deployed receiver must support full cadence parsing and +per-instance reconciliation, verified with fixture-only native sender parsing +before any credential read. A source compatibility test alone cannot satisfy +that deployed gate. + +Run `apply` only inside the existing attended Warden platform-admin `--exec` +envelope. Use credential routing first; do not export a token into a shell. +The owner wrapper must preserve its independent exit/self-revocation receipt. +The contained command is: + +```bash +python3 scripts/cadence_receiver_custody.py apply \ + --kubeconfig '' --server '' \ + --expected-receiver-image 'forgejo.coulomb.social/coulomb/audit-core@sha256:' \ + --confirm-packet-sha256 '' \ + --receipt '' +``` + +The command is silent; all subprocess outputs are captured, exceptions expose +fixed diagnostics only, and the0600 O_EXCL receipt contains metadata. It checks +the primary cluster UID, single ready receiver/pinned image, existing registry, +exact projection and custody provenance before reader grant writes. It extends +the coding-agent deny boundary for only the new data/metadata paths, provisions +CAS0 version1 with request provenance, and merges with exact registry-version +CAS. A concurrent edit fails; it never retries against an unseen registry. +It verifies exact reader path and scope/TTL, denied registry/sibling/metadata/list, +wrong service identity, revoked reader session and deny-boundary precedence; +then delivers ESO, compares bearer values privately, refreshes the receiver +registry secret, restarts the receiver and verifies loaded registry equality. +The new store is restricted to rapp-qonto, and source projection shape forbids +extra data or arbitrary templates. API-observed default fields are explicitly +pinned so a legitimate resumed projection remains equal. + +A partial failure retains custody/version1 for investigation. Inspect its +nonsecret receipt, confirm unchanged reviewed source, and rerun with a new +receipt path plus `--resume`. The same request's durable token is reused; +conflicting versions, provenance, registry scope or role/policy drift refuse. +No helper rotation, secret deletion or broad rollback occurs. A restart after +successful delivery may be repeated by an explicitly resumed attempt; it +preserves the token and registry content. + +Success is `delivered_pending_native_producer_and_independent_acceptance`. +It does not prove collector source capture, denial refusal, bearer revocation, +independent receiver reads, elapsed-hour heartbeat observation, or profile +acceptance. Read-only observer custody must remain independent; do not upgrade +the collector bearer. Keep the runtime active using the owner's authorized +health polling for the real hour, without changing scale-to-zero. A session +lasting15m must close normally; use a fresh attended session for hour-end reads. +Source-global missing-heartbeat findings must be interpreted alongside actual +active-instance inventory: intended scale-to-zero is not unexpected active +process silence. Preserve shutdown/restart scope separately. + +Residual native checks include isolated wrong-namespace store denial and +former-valid collector bearer revocation, alongside the source/tenant/read +refusal and per-instance observer matrix. The runner's wrong-service-identity +probes do not separately isolate wrong namespace while preserving the correct +service account name. Record that check honestly rather than treating the +combined refusal as the complete namespace test. diff --git a/manifests/rapp-qonto-audit-sender.yaml b/manifests/rapp-qonto-audit-sender.yaml new file mode 100644 index 0000000..70fbfeb --- /dev/null +++ b/manifests/rapp-qonto-audit-sender.yaml @@ -0,0 +1,57 @@ +# Proposed CCR-2026-0047; explicit ESO defaults make exact resume checks stable. +apiVersion: external-secrets.io/v1 +kind: ClusterSecretStore +metadata: + name: openbao-rapp-qonto-audit + annotations: + railiance.io/credential-change: CCR-2026-0047 + railiance.io/admission: proposed +spec: + provider: + vault: + server: http://openbao.openbao.svc:8200 + path: platform + version: v2 + auth: + kubernetes: + mountPath: kubernetes + role: external-secrets-rapp-qonto-audit + serviceAccountRef: + name: external-secrets + namespace: external-secrets + conditions: + - namespaces: + - rapp-qonto +--- +apiVersion: external-secrets.io/v1 +kind: ExternalSecret +metadata: + name: rapp-qonto-audit-sender + namespace: rapp-qonto + annotations: + railiance.io/credential-change: CCR-2026-0047 + railiance.io/admission: proposed +spec: + refreshInterval: 5m + secretStoreRef: + kind: ClusterSecretStore + name: openbao-rapp-qonto-audit + target: + template: + metadata: + annotations: + railiance.io/credential-change: CCR-2026-0047 + railiance.io/admission: proposed + engineVersion: v2 + mergePolicy: Replace + name: rapp-qonto-audit-sender + creationPolicy: Owner + deletionPolicy: Retain + data: + - secretKey: bearer + remoteRef: + key: workloads/rapp-qonto/audit-sender + property: AUDIT_TOKEN + conversionStrategy: Default + decodingStrategy: None + metadataPolicy: None diff --git a/openbao/policies/workload-kv-read-rapp-qonto-audit.hcl b/openbao/policies/workload-kv-read-rapp-qonto-audit.hcl new file mode 100644 index 0000000..f440116 --- /dev/null +++ b/openbao/policies/workload-kv-read-rapp-qonto-audit.hcl @@ -0,0 +1,3 @@ +path "platform/data/workloads/rapp-qonto/audit-sender" { + capabilities = ["read"] +} diff --git a/scripts/cadence_receiver_custody.py b/scripts/cadence_receiver_custody.py new file mode 100644 index 0000000..43304b2 --- /dev/null +++ b/scripts/cadence_receiver_custody.py @@ -0,0 +1,798 @@ +#!/usr/bin/env python3 +"""Proposed NK-WP-0035 attended custody+cadence delivery; no values in receipts. + +Plan is offline. Apply requires CCR0047 exact owner approval and the SHA256 of +current complete packet, immutable compatible receiver and attended platform +admin. A failed attempt retains KV version1 and must resume explicitly. +""" + +from __future__ import annotations + +import argparse +import base64 +import copy +import hashlib +import json +import os +import secrets +import sys +import time +from datetime import datetime, timezone +from pathlib import Path + +import yaml +from factory_audit_custody import ( + REGISTRY, + credential_module, + prepare_metadata, + receiver_check, + require, + snapshot, +) +from factory_audit_delivery import ( + custody_value, + projection_matches, +) +from keycape_approval_custody import read_optional, require_denied, role_matches +from state_hub_preflight_lane import ROOT, LaneError, bao, command, data, revoke + +CCR = ROOT / "credential-change-requests/CCR-2026-0047-rapp-qonto-audit-sender.yaml" +MANIFEST = ROOT / "manifests/rapp-qonto-audit-sender.yaml" +WORKSPACE = ROOT.parent +APPROVAL = WORKSPACE / "approval-engine/cadence.yaml" +QONTO = WORKSPACE / "qonto-assistant/specs/audit-emission-cadence.yaml" +PROPOSAL = ( + WORKSPACE / "net-kingdom/generated/coordination-frame/approval-engine-receiver.json" +) +BUILDER = WORKSPACE / "audit-core/audit_core/cadence_registration.py" + + +def contracts(approved=False): + module = credential_module() + ccr, errors, _ = module.validate_ccr(CCR) + require(not errors, "invalid_credential_request") + if approved: + require( + ccr["status"] in {"approved", "applied", "verified", "active"}, + "approved_ccr_required", + ) + for owner in ( + "platform-operator", + "audit-core-owner", + "rapp-qonto-owner", + "qonto-assistant-owner", + ): + require( + any( + x.get("decision") == "approved" and owner in x.get("reviewer", "") + for x in ccr["review"]["comments"] + ), + "named_owner_approval_required", + ) + policy = "workload-kv-read-rapp-qonto-audit" + role = "external-secrets-rapp-qonto-audit" + payload = { + "bound_service_account_names": ["external-secrets"], + "bound_service_account_namespaces": ["external-secrets"], + "policies": policy, + "ttl": "15m", + } + o = ccr["openbao"] + require( + o["kv_path"] == "platform/workloads/rapp-qonto/audit-sender" + and o["fields"] == ["AUDIT_TOKEN", "CUSTODY_REQUEST"] + and o["policy_name"] == policy + and o["metadata_read"] is False + and o["auth"]["mount"] == "kubernetes" + and o["auth"]["role"] == role + and o["auth"]["bound_claims_confirmed"] is True + and module.auth_payload(ccr) == payload, + "exact_contract_required", + ) + hcl = module.generated_policy_hcl(ccr) + require((ROOT / o["policy_file"]).read_text() == hcl, "policy_source_drift") + return { + "ccr": "CCR-2026-0047", + "name": "rapp-qonto-collector", + "namespace": "rapp-qonto", + "kv": "platform/data/workloads/rapp-qonto/audit-sender", + "metadata": "platform/metadata/workloads/rapp-qonto/audit-sender", + "policy": policy, + "hcl": hcl, + "role": role, + "role_payload": payload, + "store": "openbao-rapp-qonto-audit", + "secret": "rapp-qonto-audit-sender", + "secret_key": "bearer", + } + + +def reviewed_objects(): + objects = list(yaml.safe_load_all(MANIFEST.read_text())) + require(len(objects) == 2, "exact_projection_objects_required") + store, external = objects + require( + store["kind"] == "ClusterSecretStore" + and store["metadata"]["name"] == "openbao-rapp-qonto-audit" + and store["spec"]["conditions"] == [{"namespaces": ["rapp-qonto"]}], + "exact_store_scope_required", + ) + vault = store["spec"]["provider"]["vault"] + require( + vault + == { + "server": "http://openbao.openbao.svc:8200", + "path": "platform", + "version": "v2", + "auth": { + "kubernetes": { + "mountPath": "kubernetes", + "role": "external-secrets-rapp-qonto-audit", + "serviceAccountRef": { + "name": "external-secrets", + "namespace": "external-secrets", + }, + } + }, + }, + "exact_store_auth_required", + ) + require( + external["kind"] == "ExternalSecret" + and external["metadata"]["name"] == "rapp-qonto-audit-sender" + and external["metadata"]["namespace"] == "rapp-qonto" + and external["spec"]["secretStoreRef"] + == {"kind": "ClusterSecretStore", "name": "openbao-rapp-qonto-audit"}, + "exact_projection_target_required", + ) + spec = external["spec"] + require( + set(spec) == {"refreshInterval", "secretStoreRef", "target", "data"}, + "unexpected_projection_fields", + ) + require( + spec["data"] + == [ + { + "secretKey": "bearer", + "remoteRef": { + "key": "workloads/rapp-qonto/audit-sender", + "property": "AUDIT_TOKEN", + "conversionStrategy": "Default", + "decodingStrategy": "None", + "metadataPolicy": "None", + }, + } + ], + "exact_projection_field_required", + ) + target = spec["target"] + require( + set(target) == {"name", "creationPolicy", "deletionPolicy", "template"} + and target["name"] == "rapp-qonto-audit-sender" + and target["creationPolicy"] == "Owner" + and target["deletionPolicy"] == "Retain", + "exact_projection_ownership_required", + ) + template = target["template"] + require( + set(template) == {"metadata", "engineVersion", "mergePolicy"} + and template["engineVersion"] == "v2" + and template["mergePolicy"] == "Replace" + and set(template["metadata"]) == {"annotations"}, + "metadata_only_template_required", + ) + return objects + + +def packet(): + contracts() + reviewed_objects() + paths = [ + CCR, + MANIFEST, + ROOT / "openbao/policies/workload-kv-read-rapp-qonto-audit.hcl", + Path(__file__), + ROOT / "scripts/factory_audit_custody.py", + ROOT / "scripts/factory_audit_delivery.py", + ROOT / "scripts/keycape_approval_custody.py", + ROOT / "scripts/state_hub_preflight_lane.py", + ROOT / "scripts/credential-change.py", + APPROVAL, + QONTO, + PROPOSAL, + BUILDER, + WORKSPACE / "audit-core/audit_core/senders.py", + WORKSPACE / "audit-core/audit_core/emission_cadence.py", + ] + hashes = { + str(p.relative_to(WORKSPACE)): hashlib.sha256(p.read_bytes()).hexdigest() + for p in paths + } + digest = hashlib.sha256(json.dumps(hashes, sort_keys=True).encode()).hexdigest() + return { + "schema_version": "platform.cadence-custody-packet/v1", + "sha256": digest, + "source_sha256": hashes, + "status": "proposed", + "credential_values_emitted": False, + } + + +def builder(): + sys.path.insert(0, str(WORKSPACE / "audit-core")) + from audit_core.cadence_registration import prepare, qonto_metadata + + return prepare, qonto_metadata + + +def proposal_metadata(): + _, qonto_metadata = builder() + raw = QONTO.read_bytes() + return qonto_metadata(yaml.safe_load(raw), hashlib.sha256(raw).hexdigest()) + + +def preflight_custody(lane, *, resume=False): + """Refuse existing custody/scope drift before creating any reader grant.""" + _, _, rows, used = snapshot() + existing = [row for row in rows if row["name"] == lane["name"]] + old = read_optional(lane["metadata"]) + if old is None: + require(not existing, "registry_sender_without_custody") + return + require( + resume + and old["current_version"] == 1 + and not old["versions"]["1"].get("destroyed") + and not old["versions"]["1"].get("deletion_time"), + "reviewed_version1_resume_required", + ) + raw = data(bao(["read", "-format=json", lane["kv"]]))["data"] + token = custody_value(lane, raw) + wanted = dict(proposal_metadata(), tokens=[token]) + if existing: + require(existing == [wanted], "registered_sender_drift") + else: + require(token not in used, "sender_token_collision") + + +def merge_and_seed(lane, receipt, save, *, resume=False): + """CAS all registry metadata once; only new exact KV uses CAS0. + + All private values are in memory. A CAS conflict does not retry against a + new registry silently: operator re-preflights and resumes durable version1. + """ + version, body, rows, used = snapshot() + originals = copy.deepcopy(rows) + prepare, _ = builder() + updated, approval_receipt = prepare( + body, APPROVAL.read_bytes(), json.loads(PROPOSAL.read_text()) + ) + wanted = proposal_metadata() + existing = [row for row in rows if row["name"] == lane["name"]] + old = read_optional(lane["metadata"]) + if old is None: + require(not existing, "registry_sender_without_custody") + token = secrets.token_urlsafe(48) + require(token not in used, "sender_token_collision") + receipt.update(phase="custody_cas0", registry_version_before=version) + save() + result = data( + bao( + ["write", "-format=json", lane["kv"], "-"], + payload={ + "options": {"cas": 0}, + "data": {"AUDIT_TOKEN": token, "CUSTODY_REQUEST": lane["ccr"]}, + }, + ) + ) + require(result["data"]["version"] == 1, "initial_version_mismatch") + receipt.update(custody_version=1, custody_request_id=result.get("request_id")) + save() + else: + require( + resume + and old["current_version"] == 1 + and not old["versions"]["1"].get("destroyed") + and not old["versions"]["1"].get("deletion_time"), + "reviewed_version1_resume_required", + ) + raw = data(bao(["read", "-format=json", lane["kv"]]))["data"] + token = custody_value(lane, raw) + receipt["custody_version_reused"] = 1 + wanted["tokens"] = [token] + if existing: + require(existing == [wanted], "registered_sender_drift") + else: + require(token not in used, "sender_token_collision") + merged = json.loads(updated["senders.json"]) + if not existing: + merged.append(wanted) + updated["senders.json"] = json.dumps(merged, separators=(",", ":")) + # The real receiver parser validates the complete candidate before CAS. + from audit_core.senders import SenderRegistry + + SenderRegistry.from_env({"AUDIT_CORE_SENDERS": updated["senders.json"]}) + receipt.update(phase="registry_cas", approval=approval_receipt) + save() + if updated != body: + result = data( + bao( + ["write", "-format=json", REGISTRY, "-"], + payload={"options": {"cas": version}, "data": updated}, + ) + ) + receipt["registry_request_id"] = result.get("request_id") + after_version, after_body, after_rows, _ = snapshot() + require(after_body == updated, "registry_readback_mismatch") + require( + all(row in after_rows for row in originals if row["name"] != "approval-engine"), + "other_sender_changed", + ) + before_approval = next(row for row in originals if row["name"] == "approval-engine") + after_approval = next(row for row in after_rows if row["name"] == "approval-engine") + require( + { + k: v + for k, v in before_approval.items() + if k not in ("heartbeat_classes", "emission_cadence") + } + == { + k: v + for k, v in after_approval.items() + if k not in ("heartbeat_classes", "emission_cadence") + }, + "approval_credentials_or_scope_changed", + ) + receipt.update( + registry_version_after=after_version, + registry_readback_matches=True, + other_senders_preserved=True, + approval_credentials_preserved=True, + ) + save() + return token, updated + + +COMPATIBILITY = """import inspect,json,sys +from audit_core.sqlite_backend import SQLiteAuditBackend +from audit_core.postgres_backend import PostgresAuditBackend +from audit_core.senders import SenderRegistry +from audit_core.cadence_registration import qonto_metadata +entry=json.load(sys.stdin) +entry['tokens']=['fixture-only-never-native'] +r=SenderRegistry.from_env({'AUDIT_CORE_SENDERS':json.dumps([entry])}) +a=r.authenticate('Bearer fixture-only-never-native') +proof={'cadence_registration':callable(qonto_metadata), +'sqlite_instance_filter':'stream_instance_id' in inspect.signature(SQLiteAuditBackend.event_counts).parameters, +'postgres_instance_filter':'stream_instance_id' in inspect.signature(PostgresAuditBackend.event_counts).parameters, +'synthetic_cadence_parse':a.emission_cadence is not None and a.emission_cadence.source=='qonto-assistant', +'synthetic_heartbeat_parse':dict(a.heartbeat_classes)==entry['heartbeat_classes'], +'synthetic_exact_scope':a.sources==frozenset(entry['sources']) and a.tenants==frozenset(entry['tenants']) and a.may_write and not a.may_read} +print(json.dumps(proof)) +""" + + +def compatible(kube, image): + result = receiver_check(kube, image) + pod = data( + command( + kube + + [ + "-n", + "audit-core", + "get", + "pods", + "-l", + "app.kubernetes.io/name=audit-core,app.kubernetes.io/component=receiver", + "-o", + "json", + ] + ) + )["items"][0] + proof = data( + command( + kube + + [ + "-n", + "audit-core", + "exec", + "-i", + pod["metadata"]["name"], + "--", + "python", + "-c", + COMPATIBILITY, + ], + payload=proposal_metadata(), + ) + ) + require( + set(proof) + == { + "cadence_registration", + "sqlite_instance_filter", + "postgres_instance_filter", + "synthetic_cadence_parse", + "synthetic_heartbeat_parse", + "synthetic_exact_scope", + } + and all(v is True for v in proof.values()), + "receiver_cadence_instance_support_required", + ) + result["cadence_instance_capabilities"] = proof + return result + + +def reader_checks(kube, lane, token, receipt, save): + require( + role_matches(read_optional("auth/kubernetes/role/" + lane["role"]), lane), + "reader_role_drift", + ) + jwt = ( + command( + kube + + [ + "-n", + "external-secrets", + "create", + "token", + "external-secrets", + "--duration=10m", + ] + ) + .stdout.decode() + .strip() + ) + auth = data( + bao( + ["write", "-format=json", "auth/kubernetes/login", "-"], + payload={"role": lane["role"], "jwt": jwt}, + ) + )["auth"] + reader = auth["client_token"] + try: + require( + set(auth["token_policies"]) == {lane["policy"], "default"} + and 0 < auth["lease_duration"] <= 900, + "reader_scope_ttl_mismatch", + ) + raw = data(bao(["read", "-format=json", lane["kv"]], token=reader))["data"] + require(custody_value(lane, raw) == token, "reader_value_mismatch") + for path in ( + REGISTRY, + lane["metadata"], + "platform/data/workloads/approval-engine/audit-sender", + ): + require_denied( + bao(["read", "-format=json", path], token=reader, allow_failure=True) + ) + require_denied( + bao( + ["list", "-format=json", "platform/metadata/workloads"], + token=reader, + allow_failure=True, + ) + ) + finally: + revoke(reader) + require_denied( + bao(["token", "lookup", "-format=json"], token=reader, allow_failure=True) + ) + receipt["reader_exact_path_negatives_and_revocation"] = True + save() + # Exercise incorrect namespace and service account without creating workloads. + for namespace, account in ( + ("rapp-qonto", "default"), + ("external-secrets", "default"), + ): + jwt = ( + command( + kube + ["-n", namespace, "create", "token", account, "--duration=10m"] + ) + .stdout.decode() + .strip() + ) + bad = bao( + ["write", "-format=json", "auth/kubernetes/login", "-"], + payload={"role": lane["role"], "jwt": jwt}, + allow_failure=True, + ) + if bad.returncode == 0: + revoke(data(bad)["auth"]["client_token"]) + raise LaneError("wrong_service_account_or_namespace_authenticated") + require(b"403" in bad.stderr, "wrong_identity_denial_inconclusive") + child = data( + bao( + [ + "token", + "create", + "-format=json", + "-policy=" + lane["policy"], + "-policy=agent-high-risk-boundary", + "-no-default-policy", + "-ttl=60s", + ] + ) + )["auth"]["client_token"] + try: + for path in (lane["kv"], lane["metadata"]): + require_denied( + bao(["read", "-format=json", path], token=child, allow_failure=True) + ) + finally: + revoke(child) + receipt.update(wrong_identity_denied=True, coding_agent_deny_wins=True) + save() + + +def preflight_projection(kube, objects): + """Exact reviewed metadata template allowed; arbitrary template data refused.""" + for obj in objects: + meta = obj["metadata"] + prefix = kube + (["-n", meta["namespace"]] if "namespace" in meta else []) + result = command( + prefix + ["get", obj["kind"], meta["name"], "-o", "json"], + allow_failure=True, + ) + if result.returncode: + require(b"NotFound" in result.stderr, "projection_presence_unknown") + continue + live = data(result) + require( + all(live["spec"].get(k) == v for k, v in obj["spec"].items()), + "existing_projection_drift", + ) + require(not live["spec"].get("dataFrom"), "unexpected_projection_datafrom") + template = live["spec"].get("target", {}).get("template") + wanted = obj["spec"].get("target", {}).get("template") + require(template == wanted, "unexpected_projection_template") + + +def deliver(kube, lane, token, body, receipt, save, image): + objects = reviewed_objects() + preflight_projection(kube, objects) + command(kube + ["apply", "-f", str(MANIFEST)]) + command( + kube + + [ + "wait", + "--for=condition=Ready", + "clustersecretstore/" + lane["store"], + "--timeout=45s", + ] + ) + command( + kube + + [ + "-n", + lane["namespace"], + "wait", + "--for=condition=Ready", + "externalsecret/" + lane["secret"], + "--timeout=45s", + ] + ) + es = data( + command( + kube + + [ + "-n", + lane["namespace"], + "get", + "externalsecret", + lane["secret"], + "-o", + "json", + ] + ) + ) + secret = data( + command( + kube + + ["-n", lane["namespace"], "get", "secret", lane["secret"], "-o", "json"] + ) + ) + projection_matches(secret, es, lane, token) + receipt["projection_matches"] = True + save() + command( + kube + + [ + "-n", + "audit-core", + "annotate", + "externalsecret", + "audit-core-senders", + "force-sync=" + str(time.time_ns()), + "--overwrite", + ] + ) + for _ in range(25): + projected = data( + command( + kube + + [ + "-n", + "audit-core", + "get", + "secret", + "audit-core-senders", + "-o", + "json", + ] + ) + ) + if json.loads( + base64.b64decode(projected["data"]["senders.json"]) + ) == json.loads(body["senders.json"]): + break + time.sleep(2) + else: + raise LaneError("receiver_registry_projection_mismatch") + receipt["receiver_registry_projection_matches"] = True + save() + command(kube + ["-n", "audit-core", "rollout", "restart", "deployment/audit-core"]) + command( + kube + + [ + "-n", + "audit-core", + "rollout", + "status", + "deployment/audit-core", + "--timeout=50s", + ] + ) + receipt["receiver_after"] = compatible(kube, image) + # Environment inherited by exec must equal the private projected registry. + pod = data( + command( + kube + + [ + "-n", + "audit-core", + "get", + "pods", + "-l", + "app.kubernetes.io/name=audit-core,app.kubernetes.io/component=receiver", + "-o", + "json", + ] + ) + )["items"][0] + expected_hash = hashlib.sha256( + json.dumps(json.loads(body["senders.json"]), sort_keys=True).encode() + ).hexdigest() + probe = 'import os,json,hashlib;print(json.dumps({"sha256":hashlib.sha256(json.dumps(json.loads(os.environ["AUDIT_CORE_SENDERS"]),sort_keys=True).encode()).hexdigest()}))' + proof = data( + command( + kube + + [ + "-n", + "audit-core", + "exec", + pod["metadata"]["name"], + "--", + "python", + "-c", + probe, + ] + ) + ) + require(proof == {"sha256": expected_hash}, "receiver_runtime_registry_mismatch") + # Do not publish a low-entropy token hash or full registry hash in receipts. + receipt.update(receiver_reloaded=True, receiver_runtime_registry_matches=True) + save() + + +def main(): + p = argparse.ArgumentParser(description=__doc__) + p.add_argument("action", choices=["plan", "apply"]) + p.add_argument("--kubeconfig") + p.add_argument("--server") + p.add_argument("--expected-receiver-image") + p.add_argument("--confirm-packet-sha256") + p.add_argument("--receipt", type=Path) + p.add_argument("--resume", action="store_true") + a = p.parse_args() + receipt = { + "schema_version": "platform.cadence-custody/v1", + "status": "refused", + "credential_values_emitted": False, + "started_at": datetime.now(timezone.utc).isoformat(), + } + fd = None + try: + current = packet() + if a.action == "plan": + current["qonto_sender_metadata"] = proposal_metadata() + print(json.dumps(current, indent=2)) + return 0 + lane = contracts(approved=True) + require( + a.confirm_packet_sha256 == current["sha256"], + "exact_reviewed_packet_required", + ) + require( + a.kubeconfig and a.server and a.expected_receiver_image and a.receipt, + "explicit_runtime_inputs_required", + ) + require( + Path.home().parent.name == ".warden-attended-login" + and not os.getenv("BAO_TOKEN") + and not os.getenv("VAULT_TOKEN") + and not os.getenv("OPENBAO_TOKEN"), + "attended_warden_envelope_required", + ) + fd = os.open( + a.receipt, os.O_RDWR | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600 + ) + + def save(): + os.lseek(fd, 0, os.SEEK_SET) + os.ftruncate(fd, 0) + os.write(fd, (json.dumps(receipt, indent=2) + "\n").encode()) + os.fsync(fd) + + receipt["packet_sha256"] = current["sha256"] + save() + kube = [ + "kubectl", + "--kubeconfig", + a.kubeconfig, + "--server", + a.server, + "--request-timeout=20s", + ] + receipt["receiver_before"] = compatible(kube, a.expected_receiver_image) + save() + identity = data(bao(["token", "lookup", "-format=json"]))["data"] + require( + "platform-admin" in identity["policies"] + and "root" not in identity["policies"], + "attended_platform_admin_required", + ) + command(kube + ["get", "namespace", "rapp-qonto", "-o", "name"]) + # Validate registry, declarations, lane and projection before metadata writes. + _, body, _, _ = snapshot() + prepare, _ = builder() + prepare(body, APPROVAL.read_bytes(), json.loads(PROPOSAL.read_text())) + proposal_metadata() + preflight_projection(kube, reviewed_objects()) + preflight_custody(lane, resume=a.resume) + receipt["phase"] = "metadata" + save() + prepare_metadata([lane]) + token, updated = merge_and_seed(lane, receipt, save, resume=a.resume) + receipt["phase"] = "reader_verification" + save() + reader_checks(kube, lane, token, receipt, save) + receipt["phase"] = "delivery" + save() + deliver(kube, lane, token, updated, receipt, save, a.expected_receiver_image) + receipt.update( + status="delivered_pending_native_producer_and_independent_acceptance", + completed_at=datetime.now(timezone.utc).isoformat(), + ) + save() + return 0 + except Exception as exc: + receipt.update( + status="refused", + error=str(exc) + if isinstance(exc, LaneError) + else "contained_operation_failed", + ) + if fd is not None: + save() + elif a.action == "plan": + print(json.dumps(receipt, indent=2)) + return 1 + finally: + if fd is not None: + os.close(fd) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_cadence_receiver_custody.py b/tests/test_cadence_receiver_custody.py new file mode 100644 index 0000000..7037dc5 --- /dev/null +++ b/tests/test_cadence_receiver_custody.py @@ -0,0 +1,247 @@ +"""Contained model exercises, no live registry reads or credentials.""" + +import copy +import json +import subprocess +import sys +import unittest +from pathlib import Path +from unittest.mock import patch + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "scripts")) +import cadence_receiver_custody as lane # noqa: E402 + + +class PacketContracts(unittest.TestCase): + def test_proposed_refuses_before_native_calls(self): + with patch.object(lane, "bao", side_effect=AssertionError("no native call")): + with self.assertRaisesRegex(lane.LaneError, "approved_ccr_required"): + lane.contracts(approved=True) + + def test_exact_namespace_path_policy_and_projection(self): + contract = lane.contracts() + self.assertEqual(contract["namespace"], "rapp-qonto") + self.assertEqual(contract["name"], "rapp-qonto-collector") + objects = list(lane.yaml.safe_load_all(lane.MANIFEST.read_text())) + self.assertEqual( + objects[0]["spec"]["conditions"], [{"namespaces": ["rapp-qonto"]}] + ) + self.assertEqual( + objects[1]["spec"]["data"], + [ + { + "secretKey": "bearer", + "remoteRef": { + "key": "workloads/rapp-qonto/audit-sender", + "property": "AUDIT_TOKEN", + "conversionStrategy": "Default", + "decodingStrategy": "None", + "metadataPolicy": "None", + }, + } + ], + ) + self.assertEqual( + contract["hcl"], + 'path "platform/data/workloads/rapp-qonto/audit-sender" {\n capabilities = ["read"]\n}\n', + ) + + def test_offline_plan_contains_hashes_but_no_credentials(self): + result = lane.packet() + self.assertEqual(len(result["sha256"]), 64) + self.assertIn( + "audit-core/audit_core/cadence_registration.py", result["source_sha256"] + ) + self.assertFalse(result["credential_values_emitted"]) + self.assertNotIn("tokens", json.dumps(result)) + + def test_receiver_compatibility_probe_against_actual_source(self): + result = subprocess.run( + [ + str(lane.WORKSPACE / "audit-core/.venv/bin/python"), + "-c", + lane.COMPATIBILITY, + ], + cwd=lane.WORKSPACE / "audit-core", + capture_output=True, + input=json.dumps(lane.proposal_metadata()).encode(), + check=True, + ) + self.assertTrue(all(json.loads(result.stdout).values())) + + +class FakeCustody(unittest.TestCase): + def setUp(self): + self.contract = lane.contracts() + self.kv = None + self.version = 7 + self.rows = [ + dict( + name="approval-engine", + tokens=["a" * 64], + sources=["approval-engine"], + tenants=["tenant:platform"], + may_write=True, + may_read=False, + evidence_kind="load-bearing", + secret_policy="redact", + ), + dict( + name="reader", + tokens=["b" * 64], + sources=["*"], + tenants=["*"], + may_write=False, + may_read=True, + secret_policy="reject", + ), + ] + self.body = {"senders.json": json.dumps(self.rows), "other": "keep"} + self.writes = [] + self.saves = [] + self.receipt = {} + self.fail_registry = False + self.snapshot_patch = patch.object(lane, "snapshot", side_effect=self.snapshot) + self.optional_patch = patch.object( + lane, "read_optional", side_effect=self.metadata + ) + self.bao_patch = patch.object(lane, "bao", side_effect=self.bao) + for p in [self.snapshot_patch, self.optional_patch, self.bao_patch]: + p.start() + self.addCleanup(p.stop) + + def snapshot(self): + rows = json.loads(self.body["senders.json"]) + return ( + self.version, + copy.deepcopy(self.body), + rows, + {t for r in rows for t in r["tokens"]}, + ) + + def metadata(self, path): + self.assertEqual(path, self.contract["metadata"]) + return ( + None if self.kv is None else {"current_version": 1, "versions": {"1": {}}} + ) + + def bao(self, args, *, payload=None): + if args[0] == "read": + self.assertEqual(args[2], self.contract["kv"]) + result = {"data": {"metadata": {"version": 1}, "data": self.kv}} + elif args[2] == self.contract["kv"]: + self.assertIsNone(self.kv) + self.assertEqual(payload["options"], {"cas": 0}) + self.kv = copy.deepcopy(payload["data"]) + self.writes.append("custody") + result = {"data": {"version": 1}, "request_id": "test-custody"} + else: + self.assertEqual(args[2], lane.REGISTRY) + if self.fail_registry: + raise lane.LaneError("simulated_registry_conflict") + self.assertEqual(payload["options"]["cas"], self.version) + self.body = copy.deepcopy(payload["data"]) + self.version += 1 + self.writes.append("registry") + result = {"data": {"version": self.version}, "request_id": "test-registry"} + return subprocess.CompletedProcess([], 0, json.dumps(result).encode(), b"") + + def run_merge(self, resume=False): + return lane.merge_and_seed( + self.contract, + self.receipt, + lambda: self.saves.append(copy.deepcopy(self.receipt)), + resume=resume, + ) + + def test_first_provision_preserves_all_other_credentials_and_fields(self): + token, updated = self.run_merge() + rows = json.loads(updated["senders.json"]) + self.assertEqual(rows[0]["tokens"], ["a" * 64]) + self.assertEqual(rows[1], self.rows[1]) + self.assertEqual(rows[2]["tokens"], [token]) + self.assertEqual(rows[2]["heartbeat_classes"], {"audit.deny": 7200}) + self.assertEqual(updated["other"], "keep") + self.assertNotIn(token, json.dumps(self.saves)) + self.assertEqual(self.writes, ["custody", "registry"]) + + def test_idempotent_resume_reuses_version1_and_no_registry_write(self): + token, _ = self.run_merge() + prior = list(self.writes) + resumed, _ = self.run_merge(resume=True) + self.assertEqual(resumed, token) + self.assertEqual(self.writes, prior) + with self.assertRaisesRegex( + lane.LaneError, "reviewed_version1_resume_required" + ): + self.run_merge() + + def test_partial_failure_after_cas0_resumes_without_rotation(self): + self.fail_registry = True + with self.assertRaisesRegex(lane.LaneError, "simulated_registry_conflict"): + self.run_merge() + token = self.kv["AUDIT_TOKEN"] + self.assertEqual(self.writes, ["custody"]) + self.fail_registry = False + actual, _ = self.run_merge(resume=True) + self.assertEqual(actual, token) + self.assertEqual(self.writes, ["custody", "registry"]) + + def test_conflicting_custody_provenance_refuses(self): + self.run_merge() + self.kv["CUSTODY_REQUEST"] = "OTHER" + with self.assertRaisesRegex(lane.LaneError, "custody_provenance_mismatch"): + self.run_merge(resume=True) + + def test_registered_scope_drift_refuses(self): + self.run_merge() + rows = json.loads(self.body["senders.json"]) + rows[-1]["tenants"] = ["wrong"] + self.body["senders.json"] = json.dumps(rows) + with self.assertRaisesRegex(lane.LaneError, "registered_sender_drift"): + self.run_merge(resume=True) + + def test_orphan_registry_sender_refuses_without_custody_write(self): + rows = json.loads(self.body["senders.json"]) + rows.append(dict(lane.proposal_metadata(), tokens=["c" * 64])) + self.body["senders.json"] = json.dumps(rows) + with self.assertRaisesRegex(lane.LaneError, "registry_sender_without_custody"): + self.run_merge() + self.assertEqual(self.writes, []) + + def test_custody_preflight_refuses_bad_provenance_before_grant_writes(self): + self.run_merge() + self.kv["CUSTODY_REQUEST"] = "unknown-owner" + with patch.object( + lane, "prepare_metadata", side_effect=AssertionError("no grants") + ): + with self.assertRaisesRegex(lane.LaneError, "custody_provenance_mismatch"): + lane.preflight_custody(self.contract, resume=True) + + +class ProjectionResume(unittest.TestCase): + def test_exact_metadata_template_is_resumable(self): + objects = list(lane.yaml.safe_load_all(lane.MANIFEST.read_text())) + responses = [ + subprocess.CompletedProcess([], 0, json.dumps(o).encode(), b"") + for o in objects + ] + with patch.object(lane, "command", side_effect=responses): + lane.preflight_projection(["kubectl"], objects) + + def test_extra_template_payload_is_refused(self): + objects = list(lane.yaml.safe_load_all(lane.MANIFEST.read_text())) + actual = copy.deepcopy(objects) + actual[1]["spec"]["target"]["template"]["data"] = {"injected": "payload"} + responses = [ + subprocess.CompletedProcess([], 0, json.dumps(o).encode(), b"") + for o in actual + ] + with patch.object(lane, "command", side_effect=responses): + with self.assertRaisesRegex(lane.LaneError, "existing_projection_drift"): + lane.preflight_projection(["kubectl"], objects) + + +if __name__ == "__main__": + unittest.main() diff --git a/workplans/RPF-WP-0059-hourly-cadence-receiver-custody.md b/workplans/RPF-WP-0059-hourly-cadence-receiver-custody.md new file mode 100644 index 0000000..ff41666 --- /dev/null +++ b/workplans/RPF-WP-0059-hourly-cadence-receiver-custody.md @@ -0,0 +1,71 @@ +--- +id: RPF-WP-0059 +type: workplan +title: "Admit hourly receiver cadence and exact Qonto collector custody" +domain: financials +repo: railiance-platform +status: proposed +owner: codex +topic_slug: railiance-platform +created: "2026-10-15" +updated: "2026-10-15" +origin: cross-repo +origin_ref: NK-WP-0035 +--- + +# Hourly receiver cadence and Qonto collector custody + +## Prepare exact bounded source packet + +```task +id: RPF-WP-0059-T01 +status: done +priority: high +``` + +Prepare CCR-2026-0047, exact-only reader policy, namespace-restricted ESO +projection and silent attended owner runner. Preserve Approval credentials and +all other senders while admitting its source cadence and the new Qonto sender. +Focused model tests cover CAS0, CAS-preserving merge, version1 resume, +interruption, provenance/scope drift, exact ESO resume/template refusal and +no-native proposed refusal. Details: docs/cadence-receiver-custody.md. + +## Review new grant and complete concrete packet + +```task +id: RPF-WP-0059-T02 +status: wait +priority: high +``` + +Explicit approval required from platform operator, Audit Core, rapp-qonto and +qonto-assistant owners. CCR0047 remains proposed. Prior timing rollout approval +is not a new bearer grant. Review the post-approval exact packet SHA256 and +immutable compatible receiver image before attended execution. + +## Execute attended custody, delivery and receiver readback + +```task +id: RPF-WP-0059-T03 +status: wait +priority: high +``` + +After T02 and the reviewed receiver release, use Warden platform-admin --exec. +Record independent session exit/self-revocation and native command receipt. +Run exact grant/negative checks and ESO/runtime registry equality; use durable +resume on interruption. No live effect or secret read occurred in preparation. + +## Collect actual producer and independent observer acceptance + +```task +id: RPF-WP-0059-T04 +status: wait +priority: high +``` + +Coordinate QONTO-WP-0006-T02, AUDIT-WP-0013-T05 and KG-WP-0005-T03. Preserve +source/tenant/instance scope and real elapsed-hour periodic evidence, isolated +wrong-namespace denial, actual formerly-valid bearer revocation, independent +protected readback, missing-class/lag/divergence, restart/scale-zero evidence. +Never infer profile acceptance from a model test or registry projection.