RPF-WP-0054-T05: issuer binding CCR-2026-0043 (net-kingdom-admins, aal2) and probe session

New CCR request type broker-issuer-binding: an OIDC role on netkingdom that
carries only the broker issuer policy (loopback callbacks, 15m, 8 uses, no
default policy); the apply refuses unless the live issuer policy matches
source. Founder chose net-kingdom-admins + aal2 on 2026-10-10.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 469992@bnt-lap001
Assistant-Session: e4b8e420-df0a-441d-83ae-7f19130594d9

scripts/credential-change.py

@@ -824,6 +824,7 @@ LANE_MODULES = {
     "service-jwt-login": "service_jwt_lane",
     "workload-kv-operator-matrix": "operator_matrix_lane",
     "custody-generate-grant": "broker_grant_lane",
+    "broker-issuer-binding": "issuer_binding_lane",
 }
 
 

scripts/issuer_binding_lane.py

@@ -0,0 +1,165 @@
+"""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)
+    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))