c921aa0aed
Silent generate-in-custody: verifies the SecurityCase file (digest, type, materialized, confidential, client_id) per ops-warden 23985370, mints the 5m create-only grant token from the caller's session, proves absence, writes once with CAS 0, verifies by metadata only and revokes the token. Agents, root and callers without the issuer policy are refused. Secret length awaits key-cape. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 469992@bnt-lap001 Assistant-Session: e4b8e420-df0a-441d-83ae-7f19130594d9
1227 lines
41 KiB
Python
Executable file
1227 lines
41 KiB
Python
Executable file
#!/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()
|
|
if sys.argv[1:2] == ["generate"]:
|
|
# RPF-WP-0054: generate-in-custody runs on the caller's own session and
|
|
# must stay silent, so it bypasses the lease runner and issuer prompt.
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
from keycape_client_secret_generate import main as generate_main
|
|
return generate_main(sys.argv[2:])
|
|
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())
|