#!/usr/bin/env python3 """Silent, attended exact-plan caller custody. Preview is the default.""" from __future__ import annotations import argparse import hashlib import hmac import json import os import re import secrets import subprocess import uuid from pathlib import Path from urllib.request import urlopen from ledger_archive import Refused, checkpoint, private_directory ROOT = Path(__file__).resolve().parents[1] PLAN = ROOT / "deploy/closure/plan.json" KV = "platform/workloads/email-connect/callers/user-engine" WRITER = "email-connect-scoped-caller-writer" READER = "email-connect-scoped-caller-reader" READERS = [ ("email-connect", "email-connect-scoped-caller-eso"), ("user-engine", "user-engine-scoped-mail-eso"), ] FAILURE_CODES = frozenset({ "attended_bao_operation_failed", "reviewed_plan_hash_required", "reviewed_artifact_changed", "founder_exact_plan_decision_required", "exact_caller_contract_required", "existing_policy_differs", "policy_readback_differs", "bounded_writer_required", "initial_cas_write_required", "caller_record_changed", "caller_credential_invalid", "caller_readback_differs", "existing_reader_role_differs", "private_receipt_path_required", "fresh_receipt_required", }) def failure_receipt(error): code = str(error) if isinstance(error, Refused) else None return {"status": "failed", "failure_type": type(error).__name__, "failure_code": code if code in FAILURE_CODES else "contained_operation_failed"} def role(namespace, account): return { "bound_service_account_names": [account], "bound_service_account_namespaces": [namespace], "audience": "openbao", "policies": [READER], "ttl": "15m", "token_max_ttl": "15m", "token_explicit_max_ttl": "15m", "token_no_default_policy": True, "token_type": "service", } def bao(args, payload=None, token=None, missing=False): env = os.environ.copy() if token: env.update(BAO_TOKEN=token, VAULT_TOKEN=token) raw = payload if isinstance(payload, bytes) else None if payload is None else json.dumps(payload).encode() result = subprocess.run(["bao", *args], input=raw, env=env, capture_output=True, timeout=60) if result.returncode: if missing and (b"Code: 404" in result.stderr or b"No value found" in result.stderr): return None raise Refused("attended_bao_operation_failed") return result.stdout def approved_plan(plan, reviewed, decision, base): expected = hashlib.sha256(PLAN.read_bytes()).hexdigest() if reviewed != expected: raise Refused("reviewed_plan_hash_required") for name, expected_hash in plan["artifacts"].items(): path = ROOT / name if not path.resolve().is_relative_to(ROOT) or hashlib.sha256(path.read_bytes()).hexdigest() != expected_hash: raise Refused("reviewed_artifact_changed") uuid.UUID(decision) with urlopen(base.rstrip("/") + "/decisions/" + decision, timeout=15) as response: record = json.loads(response.read(65536)) if (record.get("status") != "resolved" or record.get("decision_type") != "made" or record.get("decided_by") != "User" or expected not in (record.get("description", "") or "")): raise Refused("founder_exact_plan_decision_required") custody = plan["caller_custody"] if (custody["kv_path"] != KV or custody["field"] != "TOKEN" or custody["writer_policy"] != WRITER or custody["reader_policy"] != READER or [(r["namespace"], r["service_account"]) for r in custody["readers"]] != READERS): raise Refused("exact_caller_contract_required") return expected def install_policy(name, path): intended = path.read_bytes() raw = bao(["read", "-format=json", "sys/policies/acl/" + name], missing=True) observed = None if raw is None else json.loads(raw)["data"]["policy"].encode() if observed is not None and observed.strip() != intended.strip(): raise Refused("existing_policy_differs") if observed is None: bao(["write", "sys/policies/acl/" + name, "-"], {"policy": intended.decode()}) observed = json.loads(bao(["read", "-format=json", "sys/policies/acl/" + name]))["data"]["policy"] if observed.strip() != intended.decode().strip(): raise Refused("policy_readback_differs") def provision(transport=bao): auth = json.loads(transport([ "token", "create", "-format=json", "-policy=" + WRITER, "-no-default-policy", "-ttl=5m", "-explicit-max-ttl=5m", "-use-limit=8", ]))["auth"] token = auth["client_token"] try: if (auth.get("policies") != [WRITER] or auth.get("identity_policies", []) or not 0 < auth.get("lease_duration", 0) <= 300): raise Refused("bounded_writer_required") metadata = transport(["read", "-format=json", "platform/metadata/workloads/email-connect/callers/user-engine"], token=token, missing=True) if metadata is None: value = secrets.token_urlsafe(48) written = json.loads(transport(["write", "-format=json", "platform/data/workloads/email-connect/callers/user-engine", "-"], {"options": {"cas": 0}, "data": {"TOKEN": value}}, token=token)) version = written["data"]["version"] if version != 1: raise Refused("initial_cas_write_required") created = True else: version = json.loads(metadata)["data"]["current_version"] value = None created = False readback = json.loads(transport(["read", "-format=json", "platform/data/workloads/email-connect/callers/user-engine"], token=token))["data"] if set(readback["data"]) != {"TOKEN"} or readback["metadata"]["version"] != version: raise Refused("caller_record_changed") stored = readback["data"]["TOKEN"] if not isinstance(stored, str) or not re.fullmatch(r"[!-~]{16,512}", stored): raise Refused("caller_credential_invalid") if value is not None and not hmac.compare_digest(value, stored): raise Refused("caller_readback_differs") return {"kv_version": version, "created": created, "readback_equal": True, "credential_values_printed": False} finally: transport(["token", "revoke", "-self"], token=token) def execute(plan, reviewed, decision, base): binding = approved_plan(plan, reviewed, decision, base) for name, filename in ((WRITER, "caller-writer.hcl"), (READER, "caller-reader.hcl")): install_policy(name, ROOT / "deploy/closure" / filename) for namespace, account in READERS: wanted = role(namespace, account) observed = bao(["read", "-format=json", "auth/kubernetes/role/" + account], missing=True) if observed is None: bao(["write", "auth/kubernetes/role/" + account, "-"], wanted) observed = bao(["read", "-format=json", "auth/kubernetes/role/" + account]) data = json.loads(observed)["data"] if (data.get("bound_service_account_names") != [account] or data.get("bound_service_account_namespaces") != [namespace] or data.get("audience") != "openbao" or (data.get("token_policies") or data.get("policies")) != [READER] or data.get("token_no_default_policy") is not True or data.get("token_max_ttl") not in (900, "15m") or data.get("token_explicit_max_ttl") not in (900, "15m")): raise Refused("existing_reader_role_differs") return dict(provision(), status="caller_custody_prepared", plan_sha256=binding, decision_id=decision, workloads_changed=False) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--execute", action="store_true") parser.add_argument("--reviewed-sha256") parser.add_argument("--decision-id") parser.add_argument("--api-base", default="http://127.0.0.1:8000") parser.add_argument("--receipt", type=Path) args = parser.parse_args() plan = json.loads(PLAN.read_text()) if not args.execute: print(json.dumps({"status": "prepared", "plan_sha256": hashlib.sha256(PLAN.read_bytes()).hexdigest(), "credential_generated": False, "workloads_changed": False})) return if not args.receipt or not args.receipt.is_absolute(): raise Refused("private_receipt_path_required") private_directory(args.receipt.parent) if args.receipt.exists() or args.receipt.is_symlink(): raise Refused("fresh_receipt_required") try: receipt = execute(plan, args.reviewed_sha256, args.decision_id, args.api_base) except Exception as error: checkpoint(args.receipt, failure_receipt(error)) raise checkpoint(args.receipt, receipt) if __name__ == "__main__": try: main() except Exception: raise SystemExit(1) from None