railiance-platform/scripts/issuer_binding_lane.py

"""broker-issuer-binding CCRs: an attended OIDC role that carries only a broker issuer policy.

RPF-WP-0054-T05. Loaded by scripts/credential-change.py for request_type
broker-issuer-binding. The role lets a human operator (group + aal2 on the
existing netkingdom OIDC mount) hold exactly one issuer policy, so their own
session can mint the grant token that ``credential.py generate`` uses. The role
rules (callbacks, scopes, claims, TTLs, no default policy) are the operator-matrix
lane's; the issuer policy must already be live from its own CCR. Nothing here
reads, prints or accepts a secret value; the mount is never enabled here.
"""
from __future__ import annotations

import importlib.util
import json
import re
import tempfile
from pathlib import Path
from typing import Any

REQUEST_TYPE = "broker-issuer-binding"
ISSUER_POLICY_RE = re.compile(r"^credential-broker-[a-z0-9-]+-issuer$")
ROLE_RE = re.compile(r"^[a-z0-9][a-z0-9-]*$")


def _load(name: str):
    spec = importlib.util.spec_from_file_location(name, Path(__file__).resolve().parent / f"{name}.py")
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


_matrix = _load("operator_matrix_lane")
BaoError = _matrix.BaoError
normalized = _matrix.normalized
_run = _matrix._run
_read_json = _matrix._read_json
ALLOWED_AUTH_MOUNTS = _matrix.ALLOWED_AUTH_MOUNTS
DURATION_KEYS = _matrix.DURATION_KEYS


def role_payload(ccr: dict[str, Any], repo_dir: Path) -> dict[str, Any]:
    return _matrix.role_payload(ccr, repo_dir)


def validate(ccr: dict[str, Any], repo_dir: Path) -> list[str]:
    errors: list[str] = []
    openbao = ccr.get("openbao")
    if not isinstance(openbao, dict) or not isinstance(openbao.get("auth"), dict):
        return ["openbao.auth must be an object"]
    auth = openbao["auth"]
    if (ccr.get("target") or {}).get("environment") != "production":
        errors.append("target.environment must be production")
    review = ccr.get("review") or {}
    if review.get("required") is not True or not review.get("required_approvers"):
        errors.append("review.required must be true with required_approvers")
    issuer = str(openbao.get("issuer_policy") or "")
    if not ISSUER_POLICY_RE.match(issuer):
        errors.append("openbao.issuer_policy must be credential-broker-<...>-issuer")
    elif not (repo_dir / f"openbao/policies/{issuer}.hcl").exists():
        errors.append(f"openbao.issuer_policy has no source file: openbao/policies/{issuer}.hcl")
    if auth.get("method") != "oidc":
        errors.append("openbao.auth.method must be oidc")
    if auth.get("mount") not in ALLOWED_AUTH_MOUNTS:
        errors.append(f"openbao.auth.mount must be one of {sorted(ALLOWED_AUTH_MOUNTS)}")
    if not ROLE_RE.match(str(auth.get("role") or "")):
        errors.append("openbao.auth.role contains unsafe characters")
    if not isinstance(auth.get("bound_claims_confirmed"), bool):
        errors.append("openbao.auth.bound_claims_confirmed must be a boolean")
    role = _matrix._shared._load_json(repo_dir, str(auth.get("role_file") or ""), "openbao.auth.role_file", errors)
    if role:
        _matrix._validate_role(role, issuer, errors)
        # OpenBao refuses child-token creation from a limited-use token, and
        # minting the grant token is this role's whole purpose (probe 2026-10-10).
        if role.get("token_num_uses") != 0:
            errors.append("role token_num_uses must be 0: limited-use tokens cannot create child tokens")
    return errors


def readiness_blockers(ccr: dict[str, Any], repo_dir: Path, allowed_statuses: set[str]) -> list[str]:
    blockers: list[str] = []
    if ccr.get("status") not in allowed_statuses:
        blockers.append(
            "applier requires approved, applied, verified, or active CCR status, "
            f"got {ccr.get('status')}"
        )
    if ccr["openbao"]["auth"].get("bound_claims_confirmed") is not True:
        blockers.append("applier requires confirmed OpenBao auth binding")
    blockers.extend(validate(ccr, repo_dir))
    return blockers


def dry_run_payload(ccr: dict[str, Any], repo_dir: Path, warnings: list[str]) -> dict[str, Any]:
    openbao = ccr["openbao"]
    auth = openbao["auth"]
    return {
        "id": ccr["id"],
        "title": ccr["title"],
        "status": ccr["status"],
        "environment": ccr["target"]["environment"],
        "warnings": warnings,
        "source_artifacts": {},
        "mutations": [
            {"kind": "issuer_policy_present", "openbao_path": f"sys/policies/acl/{openbao['issuer_policy']}",
             "payload": {"live_text_must_match": f"openbao/policies/{openbao['issuer_policy']}.hcl"}},
            {"kind": "auth_role_write", "openbao_path": f"auth/{auth['mount']}/role/{auth['role']}",
             "payload": role_payload(ccr, repo_dir),
             "precondition": f"auth/{auth['mount']} exists with type oidc (never enabled here)"},
            {"kind": "readback_compare", "openbao_path": "role", "payload": {"prints": "booleans only"}},
        ],
        "out_of_scope": [
            "policy writes (the issuer policy belongs to its grant CCR)",
            "secret value reads or writes",
            "token creation or login",
        ],
        "required_evidence": [
            "CCR id and approval reference",
            "readback equality for the role",
            "attended login with this role, then the RPF-WP-0054-T05 probe",
        ],
    }


def apply_details(ccr: dict[str, Any], actor: str) -> list[str]:
    openbao = ccr["openbao"]
    auth = openbao["auth"]
    return [
        f"Delegated metadata applier ran as {actor} using local bao CLI ambient authority.",
        f"Issuer policy confirmed live and matching source: sys/policies/acl/{openbao['issuer_policy']}",
        f"OIDC role metadata write: auth/{auth['mount']}/role/{auth['role']}",
        "Readback matched source for the role.",
        "No secret values were read, written, printed, or accepted in argv.",
    ]


def readback_mismatches(ccr: dict[str, Any], repo_dir: Path, bao_bin: str) -> list[str]:
    auth = ccr["openbao"]["auth"]
    mismatches: list[str] = []
    role = _read_json(bao_bin, ["read", f"auth/{auth['mount']}/role/{auth['role']}"], repo_dir).get("data", {})
    for key, want in role_payload(ccr, repo_dir).items():
        got = role.get(key)
        if key in DURATION_KEYS:
            same = _matrix.ttl_seconds(got) == _matrix.ttl_seconds(want)
        elif key in ("allowed_redirect_uris", "oidc_scopes"):
            same = sorted(got or []) == sorted(want)
        else:
            same = got == want
        if not same:
            mismatches.append(f"role {key} differs")
    return mismatches


def run_apply(ccr: dict[str, Any], repo_dir: Path, bao_bin: str, quiet: bool = False) -> None:
    openbao = ccr["openbao"]
    auth = openbao["auth"]
    mounts = _read_json(bao_bin, ["auth", "list"], repo_dir)
    mounts = mounts.get("data", mounts)
    if (mounts.get(f"{auth['mount']}/") or {}).get("type") != "oidc":
        raise BaoError(f"auth mount {auth['mount']}/ is absent or not oidc; this lane never enables it")
    live = _run(bao_bin, ["policy", "read", openbao["issuer_policy"]], repo_dir, quiet=True, capture=True)
    source = (repo_dir / f"openbao/policies/{openbao['issuer_policy']}.hcl").read_text(encoding="utf-8")
    if normalized(live) != normalized(source):
        raise BaoError("issuer policy is absent or differs from source; apply its grant CCR first")
    with tempfile.TemporaryDirectory() as tmp:
        role_path = Path(tmp) / "role.json"
        role_path.write_text(json.dumps(role_payload(ccr, repo_dir), sort_keys=True) + "\n")
        _run(bao_bin, ["write", f"auth/{auth['mount']}/role/{auth['role']}", f"@{role_path}"], repo_dir, quiet)
    mismatches = readback_mismatches(ccr, repo_dir, bao_bin)
    if mismatches:
        raise BaoError("readback mismatch: " + "; ".join(mismatches))