#!/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 : {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=" ) 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=") 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())