6e24b09eaa
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a13eab-1e6a-7b33-9c37-372e90ada3a2
333 lines
11 KiB
Python
333 lines
11 KiB
Python
#!/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")
|
|
if source == "qonto-assistant":
|
|
require(
|
|
isinstance(instance, str)
|
|
and 1 <= len(instance) <= 256
|
|
and instance.strip() == instance,
|
|
"exact_instance_required",
|
|
)
|
|
else:
|
|
require(instance is None, "approval_has_no_declared_instance_filter")
|
|
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_values = dict(source=source, tenant=tenant, since=since, until=until)
|
|
if instance is not None:
|
|
query_values["stream_instance_id"] = instance
|
|
query = urlencode(query_values)
|
|
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"]
|
|
if source == "qonto-assistant":
|
|
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")
|
|
if source == "approval-engine":
|
|
names = {"issuance", "use", "supersession", "revocation"}
|
|
require(
|
|
action in {"approval." + name for name in names} | {"audit-core.heartbeat"},
|
|
"record_undeclared_class",
|
|
)
|
|
if action == "audit-core.heartbeat":
|
|
require(
|
|
len(coverage) == 1 and coverage[0] in names,
|
|
"record_heartbeat_scope_mismatch",
|
|
)
|
|
return {
|
|
"event_id": event_id,
|
|
"source": source,
|
|
"tenant": tenant,
|
|
"stream_instance_id": instance,
|
|
"stream_sequence": detail.get("stream_sequence")
|
|
if instance is not None
|
|
else None,
|
|
"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(
|
|
reconciliation.get("source") == source
|
|
and reconciliation.get("tenant") == tenant
|
|
and reconciliation.get("stream_instance_id") == instance
|
|
and stamp(reconciliation["since"]) == stamp(since)
|
|
and stamp(reconciliation["until"]) == stamp(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"
|
|
if instance is not None
|
|
else "bounded source lifecycle acceptance window; no process-instance completeness claim"
|
|
),
|
|
}
|
|
|
|
|
|
def main():
|
|
p = argparse.ArgumentParser(description=__doc__)
|
|
p.add_argument("--source", choices=SCOPES, required=True)
|
|
for key in (
|
|
"tenant",
|
|
"since",
|
|
"until",
|
|
"kubeconfig",
|
|
"server",
|
|
"expected-receiver-image",
|
|
):
|
|
p.add_argument("--" + key, required=True)
|
|
p.add_argument(
|
|
"--instance",
|
|
help="Required only for Qonto; Approval has no declared instance field",
|
|
)
|
|
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())
|