informed-decision/tools/create_sitting_approvals.py

"""Create compact-sitting approval objects (INFD-WP-0002-T03).

Attended reader only. Does not bind, consume, or print CLIENT_SECRET.
Skips infd-20260914-c01 until its create_client is decided.
"""
from __future__ import annotations

import argparse
import json
import os
import stat
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
from urllib.error import HTTPError
from urllib.parse import urlencode
from urllib.request import HTTPRedirectHandler, ProxyHandler, Request, build_opener

ROOT = Path(__file__).resolve().parents[1]
INTENTS = ROOT / "docs" / "batches" / "2026-09-14" / "approval-create-intents.json"
RECEIPT = ROOT / "docs" / "evidence" / "2026-09-15-sitting-approval-creates.json"
POLICY = "workload-kv-read-informed-decision-sitting-requester-client"
KV = "platform/data/workloads/informed-decision/sitting-requester"
SIBLING = "platform/data/workloads/secrets-engine/approval-requester"
PARENT = "platform/metadata/workloads/informed-decision"
ISSUER = "https://kc.coulomb.social"
CLIENT_ID = "informed-decision-sitting-requester"


class NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, *args, **kwargs):
        return None


def http(url, *, body=None, headers=None):
    req = Request(url, data=body, headers=headers or {})
    try:
        with build_opener(ProxyHandler({}), NoRedirect()).open(req, timeout=20) as response:
            content = response.read(1048577)
            status = response.status
    except HTTPError as error:
        status = error.code
        content = error.read(1048577)
        error.close()
    if len(content) > 1048576:
        raise ValueError("response_too_large")
    if not content:
        return status, {}
    return status, json.loads(content)


def require_attended() -> None:
    if Path.home().parent.name != ".warden-attended-login" or os.getenv("BAO_TOKEN") or os.getenv("VAULT_TOKEN"):
        raise ValueError("attended_reader_required")


def postable_intents(data: dict) -> list[dict]:
    rows = [row for row in data.get("intents") or [] if row.get("create_client") == CLIENT_ID]
    if len(rows) != 7:
        raise ValueError("expected seven postable sitting intents")
    for row in rows:
        binding = row["binding"]
        if binding.get("actor") != "informed-decision":
            raise ValueError("binding.actor must be informed-decision")
        if set(binding) != {"action", "actor", "principal", "purpose", "target"}:
            raise ValueError("binding must be the five native fields")
    return rows


def _approval_origin_ok(origin: str | None) -> bool:
    if not origin:
        return False
    if origin.startswith("http://127.0.0.1:") and origin.count("/") == 2:
        return True
    return origin in {
        "http://approval-engine.approval-engine.svc.cluster.local:8080",
        "http://approval-engine.approval-engine.svc:8080",
    }


def dry_run() -> dict:
    data = json.loads(INTENTS.read_text())
    rows = postable_intents(data)
    return {
        "kind": "informed-decision-sitting-create-dry-run",
        "posted": False,
        "memo_ids": [row["memo_id"] for row in rows],
        "skipped": [row["memo_id"] for row in data["intents"] if row.get("create_client") != CLIENT_ID],
        "requester_client": CLIENT_ID,
        "human_control": True,
        "pdp_path": False,
    }


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--dry-run", action="store_true", help="list postable memos; no credentials, no POST")
    parser.add_argument("--approval-origin", help="approval-engine origin; or INFD_APPROVAL_ORIGIN")
    args = parser.parse_args()
    if args.dry_run:
        print(json.dumps(dry_run(), indent=2, ensure_ascii=False))
        return 0
    try:
        require_attended()
        origin = args.approval_origin or os.environ.get("INFD_APPROVAL_ORIGIN")
        if not _approval_origin_ok(origin):
            raise ValueError("approval_origin_required")
    except ValueError as exc:
        print(exc, file=sys.stderr)
        return 2
    if RECEIPT.exists():
        raise ValueError("receipt_already_exists")
    data = json.loads(INTENTS.read_text())
    rows = postable_intents(data)
    receipt = {
        "observed_at": datetime.now(timezone.utc).isoformat(),
        "status": "failed",
        "phase": "preflight",
        "requests": [],
        "credential_values_emitted": False,
        "human_entries_created": False,
        "skipped": [row["memo_id"] for row in data["intents"] if row.get("create_client") != CLIENT_ID],
    }
    try:
        _create(rows, origin.rstrip("/"), receipt)
    finally:
        RECEIPT.write_text(json.dumps(receipt, indent=2) + "\n")
        print(json.dumps({"status": receipt["status"], "phase": receipt["phase"], "count": len(receipt["requests"])}))
    return 0 if receipt["status"] == "created" else 2


def _create(rows, origin, receipt):
    import jwt
    import subprocess

    def bao(*args):
        p = subprocess.run(["bao", *args], capture_output=True, text=True, timeout=20)
        if p.returncode:
            raise ValueError("metadata_failed")
        return json.loads(p.stdout)

    def capabilities(result):
        if isinstance(result, dict):
            result = result.get("data", result).get("capabilities", result)
        return result if isinstance(result, list) else []

    def is_deny(result):
        caps = set(capabilities(result))
        return caps <= {"deny"} or not caps.intersection({"read", "create", "update", "delete", "list", "patch", "sudo"})

    lookup = bao("token", "lookup", "-format=json")["data"]
    policies = set(lookup.get("policies", [])) | set(lookup.get("identity_policies", []))
    if POLICY not in policies or policies - {POLICY, "default"} or not lookup.get("entity_id") or not 0 < int(lookup.get("ttl") or 0) <= 900:
        raise ValueError("reader_identity_failed")
    if capabilities(bao("token", "capabilities", "-format=json", KV)) != ["read"]:
        raise ValueError("reader_scope_failed")
    if not is_deny(bao("token", "capabilities", "-format=json", SIBLING)):
        raise ValueError("reader_scope_failed")
    if not is_deny(bao("token", "capabilities", "-format=json", PARENT)):
        raise ValueError("reader_scope_failed")
    helper = Path.home() / ".vault-token"
    info = helper.lstat()
    if not stat.S_ISREG(info.st_mode) or stat.S_IMODE(info.st_mode) != 0o600 or info.st_uid != os.getuid():
        raise ValueError("private_helper_required")
    payload = bao("read", "-format=json", KV)
    body = payload.get("data", payload)
    inner = body.get("data", body)
    secret = inner.get("CLIENT_SECRET") if isinstance(inner, dict) else None
    if (body.get("metadata") or {}).get("version") != 1 or not secret:
        raise ValueError("requester_delivery_failed")

    def exchange(scope, credential=secret):
        auth = __import__("base64").b64encode((CLIENT_ID + ":" + credential).encode()).decode()
        return http(
            ISSUER + "/token",
            body=urlencode({"grant_type": "client_credentials", "scope": scope}).encode(),
            headers={"Authorization": "Basic " + auth, "Content-Type": "application/x-www-form-urlencoded"},
        )

    status, tokens = exchange("approval:create")
    if status != 200 or "access_token" not in tokens:
        raise ValueError("requester_exchange_failed")
    token = tokens["access_token"]
    status, jwks = http(ISSUER + "/jwks")
    if status != 200:
        raise ValueError("jwks_failed")
    header = jwt.get_unverified_header(token)
    keys = [key for key in jwks["keys"] if key["kid"] == header.get("kid")]
    if header.get("alg") != "RS256" or len(keys) != 1:
        raise ValueError("signing_key_failed")
    claims = jwt.decode(
        token,
        jwt.PyJWK.from_dict(keys[0]).key,
        algorithms=["RS256"],
        issuer=ISSUER,
        audience="approval-engine",
        options={"strict_aud": True, "require": ["sub", "iat", "exp", "iss", "aud"]},
    )
    roles = claims.get("roles")
    if isinstance(roles, str):
        roles = [roles]
    expected = {
        "sub": "informed-decision",
        "tenant": "tenant:platform",
        "principal_type": "service",
        "scope": "approval:create",
    }
    if any(claims.get(k) != v for k, v in expected.items()) or roles != ["informed-decision-sitting-requester"]:
        raise ValueError("requester_claims_failed")
    if int(claims["exp"]) - int(claims["iat"]) != 900:
        raise ValueError("requester_claims_failed")
    for scope in ("approval:approve", "approval:consume", "approval:read"):
        if exchange(scope)[0] != 400:
            raise ValueError("excess_scope_not_refused")
    del secret
    receipt["phase"] = "requester_verified"
    now = datetime.now(timezone.utc)
    for row in rows:
        body = {
            "binding": row["binding"],
            "validity": {"not_before": now.isoformat(), "expires_at": (now + timedelta(hours=24)).isoformat()},
            "required_count": 1,
            "human_control": True,
            "pdp_path": False,
        }
        receipt["phase"] = "create_attempt:" + row["memo_id"]
        RECEIPT.write_text(json.dumps(receipt, indent=2) + "\n")
        status, result = http(
            origin + "/v1/approvals",
            body=json.dumps(body).encode(),
            headers={"Authorization": "Bearer " + token, "Content-Type": "application/json"},
        )
        binding = (result or {}).get("binding") or {}
        if (
            status != 201
            or result.get("status") != "requested"
            or result.get("entries")
            or binding.get("human_control") is not True
            or not isinstance(binding.get("digest"), str)
        ):
            raise ValueError("request_creation_requires_reconciliation")
        receipt["requests"].append({"memo_id": row["memo_id"], "approval": result})
    receipt.update(status="created", phase="seven_unapproved_requests_created")


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