railiance-platform/scripts/cadence_observer_readback.py

#!/usr/bin/env python3
"""Bounded independent operator readback for NK-WP-0035; no credential output."""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import re
import socket
import subprocess
import time
from datetime import datetime, timezone
from pathlib import Path
from urllib.error import HTTPError
from urllib.parse import urlencode
from urllib.request import ProxyHandler, Request, build_opener

from cadence_receiver_custody import compatible
from factory_audit_custody import require, snapshot
from native_factory_acceptance import NoRedirect, operator_identity
from state_hub_preflight_lane import LaneError, bao, data

SCOPES = {
    "qonto-assistant": "tenant:friendly:binky",
    "approval-engine": "tenant:platform",
}


def stamp(value):
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    require(parsed.tzinfo is not None, "timezone_required")
    return parsed.astimezone(timezone.utc)


def paths(source, tenant, instance, since, until, event_ids):
    require(SCOPES.get(source) == tenant, "exact_source_tenant_required")
    require(
        isinstance(instance, str)
        and 1 <= len(instance) <= 256
        and instance.strip() == instance,
        "exact_instance_required",
    )
    require(
        stamp(since) <= stamp(until) <= datetime.now(timezone.utc),
        "invalid_acceptance_window",
    )
    require(
        1 <= len(event_ids) <= 256 and len(set(event_ids)) == len(event_ids),
        "bounded_unique_event_ids_required",
    )
    require(
        all(
            isinstance(x, str) and re.fullmatch(r"[A-Za-z0-9_.:-]{1,200}", x)
            for x in event_ids
        ),
        "invalid_event_id",
    )
    query = urlencode(
        dict(
            source=source,
            tenant=tenant,
            stream_instance_id=instance,
            since=since,
            until=until,
        )
    )
    return ["/v1/reconciliation?" + query, "/v1/stream-findings"] + [
        "/v1/events/" + x for x in event_ids
    ]


def read_http(origin, token, path, allowed):
    require(path in allowed, "read_path_not_in_exact_packet")
    request = Request(origin + path, headers={"Authorization": "Bearer " + token})
    try:
        response = build_opener(ProxyHandler({}), NoRedirect()).open(
            request, timeout=10
        )
    except HTTPError as exc:
        response = exc
    with response:
        raw = response.read(1048577)
        require(len(raw) <= 1048576, "readback_too_large")
        require(response.status == 200, "independent_readback_failed")
        return json.loads(raw)


def sanitize_record(record, source, tenant, instance, event_id, since, until):
    require(
        record.get("event_id") == event_id
        and record.get("source") == source
        and record.get("tenant") == tenant,
        "record_source_tenant_id_mismatch",
    )
    detail = record["details"]["data"]
    require(detail.get("stream_instance_id") == instance, "record_instance_mismatch")
    require(
        type(detail.get("stream_sequence")) is int and detail["stream_sequence"] > 0,
        "record_sequence_missing",
    )
    accepted = record["accepted_at"]
    require(
        stamp(since) <= stamp(accepted) < stamp(until),
        "record_outside_acceptance_window",
    )
    action = record["action"]
    coverage = (
        [detail["class"]]
        if action == "audit-core.heartbeat" and "class" in detail
        else []
    )
    if source == "qonto-assistant":
        require(
            action in {"audit.allow", "audit.deny", "audit-core.heartbeat"},
            "record_undeclared_class",
        )
        if action == "audit-core.heartbeat":
            require(coverage == ["audit.deny"], "record_heartbeat_scope_mismatch")
    return {
        "event_id": event_id,
        "source": source,
        "tenant": tenant,
        "stream_instance_id": instance,
        "stream_sequence": detail["stream_sequence"],
        "type": action,
        "accepted_at": accepted,
        "observed_at": record["observed_at"],
        "covered_event_classes": coverage,
        "independent_custody_readback": True,
    }


def readback(origin, token, source, tenant, instance, since, until, event_ids):
    allowed = paths(source, tenant, instance, since, until, event_ids)
    reconciliation = read_http(origin, token, allowed[0], allowed)
    require(
        all(
            reconciliation.get(k) == v
            for k, v in (
                ("source", source),
                ("tenant", tenant),
                ("stream_instance_id", instance),
                ("since", since),
                ("until", until),
            )
        ),
        "reconciliation_scope_window_mismatch",
    )
    raw_findings = read_http(origin, token, allowed[1], allowed)
    findings = raw_findings.get("stream_findings", raw_findings.get("findings"))
    require(isinstance(findings, list), "unexpected_findings_shape")
    safe_findings = [
        {
            key: row.get(key)
            for key in (
                "kind",
                "sender",
                "source",
                "class",
                "expected_within_seconds",
                "last_seen",
                "overdue_seconds",
            )
        }
        for row in findings
        if row.get("source") == source
    ]
    records = [
        sanitize_record(
            read_http(origin, token, path, allowed),
            source,
            tenant,
            instance,
            event_id,
            since,
            until,
        )
        for path, event_id in zip(allowed[2:], event_ids, strict=True)
    ]
    return {
        "reconciliation": reconciliation,
        "stream_findings": safe_findings,
        "events": records,
        "scope": "bounded same-instance accepted custody; source-global findings require active-instance interpretation",
    }


def main():
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument("--source", choices=SCOPES, required=True)
    for key in (
        "tenant",
        "instance",
        "since",
        "until",
        "kubeconfig",
        "server",
        "expected-receiver-image",
    ):
        p.add_argument("--" + key, required=True)
    p.add_argument("--event-id", action="append", required=True)
    p.add_argument("--receipt", type=Path, required=True)
    a = p.parse_args()
    receipt = {
        "schema_version": "platform.independent-cadence-readback/v1",
        "status": "refused",
        "credential_values_emitted": False,
        "observed_at": datetime.now(timezone.utc).isoformat(),
    }
    fd = None
    forward = None
    try:
        paths(a.source, a.tenant, a.instance, a.since, a.until, a.event_id)
        require(
            Path.home().parent.name == ".warden-attended-login"
            and not any(
                os.getenv(x) for x in ("BAO_TOKEN", "VAULT_TOKEN", "OPENBAO_TOKEN")
            ),
            "fresh_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)

        kube = [
            "kubectl",
            "--kubeconfig",
            a.kubeconfig,
            "--server",
            a.server,
            "--request-timeout=20s",
        ]
        receipt["receiver"] = 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",
        )
        version, _, rows, _ = snapshot()
        operator = operator_identity(rows)
        token = (operator.get("tokens") or [operator["token"]])[0]
        receipt.update(reader_name=operator["name"], registry_version=version)
        with socket.socket() as sock:
            sock.bind(("127.0.0.1", 0))
            port = sock.getsockname()[1]
        forward = subprocess.Popen(
            kube
            + [
                "-n",
                "audit-core",
                "port-forward",
                "--address=127.0.0.1",
                "service/audit-core",
                str(port) + ":8080",
            ],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        )
        for _ in range(40):
            require(forward.poll() is None, "port_forward_failed")
            try:
                with socket.create_connection(("127.0.0.1", port), timeout=0.2):
                    break
            except OSError:
                time.sleep(0.2)
        else:
            raise LaneError("port_forward_not_ready")
        receipt.update(
            readback(
                "http://127.0.0.1:" + str(port),
                token,
                a.source,
                a.tenant,
                a.instance,
                a.since,
                a.until,
                a.event_id,
            )
        )
        receipt.update(
            status="independent_custody_readback_complete",
            helper_sha256=hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
        )
        save()
        return 0
    except Exception as exc:
        receipt["error"] = (
            str(exc) if isinstance(exc, LaneError) else "contained_readback_failed"
        )
        if fd is not None:
            save()
        return 1
    finally:
        if forward is not None:
            forward.terminate()
            try:
                forward.wait(timeout=5)
            except subprocess.TimeoutExpired:
                forward.kill()
                forward.wait(timeout=5)
        if fd is not None:
            os.close(fd)


if __name__ == "__main__":
    raise SystemExit(main())