coulomb/railiance-platform
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Prepare exact Qonto audit custody and hourly receiver admission packet

Browse source 
Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a13eab-1e6a-7b33-9c37-372e90ada3a2
This commit is contained in:
codex 2026-10-15 21:26:11 +02:00
parent 643006aa25
commit d5a86b74d1
7 changed files with 1379 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

108
credential-change-requests/CCR-2026-0047-rapp-qonto-audit-sender.yaml Normal file
View file

@ -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

95
docs/cadence-receiver-custody.md Normal file
View file

@ -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 '<owner-selected-kubeconfig>' --server '<verified-primary-server>' \
--expected-receiver-image 'forgejo.coulomb.social/coulomb/audit-core@sha256:<reviewed-digest>' \
--confirm-packet-sha256 '<reviewed-post-approval-plan-sha256>' \
--receipt '<new-private-receipt-path>'
```
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.

57
manifests/rapp-qonto-audit-sender.yaml Normal file
View file

@ -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

3
openbao/policies/workload-kv-read-rapp-qonto-audit.hcl Normal file
View file

@ -0,0 +1,3 @@
path "platform/data/workloads/rapp-qonto/audit-sender" {
capabilities = ["read"]
}

798
scripts/cadence_receiver_custody.py Normal file
View file

@ -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())

247
tests/test_cadence_receiver_custody.py Normal file
View file

@ -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()

71
workplans/RPF-WP-0059-hourly-cadence-receiver-custody.md Normal file
View file

@ -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.