#!/usr/bin/env python3
from __future__ import annotations

import argparse
import getpass
import hashlib
import json
import os
import re
import shlex
import subprocess
import sys
import urllib.error
import urllib.request
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any

import yaml

REPO_DIR = Path(__file__).resolve().parents[1]
DEFAULT_CATALOG = REPO_DIR / "credential-grants/catalog.yaml"
DEFAULT_LEASE_DIR = REPO_DIR / ".local/credential-leases"
TOKEN_MARKERS = re.compile(
    r"\bhv[bcms]\.[A-Za-z0-9._-]+|\b(?:VAULT_TOKEN|BAO_TOKEN|OPENBAO_ROOT_TOKEN)=[^\s]+"
)
ENV_ASSIGNMENT = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*=")
UNSAFE_TOKEN_ENV = {"VAULT_TOKEN", "BAO_TOKEN", "OPENBAO_ROOT_TOKEN"}
UNSAFE_VERBOSE_VALUES = {"debug", "trace"}
DEFAULT_ACTOR_TYPE = "approved-agent"
DEFAULT_ACTOR = f"codex:{os.environ.get('USER', 'unknown')}"
DEFAULT_SUBJECT = "agent:codex/railiance-platform"
FLEX_AUTH_SUBJECT_TYPES = {
    "human-operator": "Human",
    "approved-agent": "Agent",
    "ci-runner": "Automation",
}


@dataclass(frozen=True)
class AuthorizationResult:
    allowed: bool
    mode: str
    decision_id: str | None = None
    reason: str | None = None


def emit_json(payload: dict[str, Any]) -> None:
    print(json.dumps(payload, indent=2, sort_keys=True))


def fail(message: str) -> None:
    raise SystemExit(f"ERROR: {message}")


def warn(message: str) -> None:
    print(f"WARN: {message}", file=sys.stderr)


def redact(text: str, extra_secrets: list[str] | None = None) -> str:
    redacted = TOKEN_MARKERS.sub("[REDACTED]", text)
    for secret in extra_secrets or []:
        if secret:
            redacted = redacted.replace(secret, "[REDACTED]")
    return redacted


def guard_verbose_env() -> None:
    for name in ("BAO_LOG_LEVEL", "VAULT_LOG_LEVEL"):
        value = os.environ.get(name, "").lower()
        if value in UNSAFE_VERBOSE_VALUES:
            fail(
                f"refusing to run with {name}={value}; it may print secret-bearing traces"
            )


def read_issuer_token(
    token_file: str | None, dry_run: bool, use_token_helper: bool
) -> str | None:
    if dry_run or use_token_helper:
        return None
    if token_file:
        path = Path(token_file).expanduser()
        if not path.exists():
            fail(f"OPENBAO_TOKEN_FILE does not exist: {path}")
        lines = path.read_text(encoding="utf-8").splitlines()
        token = lines[0].strip() if lines else ""
        if not token:
            fail(f"OPENBAO_TOKEN_FILE is empty: {path}")
        return token
    token = getpass.getpass("OpenBao issuer token: ")
    if not token:
        fail("empty OpenBao issuer token")
    return token


def ttl_seconds(value: str) -> int:
    match = re.match(r"^([1-9][0-9]*)([smhd])$", value)
    if not match:
        fail(f"TTL must match <positive integer><s|m|h|d>: {value!r}")
    amount = int(match.group(1))
    multiplier = {"s": 1, "m": 60, "h": 3600, "d": 86400}[match.group(2)]
    return amount * multiplier


def expires_at(ttl: str) -> str:
    return (
        datetime.now(timezone.utc) + timedelta(seconds=ttl_seconds(ttl))
    ).isoformat()


def resolve_repo_path(path: Path) -> Path:
    path = path.expanduser()
    if path.is_absolute():
        return path
    return (REPO_DIR / path).resolve()


def load_catalog(path: Path) -> dict[str, Any]:
    data = yaml.safe_load(path.read_text(encoding="utf-8"))
    if not isinstance(data, dict):
        fail(f"catalog root must be an object: {path}")
    return data


def get_grant(catalog: dict[str, Any], grant_id: str) -> dict[str, Any]:
    grants = catalog.get("grants") or []
    if not isinstance(grants, list):
        fail("catalog grants must be a list")
    for grant in grants:
        if isinstance(grant, dict) and grant.get("id") == grant_id:
            if grant.get("credential_type") not in {
                "openbao-token",
                "openbao-database-credential",
            }:
                fail(
                    f"unsupported credential_type for {grant_id}: {grant.get('credential_type')}"
                )
            return grant
    fail(f"grant not found in catalog: {grant_id}")


def validate_issue_request(
    grant: dict[str, Any],
    ttl: str,
    purpose: str,
    delivery: str | None = None,
    actor_type: str | None = None,
) -> None:
    if not purpose.strip():
        fail("--purpose is required and must not be empty")
    ttl_cfg = grant.get("ttl") or {}
    max_ttl = ttl_cfg.get("max")
    if not isinstance(max_ttl, str):
        fail(f"grant {grant.get('id')} has no ttl.max")
    if ttl_seconds(ttl) > ttl_seconds(max_ttl):
        fail(f"requested TTL {ttl} exceeds grant max TTL {max_ttl}")
    if delivery is not None:
        allowed = set((grant.get("delivery") or {}).get("allowed") or [])
        if delivery not in allowed:
            fail(
                f"delivery mode {delivery!r} is not allowed for grant {grant.get('id')}"
            )
    if actor_type:
        allowed_actor_types = set(
            (grant.get("actors") or {}).get("allowed_types") or []
        )
        if actor_type not in allowed_actor_types:
            fail(
                f"actor type {actor_type!r} is not allowed for grant {grant.get('id')}"
            )


def safe_handle(handle: str) -> str:
    digest = hashlib.sha256(handle.encode("utf-8")).hexdigest()[:16]
    cleaned = re.sub(r"[^A-Za-z0-9_.-]", "_", handle)[:64]
    return f"{cleaned}-{digest}"


def bool_env(name: str) -> bool:
    return os.environ.get(name, "").lower() in {"1", "true", "yes", "on"}


def join_url(base: str, path: str) -> str:
    return base.rstrip("/") + "/" + path.lstrip("/")


def post_json(
    url: str, payload: dict[str, Any], timeout: float = 10.0
) -> dict[str, Any]:
    body = json.dumps(payload).encode("utf-8")
    request = urllib.request.Request(
        url,
        data=body,
        headers={"Content-Type": "application/json", "Accept": "application/json"},
        method="POST",
    )
    try:
        with urllib.request.urlopen(request, timeout=timeout) as response:
            raw = response.read().decode("utf-8")
    except urllib.error.HTTPError as exc:
        detail = exc.read().decode("utf-8", errors="replace")
        raise RuntimeError(f"HTTP {exc.code} from {url}: {detail}") from exc
    except urllib.error.URLError as exc:
        raise RuntimeError(f"could not reach {url}: {exc.reason}") from exc
    if not raw.strip():
        return {}
    data = json.loads(raw)
    if not isinstance(data, dict):
        raise RuntimeError(f"expected JSON object from {url}")
    return data


def flex_auth_check_request(
    *,
    grant: dict[str, Any],
    ttl: str,
    purpose: str,
    delivery: str,
    actor: str,
    actor_type: str,
    subject: str,
) -> dict[str, Any]:
    subject_type = FLEX_AUTH_SUBJECT_TYPES.get(actor_type)
    if subject_type is None:
        fail(f"actor type {actor_type!r} has no flex-auth subject type mapping")
    return {
        "tenant": "tenant:platform",
        "subject": {"id": subject, "type": subject_type},
        "action": "issue",
        "resource": {
            "id": f"credential-grant:{grant['id']}",
            "type": "credential-grant",
            "system": "railiance-platform",
        },
        "context": {
            "actor": actor,
            "actor_type": actor_type,
            "bound_subject": subject,
            "purpose": purpose,
            "delivery_mode": delivery,
            "requested_ttl_seconds": ttl_seconds(ttl),
        },
    }


def authorize_request(
    *,
    args: argparse.Namespace,
    grant: dict[str, Any],
    ttl: str,
    purpose: str,
    delivery: str,
) -> AuthorizationResult:
    validate_issue_request(grant, ttl, purpose, delivery, args.actor_type)
    if args.decision_id:
        return AuthorizationResult(
            True,
            "provided-decision",
            args.decision_id,
            "caller supplied prior decision id",
        )
    if args.dry_run:
        return AuthorizationResult(
            True, "dry-run-local", None, "dry-run does not call flex-auth"
        )
    if not args.flex_auth_url:
        if args.require_flex_auth:
            fail(
                "--require-flex-auth was set, but no --flex-auth-url or FLEX_AUTH_URL is configured"
            )
        return AuthorizationResult(
            True,
            "local-preauthorized",
            None,
            "flex-auth unavailable; grant policy is locally preauthorized",
        )

    endpoint = join_url(args.flex_auth_url, args.flex_auth_path)
    payload = flex_auth_check_request(
        grant=grant,
        ttl=ttl,
        purpose=purpose,
        delivery=delivery,
        actor=args.actor,
        actor_type=args.actor_type,
        subject=args.subject,
    )
    try:
        response = post_json(endpoint, payload, timeout=args.http_timeout)
    except Exception as exc:  # noqa: BLE001
        if args.require_flex_auth:
            fail(f"flex-auth preflight failed: {exc}")
        warn(
            f"flex-auth preflight unavailable; continuing by local preauthorization: {exc}"
        )
        return AuthorizationResult(
            True,
            "local-preauthorized",
            None,
            "flex-auth unavailable; local preauthorization used",
        )

    effect = response.get("effect")
    decision_id = response.get("id")
    reason = response.get("reason")
    if effect != "allow":
        fail(f"flex-auth denied credential request: {reason or 'no reason supplied'}")
    if not isinstance(decision_id, str) or not decision_id:
        fail("flex-auth returned an allow effect without a decision id")
    response_subject = response.get("subject")
    if (
        not isinstance(response_subject, dict)
        or response_subject.get("id") != payload["subject"]["id"]
    ):
        fail("flex-auth allow decision subject does not match the request")
    response_resource = response.get("resource")
    if not isinstance(response_resource, dict) or any(
        response_resource.get(field) != payload["resource"][field]
        for field in ("id", "type", "system")
    ):
        fail("flex-auth allow decision resource does not match the request")
    provenance = response.get("provenance")
    if (
        not isinstance(provenance, dict)
        or not provenance.get("evaluator")
        or not provenance.get("mode")
    ):
        fail("flex-auth allow decision has no evaluator provenance")
    return AuthorizationResult(
        True,
        "flex-auth",
        decision_id,
        str(reason) if reason else None,
    )


def state_hub_enabled(args: argparse.Namespace) -> bool:
    return args.record_state_hub or bool_env("CREDENTIAL_RECORD_STATE_HUB")


def state_hub_metadata(
    *,
    args: argparse.Namespace,
    grant: dict[str, Any] | None,
    status: str,
    purpose: str | None = None,
    ttl: str | None = None,
    delivery: str | None = None,
    authz: AuthorizationResult | None = None,
    lease_handle: str | None = None,
    wrapping_accessor: str | None = None,
    wrapped_accessor: str | None = None,
    exit_code: int | None = None,
) -> dict[str, Any]:
    metadata: dict[str, Any] = {
        "status": status,
        "actor": args.actor,
        "actor_type": args.actor_type,
        "subject": args.subject,
    }
    if grant is not None:
        metadata.update(
            {
                "grant_id": grant.get("id"),
                "audience": grant.get("audience"),
                "issuer": grant.get("issuer"),
                "credential_type": grant.get("credential_type"),
            }
        )
    if purpose is not None:
        metadata["purpose"] = purpose
    if ttl is not None:
        metadata["requested_ttl"] = ttl
    if delivery is not None:
        metadata["delivery_mode"] = delivery
    if authz is not None:
        metadata["authorization_mode"] = authz.mode
        metadata["decision_id"] = authz.decision_id
        metadata["decision_reason"] = authz.reason
    if lease_handle is not None:
        metadata["lease_accessor"] = lease_handle
    if wrapping_accessor is not None:
        metadata["wrapping_accessor"] = wrapping_accessor
    if wrapped_accessor is not None:
        metadata["wrapped_accessor"] = wrapped_accessor
    if exit_code is not None:
        metadata["exit_code"] = exit_code
    return metadata


def record_state_hub(args: argparse.Namespace, metadata: dict[str, Any]) -> None:
    if not state_hub_enabled(args):
        return
    if args.dry_run:
        print("DRY-RUN: would record non-secret State Hub credential metadata")
        return
    url = args.state_hub_url.rstrip("/") + "/progress/"
    summary_parts = [
        "credential-broker",
        str(metadata.get("status")),
        f"grant={metadata.get('grant_id')}",
        f"actor={metadata.get('actor')}",
        f"purpose={metadata.get('purpose')}",
        f"ttl={metadata.get('requested_ttl')}",
        f"delivery={metadata.get('delivery_mode')}",
    ]
    if metadata.get("decision_id"):
        summary_parts.append(f"decision={metadata.get('decision_id')}")
    if metadata.get("lease_accessor"):
        summary_parts.append(f"lease={metadata.get('lease_accessor')}")
    payload = {
        "summary": " ".join(
            part for part in summary_parts if part and not part.endswith("=None")
        ),
        "detail": json.dumps(metadata, sort_keys=True),
        "event_type": "note",
        "author": "credential-broker",
    }
    scope_id = getattr(args, "state_hub_workplan_id", None) or args.state_hub_workstream_id
    if scope_id:
        payload["workplan_id"] = scope_id
    try:
        post_json(url, payload, timeout=args.http_timeout)
    except Exception as exc:  # noqa: BLE001
        warn(f"could not record State Hub metadata: {exc}")


class BaoRunner:
    def __init__(
        self,
        *,
        kubectl: str,
        namespace: str,
        release: str,
        dry_run: bool,
        use_token_helper: bool,
        issuer_token: str | None,
        local_bao: bool = False,
    ) -> None:
        self.kubectl_parts = shlex.split(kubectl)
        self.namespace = namespace
        self.pod = f"{release}-0"
        self.dry_run = dry_run
        self.use_token_helper = use_token_helper
        self.issuer_token = issuer_token
        self.local_bao = local_bao

    def run(
        self, args: list[str], *, input_text: str | None = None, quiet: bool = False
    ) -> subprocess.CompletedProcess[str]:
        if self.dry_run:
            print("DRY-RUN: bao " + shlex.join(args))
            return subprocess.CompletedProcess(args, 0, "", "")

        if self.local_bao:
            if not self.issuer_token:
                raise RuntimeError("issuer token is required for --local-bao")
            env = os.environ.copy()
            env["BAO_TOKEN"] = self.issuer_token
            env["VAULT_TOKEN"] = self.issuer_token
            result = subprocess.run(
                ["bao", *args], input=input_text, env=env,
                capture_output=True, text=True, check=False,
            )
            if result.returncode != 0:
                if result.stdout and not quiet:
                    print(redact(result.stdout, [self.issuer_token]), end="")
                if result.stderr:
                    print(redact(result.stderr, [self.issuer_token]), file=sys.stderr, end="")
                raise SystemExit(result.returncode)
            if result.stdout and not quiet:
                print(redact(result.stdout, [self.issuer_token]), end="")
            if result.stderr and not quiet:
                print(redact(result.stderr, [self.issuer_token]), file=sys.stderr, end="")
            return result

        if self.use_token_helper:
            cmd = (
                self.kubectl_parts
                + ["exec", "-i", "-n", self.namespace, self.pod, "--", "bao"]
                + args
            )
            proc_input = input_text
        else:
            if not self.issuer_token:
                raise RuntimeError(
                    "issuer token is required unless --use-token-helper is set"
                )
            cmd = (
                self.kubectl_parts
                + [
                    "exec",
                    "-i",
                    "-n",
                    self.namespace,
                    self.pod,
                    "--",
                    "sh",
                    "-c",
                    'read -r BAO_TOKEN; export BAO_TOKEN; export VAULT_TOKEN="$BAO_TOKEN"; exec bao "$@"',
                    "sh",
                ]
                + args
            )
            proc_input = self.issuer_token + "\n" + (input_text or "")

        result = subprocess.run(
            cmd, input=proc_input, capture_output=True, text=True, check=False
        )
        if result.returncode != 0:
            if result.stdout and not quiet:
                print(redact(result.stdout), end="")
            if result.stderr:
                print(redact(result.stderr), file=sys.stderr, end="")
            raise SystemExit(result.returncode)
        if result.stdout and not quiet:
            print(redact(result.stdout), end="")
        if result.stderr and not quiet:
            print(redact(result.stderr), file=sys.stderr, end="")
        return result

    def run_with_token(
        self, token: str, args: list[str], *, quiet: bool = False
    ) -> subprocess.CompletedProcess[str]:
        if self.dry_run:
            print("DRY-RUN: bao " + shlex.join(args))
            return subprocess.CompletedProcess(args, 0, "", "")
        if self.local_bao:
            env = os.environ.copy()
            env["BAO_TOKEN"] = token
            env["VAULT_TOKEN"] = token
            result = subprocess.run(
                ["bao", *args], env=env, capture_output=True, text=True, check=False
            )
            if result.returncode != 0:
                if result.stdout and not quiet:
                    print(redact(result.stdout, [token]), end="")
                if result.stderr:
                    print(redact(result.stderr, [token]), file=sys.stderr, end="")
                raise SystemExit(result.returncode)
            if result.stdout and not quiet:
                print(redact(result.stdout, [token]), end="")
            if result.stderr and not quiet:
                print(redact(result.stderr, [token]), file=sys.stderr, end="")
            return result
        cmd = (
            self.kubectl_parts
            + [
                "exec", "-i", "-n", self.namespace, self.pod, "--", "sh", "-c",
                'read -r BAO_TOKEN; export BAO_TOKEN; export VAULT_TOKEN="$BAO_TOKEN"; exec bao "$@"',
                "sh",
            ]
            + args
        )
        result = subprocess.run(
            cmd, input=token + "\n", capture_output=True, text=True, check=False
        )
        if result.returncode != 0:
            if result.stdout and not quiet:
                print(redact(result.stdout, [token]), end="")
            if result.stderr:
                print(redact(result.stderr, [token]), file=sys.stderr, end="")
            raise SystemExit(result.returncode)
        if result.stdout and not quiet:
            print(redact(result.stdout, [token]), end="")
        if result.stderr and not quiet:
            print(redact(result.stderr, [token]), file=sys.stderr, end="")
        return result


def token_create_args(
    grant: dict[str, Any], ttl: str, wrap_ttl: str | None = None
) -> list[str]:
    openbao = grant["openbao"]
    args = ["token", "create"]
    if wrap_ttl:
        args.append(f"-wrap-ttl={wrap_ttl}")
    args.extend([f"-role={openbao['token_role']}", f"-ttl={ttl}", "-format=json"])
    for policy in openbao["policies"]:
        args.append(f"-policy={policy}")
    return args


def parse_token_create(stdout: str) -> tuple[str, str]:
    try:
        payload = json.loads(stdout)
        auth = payload.get("auth") or payload.get("data") or {}
        token = auth["client_token"]
        accessor = auth["accessor"]
    except Exception as exc:  # noqa: BLE001
        raise SystemExit(
            f"ERROR: could not parse token create response: {exc}"
        ) from exc
    return token, accessor


def parse_database_credential(stdout: str) -> tuple[str, str, str]:
    try:
        payload = json.loads(stdout)
        return payload["lease_id"], payload["data"]["username"], payload["data"]["password"]
    except Exception as exc:  # noqa: BLE001
        raise SystemExit(
            f"ERROR: could not parse database credential response: {exc}"
        ) from exc


def parse_wrap_create(stdout: str) -> dict[str, Any]:
    try:
        payload = json.loads(stdout)
        wrap_info = payload["wrap_info"]
    except Exception as exc:  # noqa: BLE001
        raise SystemExit(
            f"ERROR: could not parse wrapped token response: {exc}"
        ) from exc
    return wrap_info


def write_mode_0600(path: Path, content: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.parent.chmod(0o700)
    flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC
    if hasattr(os, "O_NOFOLLOW"):
        flags |= os.O_NOFOLLOW
    fd = os.open(path, flags, 0o600)
    with os.fdopen(fd, "w", encoding="utf-8") as handle:
        handle.write(content)
    path.chmod(0o600)


def write_local_lease(
    *,
    lease_dir: Path,
    grant: dict[str, Any],
    purpose: str,
    ttl: str,
    token: str,
    accessor: str,
    authz: AuthorizationResult,
) -> dict[str, Any]:
    stem = safe_handle(accessor)
    token_path = lease_dir / f"{stem}.openbao-token"
    meta_path = lease_dir / f"{stem}.json"
    write_mode_0600(token_path, token + "\n")
    metadata = {
        "grant_id": grant["id"],
        "lease_handle": accessor,
        "accessor": accessor,
        "purpose": purpose,
        "issued_at": datetime.now(timezone.utc).isoformat(),
        "expires_at": expires_at(ttl),
        "ttl": ttl,
        "delivery_mode": "local-token-file",
        "authorization_mode": authz.mode,
        "decision_id": authz.decision_id,
        "token_file": str(token_path),
        "status": "issued",
    }
    write_mode_0600(meta_path, json.dumps(metadata, indent=2, sort_keys=True) + "\n")
    return {key: value for key, value in metadata.items() if key != "token_file"} | {
        "token_file": str(token_path),
        "metadata_file": str(meta_path),
    }


def find_local_metadata(
    lease_dir: Path, handle: str
) -> tuple[Path, dict[str, Any]] | None:
    if not lease_dir.exists():
        return None
    for path in lease_dir.glob("*.json"):
        try:
            data = json.loads(path.read_text(encoding="utf-8"))
        except json.JSONDecodeError:
            continue
        if data.get("lease_handle") == handle or data.get("accessor") == handle:
            return path, data
    return None


def path_is_within(path: Path, root: Path) -> bool:
    try:
        path.resolve().relative_to(root.resolve())
        return True
    except ValueError:
        return False


def remove_local_lease_files(lease_dir: Path, handle: str) -> list[str]:
    found = find_local_metadata(lease_dir, handle)
    if not found:
        return []
    meta_path, metadata = found
    lease_root = lease_dir.resolve()
    removed: list[str] = []
    for key in ("token_file",):
        value = metadata.get(key)
        if isinstance(value, str):
            path = Path(value)
            if not path.is_absolute():
                path = (REPO_DIR / path).resolve()
            if path.exists() and path.is_file():
                if not path_is_within(path, lease_root):
                    fail(
                        f"refusing to remove local lease file outside {lease_root}: {path}"
                    )
                path.unlink()
                removed.append(str(path))
    if meta_path.exists():
        if not path_is_within(meta_path, lease_root):
            fail(f"refusing to remove metadata file outside {lease_root}: {meta_path}")
        meta_path.unlink()
        removed.append(str(meta_path))
    return removed


def split_env_prefix(command: list[str]) -> tuple[dict[str, str], list[str]]:
    if command and command[0] == "--":
        command = command[1:]
    extra_env: dict[str, str] = {}
    index = 0
    while index < len(command) and ENV_ASSIGNMENT.match(command[index]):
        name, value = command[index].split("=", 1)
        if name in UNSAFE_TOKEN_ENV:
            fail(
                f"refusing caller-supplied {name}; the helper owns credential injection"
            )
        if TOKEN_MARKERS.search(value):
            fail(f"refusing secret-looking value in env assignment {name}")
        extra_env[name] = value
        index += 1
    program = command[index:]
    if not program:
        fail("missing child command after --")
    return extra_env, program


def kubernetes_auth_payload(
    grant: dict[str, Any], ttl: str, purpose: str, authz: AuthorizationResult
) -> dict[str, Any]:
    delivery = grant.get("delivery") or {}
    kubernetes_auth = delivery.get("kubernetes_auth") or {}
    return {
        "grant_id": grant["id"],
        "purpose": purpose,
        "ttl": ttl,
        "delivery_mode": "kubernetes-auth",
        "status": "delegated",
        "authorization_mode": authz.mode,
        "decision_id": authz.decision_id,
        "openbao_auth_mount": kubernetes_auth.get("mount", "auth/kubernetes"),
        "openbao_auth_role": kubernetes_auth.get("role")
        or grant["openbao"].get("kubernetes_auth_role"),
        "service_account_names": kubernetes_auth.get("service_account_names", []),
        "namespaces": kubernetes_auth.get("namespaces", []),
        "note": "Workload must authenticate with its Kubernetes service account JWT; no bearer token is issued by this helper.",
    }


def command_request(
    args: argparse.Namespace, runner: BaoRunner, grant: dict[str, Any]
) -> int:
    ttl = args.ttl or grant["ttl"]["default"]
    authz = authorize_request(
        args=args, grant=grant, ttl=ttl, purpose=args.purpose, delivery=args.delivery
    )
    if args.delivery == "kubernetes-auth":
        payload = kubernetes_auth_payload(grant, ttl, args.purpose, authz)
        payload["status"] = "dry-run" if args.dry_run else payload["status"]
        emit_json(payload)
        record_state_hub(
            args,
            state_hub_metadata(
                args=args,
                grant=grant,
                status=payload["status"],
                purpose=args.purpose,
                ttl=ttl,
                delivery=args.delivery,
                authz=authz,
            ),
        )
        return 0

    if args.dry_run:
        emit_json(
            {
                "status": "dry-run",
                "command": "request",
                "grant_id": grant["id"],
                "purpose": args.purpose,
                "ttl": ttl,
                "delivery_mode": args.delivery,
                "authorization_mode": authz.mode,
                "decision_id": authz.decision_id,
            }
        )
        if args.delivery == "response-wrap":
            print(
                "DRY-RUN: bao "
                + shlex.join(token_create_args(grant, ttl, args.wrap_ttl))
            )
        else:
            print("DRY-RUN: bao " + shlex.join(token_create_args(grant, ttl)))
        record_state_hub(
            args,
            state_hub_metadata(
                args=args,
                grant=grant,
                status="dry-run",
                purpose=args.purpose,
                ttl=ttl,
                delivery=args.delivery,
                authz=authz,
            ),
        )
        return 0

    record_state_hub(
        args,
        state_hub_metadata(
            args=args,
            grant=grant,
            status="requested",
            purpose=args.purpose,
            ttl=ttl,
            delivery=args.delivery,
            authz=authz,
        ),
    )
    if args.delivery == "response-wrap":
        result = runner.run(token_create_args(grant, ttl, args.wrap_ttl), quiet=True)
        wrap_info = parse_wrap_create(result.stdout)
        payload = {
            "grant_id": grant["id"],
            "purpose": args.purpose,
            "ttl": ttl,
            "delivery_mode": "response-wrap",
            "wrapping_token": wrap_info.get("token"),
            "wrapping_accessor": wrap_info.get("accessor"),
            "wrapped_accessor": wrap_info.get("wrapped_accessor"),
            "wrap_ttl": wrap_info.get("ttl"),
            "authorization_mode": authz.mode,
            "decision_id": authz.decision_id,
            "status": "wrapped",
        }
        emit_json(payload)
        record_state_hub(
            args,
            state_hub_metadata(
                args=args,
                grant=grant,
                status="wrapped",
                purpose=args.purpose,
                ttl=ttl,
                delivery=args.delivery,
                authz=authz,
                wrapping_accessor=payload.get("wrapping_accessor"),
                wrapped_accessor=payload.get("wrapped_accessor"),
            ),
        )
        return 0

    result = runner.run(token_create_args(grant, ttl), quiet=True)
    token, accessor = parse_token_create(result.stdout)
    payload = write_local_lease(
        lease_dir=args.lease_dir,
        grant=grant,
        purpose=args.purpose,
        ttl=ttl,
        token=token,
        accessor=accessor,
        authz=authz,
    )
    emit_json(payload)
    record_state_hub(
        args,
        state_hub_metadata(
            args=args,
            grant=grant,
            status="issued",
            purpose=args.purpose,
            ttl=ttl,
            delivery=args.delivery,
            authz=authz,
            lease_handle=accessor,
        ),
    )
    return 0


def command_exec(
    args: argparse.Namespace, runner: BaoRunner, grant: dict[str, Any]
) -> int:
    ttl = args.ttl or grant["ttl"]["default"]
    authz = authorize_request(
        args=args, grant=grant, ttl=ttl, purpose=args.purpose, delivery="exec-env"
    )
    extra_env, program = split_env_prefix(args.command)
    credential_type = grant["credential_type"]
    child_env = [*extra_env.keys()]
    if credential_type == "openbao-token":
        child_env.append("VAULT_TOKEN")
    else:
        child_env.extend((grant.get("database") or {}).get("env", {}).keys())
    if args.dry_run:
        emit_json(
            {
                "status": "dry-run",
                "command": "exec",
                "grant_id": grant["id"],
                "purpose": args.purpose,
                "ttl": ttl,
                "delivery_mode": "exec-env",
                "authorization_mode": authz.mode,
                "decision_id": authz.decision_id,
                "child_env": sorted(child_env),
                "child_command": program,
            }
        )
        print("DRY-RUN: bao " + shlex.join(token_create_args(grant, ttl)))
        record_state_hub(
            args,
            state_hub_metadata(
                args=args,
                grant=grant,
                status="dry-run",
                purpose=args.purpose,
                ttl=ttl,
                delivery="exec-env",
                authz=authz,
            ),
        )
        return 0

    record_state_hub(
        args,
        state_hub_metadata(
            args=args,
            grant=grant,
            status="requested",
            purpose=args.purpose,
            ttl=ttl,
            delivery="exec-env",
            authz=authz,
        ),
    )
    result = runner.run(token_create_args(grant, ttl), quiet=True)
    token, accessor = parse_token_create(result.stdout)
    lease_handle = accessor
    secret_values = [token]
    injected: dict[str, str]
    if credential_type == "openbao-token":
        injected = {"VAULT_TOKEN": token}
    else:
        database = grant.get("database") or {}
        role = database.get("role")
        env_map = database.get("env") or {}
        credential = runner.run_with_token(
            token, ["read", "-format=json", f"database/creds/{role}"], quiet=True
        )
        lease_handle, username, password = parse_database_credential(credential.stdout)
        values = {
            "username": username,
            "password": password,
            "host": str(database.get("host", "")),
            "port": str(database.get("port", "5432")),
            "database": str(database.get("name", "")),
        }
        injected = {name: values[field] for name, field in env_map.items()}
        secret_values.extend((username, password))
    record_state_hub(
        args,
        state_hub_metadata(
            args=args,
            grant=grant,
            status="issued",
            purpose=args.purpose,
            ttl=ttl,
            delivery="exec-env",
            authz=authz,
            lease_handle=lease_handle,
        ),
    )
    env = os.environ.copy()
    env.update(extra_env)
    env.update(injected)
    exit_code = 1
    try:
        child = subprocess.run(
            program, env=env, capture_output=True, text=True, check=False
        )
        exit_code = child.returncode
        if child.stdout:
            print(redact(child.stdout, secret_values), end="")
        if child.stderr:
            print(redact(child.stderr, secret_values), file=sys.stderr, end="")
        return child.returncode
    finally:
        if credential_type == "openbao-database-credential":
            runner.run(["lease", "revoke", lease_handle], quiet=True)
        runner.run(["write", "auth/token/revoke-accessor", f"accessor={accessor}"], quiet=True)
        record_state_hub(
            args,
            state_hub_metadata(
                args=args,
                grant=grant,
                status="revoked",
                purpose=args.purpose,
                ttl=ttl,
                delivery="exec-env",
                authz=authz,
                lease_handle=lease_handle,
                exit_code=exit_code,
            ),
        )


def command_status(args: argparse.Namespace, runner: BaoRunner) -> int:
    local = find_local_metadata(args.lease_dir, args.lease_handle)
    if args.dry_run:
        print(
            "DRY-RUN: bao write -format=json auth/token/lookup-accessor accessor=<lease_handle>"
        )
        emit_json(
            {
                "status": "dry-run",
                "lease_handle": args.lease_handle,
                "local_metadata_found": local is not None,
            }
        )
        return 0
    result = runner.run(
        [
            "write",
            "-format=json",
            "auth/token/lookup-accessor",
            f"accessor={args.lease_handle}",
        ],
        quiet=True,
    )
    payload = json.loads(result.stdout)
    if local:
        _, metadata = local
        payload.setdefault("local", metadata | {"token_file": "[LOCAL-SECRET-FILE]"})
    print(redact(json.dumps(payload, indent=2, sort_keys=True)))
    return 0


def command_revoke(args: argparse.Namespace, runner: BaoRunner) -> int:
    if args.dry_run:
        print("DRY-RUN: bao write auth/token/revoke-accessor accessor=<lease_handle>")
        emit_json({"status": "dry-run", "lease_handle": args.lease_handle})
        return 0
    runner.run(
        ["write", "auth/token/revoke-accessor", f"accessor={args.lease_handle}"],
        quiet=True,
    )
    removed = remove_local_lease_files(args.lease_dir, args.lease_handle)
    emit_json(
        {
            "status": "revoked",
            "lease_handle": args.lease_handle,
            "removed_local_files": removed,
        }
    )
    return 0


def add_common_issue_args(parser: argparse.ArgumentParser) -> None:
    parser.add_argument("--grant", default="ops-warden/warden-sign")
    parser.add_argument("--purpose", required=True)
    parser.add_argument("--ttl")
    parser.add_argument("--dry-run", action="store_true")


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Request, use, inspect, and revoke bounded OpenBao credential leases."
    )
    parser.add_argument("--catalog", default=str(DEFAULT_CATALOG))
    parser.add_argument(
        "--namespace", default=os.environ.get("OPENBAO_NAMESPACE", "openbao")
    )
    parser.add_argument(
        "--release", default=os.environ.get("OPENBAO_RELEASE", "openbao")
    )
    parser.add_argument("--kubectl", default=os.environ.get("KUBECTL", "kubectl"))
    parser.add_argument(
        "--issuer-token-file", default=os.environ.get("OPENBAO_TOKEN_FILE")
    )
    parser.add_argument(
        "--use-token-helper",
        action="store_true",
        help="Use the OpenBao CLI token helper inside the pod",
    )
    parser.add_argument(
        "--local-bao", action="store_true",
        help="Run the workstation bao CLI instead of execing into the server pod",
    )
    parser.add_argument("--lease-dir", type=Path, default=DEFAULT_LEASE_DIR)
    parser.add_argument(
        "--actor", default=os.environ.get("CREDENTIAL_ACTOR", DEFAULT_ACTOR)
    )
    parser.add_argument(
        "--actor-type",
        default=os.environ.get("CREDENTIAL_ACTOR_TYPE", DEFAULT_ACTOR_TYPE),
    )
    parser.add_argument(
        "--subject", default=os.environ.get("CREDENTIAL_SUBJECT", DEFAULT_SUBJECT)
    )
    parser.add_argument(
        "--decision-id", default=os.environ.get("CREDENTIAL_DECISION_ID")
    )
    parser.add_argument("--flex-auth-url", default=os.environ.get("FLEX_AUTH_URL"))
    parser.add_argument(
        "--flex-auth-path",
        default=os.environ.get("FLEX_AUTH_PATH", "/v1/check"),
    )
    parser.add_argument("--require-flex-auth", action="store_true")
    parser.add_argument(
        "--state-hub-url",
        default=os.environ.get("STATE_HUB_URL", "http://127.0.0.1:8000"),
    )
    parser.add_argument(
        "--state-hub-workplan-id", default=os.environ.get("STATE_HUB_WORKPLAN_ID")
    )
    parser.add_argument(
        "--state-hub-workstream-id", default=os.environ.get("STATE_HUB_WORKSTREAM_ID")
    )
    parser.add_argument(
        "--record-state-hub",
        action="store_true",
        help="Record non-secret request lifecycle metadata to State Hub",
    )
    parser.add_argument(
        "--http-timeout",
        type=float,
        default=float(os.environ.get("CREDENTIAL_HTTP_TIMEOUT", "10")),
    )

    subparsers = parser.add_subparsers(dest="command_name", required=True)

    request = subparsers.add_parser(
        "request",
        help="Issue a wrapped token, local-token-file lease, or Kubernetes-auth delegation",
    )
    add_common_issue_args(request)
    request.add_argument(
        "--delivery",
        choices=["local-token-file", "response-wrap", "kubernetes-auth"],
        default="local-token-file",
    )
    request.add_argument("--wrap-ttl", default="5m")

    exec_parser = subparsers.add_parser(
        "exec", help="Run a child process with VAULT_TOKEN injected"
    )
    add_common_issue_args(exec_parser)
    exec_parser.add_argument("command", nargs=argparse.REMAINDER)

    status = subparsers.add_parser(
        "status", help="Look up a lease by non-secret accessor handle"
    )
    status.add_argument("--dry-run", action="store_true")
    status.add_argument("lease_handle")

    revoke = subparsers.add_parser(
        "revoke", help="Revoke a lease by non-secret accessor handle"
    )
    revoke.add_argument("--dry-run", action="store_true")
    revoke.add_argument("lease_handle")

    return parser


def main() -> int:
    guard_verbose_env()
    parser = build_parser()
    args = parser.parse_args()
    args.lease_dir = resolve_repo_path(args.lease_dir)
    catalog = load_catalog(resolve_repo_path(Path(args.catalog)))
    if not args.state_hub_workplan_id:
        args.state_hub_workplan_id = catalog.get("state_hub_workplan_id")
    if not args.state_hub_workstream_id:
        args.state_hub_workstream_id = catalog.get("state_hub_workstream_id")
    if not args.state_hub_workplan_id:
        args.state_hub_workplan_id = args.state_hub_workstream_id
    grant = None
    if args.command_name in {"request", "exec"}:
        grant = get_grant(catalog, args.grant)
    issuer_token = read_issuer_token(
        args.issuer_token_file, args.dry_run, args.use_token_helper
    )
    runner = BaoRunner(
        kubectl=args.kubectl,
        namespace=args.namespace,
        release=args.release,
        dry_run=args.dry_run,
        use_token_helper=args.use_token_helper,
        issuer_token=issuer_token,
        local_bao=args.local_bao,
    )

    if args.command_name == "request":
        assert grant is not None
        return command_request(args, runner, grant)
    if args.command_name == "exec":
        assert grant is not None
        return command_exec(args, runner, grant)
    if args.command_name == "status":
        return command_status(args, runner)
    if args.command_name == "revoke":
        return command_revoke(args, runner)
    parser.error(f"unknown command: {args.command_name}")
    return 2


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