railiance-platform/scripts/cadence_receiver_custody.py

#!/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())