railiance-platform/scripts/broker_grant_lane.py

"""custody-generate-grant CCRs: a create-only KV grant the credential broker mints.

RPF-WP-0054-T04. Loaded by scripts/credential-change.py for request_type
custody-generate-grant. The lane writes three reviewed objects and nothing else:

* the grant policy: ``create`` on ``<prefix>/+`` data, ``read`` on ``<prefix>/+``
  metadata, explicit ``deny`` on reserved entries, ``revoke-self``. It has no
  ``update`` anywhere (an existing entry can never be overwritten), no data
  read, no metadata write (custom_metadata writes would also let the token
  change ``delete_version_after``/``max_versions`` of other entries), no list
  and no delete;
* the broker issuer policy: create tokens only through this grant's token role;
* the token role: exactly the grant policy, orphan, non-renewable, no default
  policy, at most 15 minutes.

Binding the issuer policy to the calling identity is a separate review once the
caller is known (RPF-WP-0054-T01); until then nothing can mint the token, so
the apply is inert. Nothing here reads, prints or accepts a secret value.
"""
from __future__ import annotations

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

import yaml

REQUEST_TYPE = "custody-generate-grant"
PREFIX_RE = re.compile(r"^platform/workloads/[a-z0-9-]+$")
ENTRY_RE = re.compile(r"^[a-z0-9][a-z0-9-]*$")
NAME_RE = re.compile(r"^[a-z0-9][a-z0-9-]*$")
CATALOG = "credential-grants/catalog.yaml"
MAX_TTL_SECONDS = 15 * 60
# Same list scripts/openbao-apply-token-grants.py writes; default is excluded by
# token_no_default_policy.
DISALLOWED_POLICIES = ["root", "platform-admin"]


def _service_jwt_lane():
    spec = importlib.util.spec_from_file_location(
        "service_jwt_lane", Path(__file__).resolve().parent / "service_jwt_lane.py"
    )
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


_shared = _service_jwt_lane()
ttl_seconds = _shared.ttl_seconds
normalized = _shared.normalized
BaoError = _shared.BaoError
_run = _shared._run
_read_json = _shared._read_json


def _paths(prefix: str) -> tuple[str, str]:
    mount, rest = prefix.split("/", 1)
    return f"{mount}/data/{rest}", f"{mount}/metadata/{rest}"


def generated_policy_hcl(ccr: dict[str, Any]) -> str:
    openbao = ccr["openbao"]
    data, meta = _paths(openbao["kv_prefix"])
    lines = [
        f"# {ccr['id']} (RPF-WP-0054): grant {openbao['grant_id']}. Create-only: no update,",
        "# data read, metadata write, list or delete. Exact deny beats the + glob.",
        f'path "{data}/+" {{', '  capabilities = ["create"]', "}",
        f'path "{meta}/+" {{', '  capabilities = ["read"]', "}",
    ]
    for entry in openbao["reserved_entries"]:
        for base in (data, meta):
            lines += [f'path "{base}/{entry}" {{', '  capabilities = ["deny"]', "}"]
    lines += ['path "auth/token/revoke-self" {', '  capabilities = ["update"]', "}"]
    return "\n".join(lines) + "\n"


def generated_issuer_hcl(ccr: dict[str, Any]) -> str:
    openbao = ccr["openbao"]
    lines = [
        f"# {ccr['id']} (RPF-WP-0054): the credential broker may mint {openbao['grant_id']}",
        "# tokens only through this token role. Bind to the reviewed caller identity only.",
    ]
    for path, caps in (
        (f"auth/token/create/{openbao['token_role']}", '"create", "update"'),
        ("auth/token/lookup-accessor", '"update"'),
        ("auth/token/revoke-accessor", '"update"'),
        ("auth/token/lookup-self", '"read"'),
        ("sys/capabilities-self", '"update"'),
    ):
        lines += [f'path "{path}" {{', f"  capabilities = [{caps}]", "}"]
    return "\n".join(lines) + "\n"


def role_payload(ccr: dict[str, Any]) -> dict[str, Any]:
    openbao = ccr["openbao"]
    return {
        "allowed_policies": [openbao["policy_name"]],
        "disallowed_policies": DISALLOWED_POLICIES,
        "orphan": True,
        "renewable": False,
        "token_explicit_max_ttl": openbao["token_ttl"],
        "token_no_default_policy": True,
        "token_type": "service",
    }


def _catalog_grant(repo_dir: Path, grant_id: str) -> dict[str, Any] | None:
    catalog = yaml.safe_load((repo_dir / CATALOG).read_text(encoding="utf-8")) or {}
    for grant in catalog.get("grants") or []:
        if grant.get("id") == grant_id:
            return grant
    return None


def _check_file(repo_dir: Path, rel: str, expected: str, field: str, errors: list[str]) -> None:
    path = repo_dir / rel
    if not path.exists():
        errors.append(f"{field} does not exist: {rel}")
    elif normalized(path.read_text(encoding="utf-8")) != normalized(expected):
        errors.append(f"{field} does not match the generated policy")


def validate(ccr: dict[str, Any], repo_dir: Path) -> list[str]:
    errors: list[str] = []
    openbao = ccr.get("openbao")
    if not isinstance(openbao, dict):
        return ["openbao must be an object"]
    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")
    prefix = str(openbao.get("kv_prefix") or "")
    if not PREFIX_RE.match(prefix):
        errors.append("openbao.kv_prefix must be platform/workloads/<owner>")
    reserved = openbao.get("reserved_entries")
    if not isinstance(reserved, list) or not all(ENTRY_RE.match(str(e)) for e in reserved):
        errors.append("openbao.reserved_entries must list exact entry names")
    for key in ("policy_name", "issuer_policy", "token_role"):
        if not NAME_RE.match(str(openbao.get(key) or "")):
            errors.append(f"openbao.{key} contains unsafe characters")
    seconds = ttl_seconds(openbao.get("token_ttl"))
    if seconds is None or not 0 < seconds <= MAX_TTL_SECONDS:
        errors.append("openbao.token_ttl must be set and at most 15m")
    if errors:
        return errors

    for key, file_key, expected in (
        ("policy_name", "policy_file", generated_policy_hcl(ccr)),
        ("issuer_policy", "issuer_policy_file", generated_issuer_hcl(ccr)),
    ):
        rel = str(openbao.get(file_key) or "")
        if rel != f"openbao/policies/{openbao[key]}.hcl":
            errors.append(f"openbao.{file_key} must be openbao/policies/<{key}>.hcl")
        else:
            _check_file(repo_dir, rel, expected, f"openbao.{file_key}", errors)

    grant = _catalog_grant(repo_dir, str(openbao.get("grant_id") or ""))
    if grant is None:
        errors.append(f"openbao.grant_id not in {CATALOG}")
    else:
        gb = grant.get("openbao") or {}
        if gb.get("policies") != [openbao["policy_name"]]:
            errors.append("catalog grant policies must be exactly [openbao.policy_name]")
        if gb.get("token_role") != openbao["token_role"] or gb.get("issuer_policy") != openbao["issuer_policy"]:
            errors.append("catalog grant token_role/issuer_policy must match the CCR")
        if ttl_seconds((grant.get("ttl") or {}).get("max")) != seconds:
            errors.append("catalog grant ttl.max must equal openbao.token_ttl")
    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')}"
        )
    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"]
    return {
        "id": ccr["id"],
        "title": ccr["title"],
        "status": ccr["status"],
        "environment": ccr["target"]["environment"],
        "warnings": warnings,
        "source_artifacts": {},
        "mutations": [
            {"kind": "policy_write", "openbao_path": f"sys/policies/acl/{openbao['policy_name']}",
             "policy_name": openbao["policy_name"], "source": openbao["policy_file"],
             "body": generated_policy_hcl(ccr).rstrip()},
            {"kind": "policy_write", "openbao_path": f"sys/policies/acl/{openbao['issuer_policy']}",
             "policy_name": openbao["issuer_policy"], "source": openbao["issuer_policy_file"],
             "body": generated_issuer_hcl(ccr).rstrip()},
            {"kind": "token_role_write", "openbao_path": f"auth/token/roles/{openbao['token_role']}",
             "payload": role_payload(ccr)},
            {"kind": "readback_compare", "openbao_path": "both policies, token role",
             "payload": {"prints": "booleans only"}},
        ],
        "out_of_scope": [
            "secret value reads or writes",
            "binding the issuer policy to any identity (separate review after RPF-WP-0054-T01)",
            "token creation",
            "changes to any other policy or role",
        ],
        "required_evidence": [
            "CCR id and approval reference",
            "readback equality for both policies and the token role",
            "disposable probe (RPF-WP-0054-T05): create once, second create and data read denied",
        ],
    }


def apply_details(ccr: dict[str, Any], actor: str) -> list[str]:
    openbao = ccr["openbao"]
    return [
        f"Delegated metadata applier ran as {actor} using local bao CLI ambient authority.",
        f"Policy metadata write: sys/policies/acl/{openbao['policy_name']}",
        f"Issuer policy metadata write: sys/policies/acl/{openbao['issuer_policy']}",
        f"Token role metadata write: auth/token/roles/{openbao['token_role']}",
        "Readback matched source for both policies and the token role.",
        "Issuer policy is not bound to any identity by this apply.",
        "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"]
    mismatches: list[str] = []
    for name, expected in (
        (openbao["policy_name"], generated_policy_hcl(ccr)),
        (openbao["issuer_policy"], generated_issuer_hcl(ccr)),
    ):
        text = _run(bao_bin, ["policy", "read", name], repo_dir, quiet=True, capture=True)
        if normalized(text) != normalized(expected):
            mismatches.append(f"policy {name} differs from source")
    role = _read_json(bao_bin, ["read", f"auth/token/roles/{openbao['token_role']}"], repo_dir).get("data", {})
    for key, want in role_payload(ccr).items():
        got = role.get(key)
        if key == "token_explicit_max_ttl":
            same = ttl_seconds(got) == ttl_seconds(want)
        elif key == "disallowed_policies":
            same = sorted(got or []) == sorted(want)
        else:
            same = got == want
        if not same:
            mismatches.append(f"token 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"]
    _run(bao_bin, ["policy", "write", openbao["policy_name"], openbao["policy_file"]], repo_dir, quiet)
    _run(bao_bin, ["policy", "write", openbao["issuer_policy"], openbao["issuer_policy_file"]], repo_dir, quiet)
    role = role_payload(ccr)
    args = ["write", f"auth/token/roles/{openbao['token_role']}"]
    for key, value in role.items():
        rendered = ",".join(value) if isinstance(value, list) else str(value).lower() if isinstance(value, bool) else value
        args.append(f"{key}={rendered}")
    _run(bao_bin, args, repo_dir, quiet)
    mismatches = readback_mismatches(ccr, repo_dir, bao_bin)
    if mismatches:
        raise BaoError("readback mismatch: " + "; ".join(mismatches))