railiance-platform/scripts/auth_role_watch_lane.py

"""auth-role-presence-watch CCRs: a workload may watch whether one Kubernetes-auth role exists.

net-kingdom NK-WP-0058-T06 (contract row O5). Loaded by scripts/credential-change.py for request_type
auth-role-presence-watch. A watcher ServiceAccount logs in with its own Kubernetes-auth role and may read exactly one
other role's path, so it can see that a window-scoped role (CCR-2026-0058) outlived its run. Mode `flag` grants read;
mode `close` adds delete on that same path, which only ever reduces privilege. Nothing else: no KV path, no other
auth path, no list, no sys. Nothing here reads, prints or accepts a secret value.
"""
from __future__ import annotations

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

REQUEST_TYPE = "auth-role-presence-watch"
ROLE_RE = re.compile(r"^[a-z0-9][a-z0-9-]*$")
NAME_RE = re.compile(r"^[a-z0-9][a-z0-9-]*$")
MODES = {"flag": ["read"], "close": ["read", "delete"]}
SELF_GRANTS = [("auth/token/lookup-self", "read"), ("auth/token/revoke-self", "update")]
DURATIONS = ("ttl", "token_max_ttl", "token_explicit_max_ttl")


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


_shared = _load("service_jwt_lane")
BaoError = _shared.BaoError
normalized = _shared.normalized
ttl_seconds = _shared.ttl_seconds
_run = _shared._run
_read_json = _shared._read_json


def watched_path(ccr: dict[str, Any]) -> str:
    watch = ccr["openbao"]["watch"]
    return f"auth/{watch['mount']}/role/{watch['role']}"


def generated_policy_hcl(ccr: dict[str, Any]) -> str:
    openbao = ccr["openbao"]
    caps = ", ".join(f'"{c}"' for c in MODES[openbao["watch"]["mode"]])
    lines = [f"# {openbao['policy_name']}: {ccr['id']} ({REQUEST_TYPE}, mode {openbao['watch']['mode']}).",
             "# Platform-generated from the CCR by scripts/auth_role_watch_lane.py; do not edit by hand.", "",
             f'path "{watched_path(ccr)}" {{', f"  capabilities = [{caps}]", "}"]
    for path, capability in SELF_GRANTS:
        lines += ["", f'path "{path}" {{', f'  capabilities = ["{capability}"]', "}"]
    return "\n".join(lines) + "\n"


def role_payload(ccr: dict[str, Any]) -> dict[str, Any]:
    auth = ccr["openbao"]["auth"]
    return {"bound_service_account_names": list(auth["bound_claims"]["service_account_names"]),
            "bound_service_account_namespaces": list(auth["bound_claims"]["service_account_namespaces"]),
            "token_policies": [ccr["openbao"]["policy_name"]], "audience": auth["audience"],
            "token_ttl": auth["ttl"], "token_max_ttl": auth["token_max_ttl"],
            "token_explicit_max_ttl": auth["token_explicit_max_ttl"], "token_no_default_policy": True}


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) \
            or not isinstance(openbao.get("watch"), dict):
        return ["openbao.auth and openbao.watch must be objects"]
    auth, watch = openbao["auth"], openbao["watch"]
    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")
    if watch.get("mode") not in MODES:
        errors.append(f"openbao.watch.mode must be one of {sorted(MODES)}")
    if watch.get("mount") != "kubernetes" or not ROLE_RE.match(str(watch.get("role") or "")):
        errors.append("openbao.watch must name one role on the kubernetes auth mount")
    if not NAME_RE.match(str(openbao.get("policy_name") or "")):
        errors.append("openbao.policy_name contains unsafe characters")
    if openbao.get("policy_file") != f"openbao/policies/{openbao.get('policy_name')}.hcl":
        errors.append("openbao.policy_file must be openbao/policies/<policy_name>.hcl")
    if auth.get("method") != "kubernetes" or auth.get("mount") != "kubernetes":
        errors.append("openbao.auth must be the kubernetes method on mount kubernetes")
    if not ROLE_RE.match(str(auth.get("role") or "")) or auth.get("role") == watch.get("role"):
        errors.append("openbao.auth.role must be safe and differ from the watched role")
    if auth.get("audience") != "openbao":
        errors.append("openbao.auth.audience must be openbao")
    if any(ttl_seconds(auth.get(k)) is None or ttl_seconds(auth.get(k)) > 900 for k in DURATIONS):
        errors.append("openbao.auth ttl, token_max_ttl and token_explicit_max_ttl must be at most 15m")
    if auth.get("token_no_default_policy") is not True:
        errors.append("openbao.auth.token_no_default_policy must be true")
    if auth.get("policies") != [openbao.get("policy_name")]:
        errors.append("openbao.auth.policies must be exactly [policy_name]")
    claims = auth.get("bound_claims") or {}
    names, namespaces = claims.get("service_account_names"), claims.get("service_account_namespaces")
    if not (isinstance(names, list) and len(names) == 1 and isinstance(namespaces, list) and len(namespaces) == 1
            and set(claims) == {"service_account_names", "service_account_namespaces"}):
        errors.append("openbao.auth.bound_claims must name exactly one ServiceAccount in one namespace")
    if not isinstance(auth.get("bound_claims_confirmed"), bool):
        errors.append("openbao.auth.bound_claims_confirmed must be a boolean")
    if errors:
        return errors
    policy = repo_dir / openbao["policy_file"]
    if not policy.exists():
        errors.append(f"openbao.policy_file does not exist: {openbao['policy_file']}")
    elif normalized(policy.read_text(encoding="utf-8")) != normalized(generated_policy_hcl(ccr)):
        errors.append("openbao.policy_file differs from the policy generated from the CCR")
    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.get("openbao") or {}).get("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": {"policy_file": openbao["policy_file"]},
        "mutations": [
            {"kind": "policy_write", "openbao_path": f"sys/policies/acl/{openbao['policy_name']}",
             "payload": {"source": openbao["policy_file"], "watched_path": watched_path(ccr),
                         "capabilities": MODES[openbao["watch"]["mode"]]}},
            {"kind": "auth_role_write", "openbao_path": f"auth/{auth['mount']}/role/{auth['role']}",
             "payload": role_payload(ccr), "precondition": "auth/kubernetes exists (never enabled here)"},
            {"kind": "readback_compare", "openbao_path": "policy and role", "payload": {"prints": "booleans only"}},
        ],
        "out_of_scope": [
            "the watched role itself (CCR-2026-0058 window-open and window-close)",
            "secret value reads or writes",
            "token creation or login",
        ],
        "required_evidence": [
            "CCR id and approval reference",
            "readback equality for policy and role",
            "the watch CronJob's first run in mode " + openbao["watch"]["mode"],
        ],
    }


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"Policy write: sys/policies/acl/{openbao['policy_name']} ({openbao['watch']['mode']} on {watched_path(ccr)})",
        f"Kubernetes auth role write: auth/{auth['mount']}/role/{auth['role']}",
        "Readback matched source for policy and 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]:
    openbao = ccr["openbao"]
    auth = openbao["auth"]
    mismatches: list[str] = []
    live = _run(bao_bin, ["policy", "read", openbao["policy_name"]], repo_dir, quiet=True, capture=True)
    if normalized(live) != normalized(generated_policy_hcl(ccr)):
        mismatches.append("policy differs")
    role = _read_json(bao_bin, ["read", f"auth/{auth['mount']}/role/{auth['role']}"], repo_dir).get("data", {})
    for key, want in role_payload(ccr).items():
        got = role.get(key)
        same = ttl_seconds(got) == ttl_seconds(want) if key.startswith("token_") and key.endswith("ttl") else 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("kubernetes/") or {}).get("type") != "kubernetes":
        raise BaoError("auth mount kubernetes/ is absent or not kubernetes; this lane never enables it")
    _run(bao_bin, ["policy", "write", openbao["policy_name"], openbao["policy_file"]], repo_dir, quiet)
    with tempfile.TemporaryDirectory() as tmp:
        role_path = Path(tmp) / "role.json"
        role_path.write_text(json.dumps(role_payload(ccr), 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))