feat: add auth-capability lanes and pilot closeout

Add the warden-sign auth-capability lane, AppRole handoff, verification guards, docs, and tests.

Point the whynot-design pilot at the canonical decision and add the real publish closeout preflight/runbook.
This commit is contained in:
tegwick 2026-06-29 16:58:16 +02:00
parent a621fbaffd
commit 6382139890
27 changed files with 1455 additions and 107 deletions

View file

@ -35,12 +35,16 @@ def apply_plan(client: OpenBaoClient, entry: CatalogEntry, plan: Plan, ttl: str
applied: list[str] = []
skipped: list[str] = []
# 1. KV mount.
if client.kv_mount_exists(entry.mount):
skipped.append(f"kv-mount {entry.mount} (already present)")
# 1. KV mount. Auth-capability lanes grant operational access on an existing
# mount (for example ssh/sign/<role>) and never create or store KV values.
if entry.stores_kv_value():
if client.kv_mount_exists(entry.mount):
skipped.append(f"kv-mount {entry.mount} (already present)")
else:
client.ensure_kv_mount(entry.mount)
applied.append(f"kv-mount {entry.mount}")
else:
client.ensure_kv_mount(entry.mount)
applied.append(f"kv-mount {entry.mount}")
skipped.append(f"kv-mount {entry.mount} (not applicable for {entry.kind})")
# 2. Consumer ACL policy (write only if changed).
current = client.read_policy(plan.policy_name)
@ -52,7 +56,18 @@ def apply_plan(client: OpenBaoClient, entry: CatalogEntry, plan: Plan, ttl: str
# 3. Consumer approle bound to that policy.
client.ensure_approle_enabled()
client.write_approle(plan.role_name, [plan.policy_name], ttl=ttl)
if entry.kind == "auth-capability":
client.write_approle(
plan.role_name,
[plan.policy_name],
ttl=entry.token_ttl,
max_ttl=entry.token_max_ttl,
secret_id_ttl=entry.secret_id_ttl,
secret_id_num_uses=entry.secret_id_num_uses,
token_num_uses=entry.token_num_uses,
)
else:
client.write_approle(plan.role_name, [plan.policy_name], ttl=ttl)
applied.append(f"approle {plan.role_name} -> [{plan.policy_name}]")
return ApplyResult(applied=applied, skipped=skipped)

View file

@ -16,18 +16,26 @@ import yaml
from secrets_engine.errors import CatalogError
from secrets_engine.redact import looks_secret
VALID_KINDS = ("kv", "auth-capability")
VALID_STAGES = ("build", "test", "prod")
VALID_DELIVERY_MODES = ("exec-env", "exec-file", "npm-config", "wrapped", "read-check")
VALID_DELIVERY_MODES = (
"exec-env",
"exec-file",
"npm-config",
"wrapped",
"read-check",
"approle-login",
)
VALID_APPROVAL_MODELS = ("decision", "ccr", "dual-control", "bootstrap-only")
REQUIRED_FIELDS = (
"id",
"kind",
"org",
"repo",
"stage",
"mount",
"path",
"fields",
"consumers",
"delivery_modes",
"approval",
@ -41,6 +49,7 @@ REQUIRED_FIELDS = (
@dataclass(frozen=True)
class CatalogEntry:
id: str
kind: str
# Gitea coordinates, kept explicit to avoid the overloaded word "project".
# org = the Gitea organisation (e.g. "coulomb")
# repo = the Gitea repository / product (e.g. "whynot-design")
@ -58,6 +67,7 @@ class CatalogEntry:
deactivation: dict[str, Any]
audit: dict[str, Any]
delivery_config: dict[str, Any] = field(default_factory=dict)
auth_capability: dict[str, Any] = field(default_factory=dict)
description: str = ""
raw: dict[str, Any] = field(default_factory=dict)
@ -83,22 +93,75 @@ class CatalogEntry:
@property
def policy_name(self) -> str:
if self.kind == "auth-capability":
return self.auth_capability.get("policy_name") or self.id
return f"se-{self.stage}-{self.id}"
@property
def role_name(self) -> str:
if self.kind == "auth-capability":
return self.auth_capability.get("role_name") or self.id
return f"se-{self.stage}-{self.id}"
@property
def auth_allowed_paths(self) -> dict[str, list[str]]:
"""Allowed OpenBao paths/capabilities for an auth-capability lane."""
paths: dict[str, list[str]] = {}
for item in self.auth_capability.get("allowed_paths", []):
paths[item["path"]] = list(item["capabilities"])
return paths
@property
def auth_denied_probe_paths(self) -> list[str]:
"""Non-secret denial probes proving the lane did not get broader."""
return list(self.auth_capability.get("denied_probe_paths", []))
@property
def token_ttl(self) -> str:
return str(self.auth_capability.get("token_ttl", self.rotation.get("ttl", "30m")))
@property
def token_max_ttl(self) -> str:
return str(self.auth_capability.get("token_max_ttl", self.token_ttl))
@property
def secret_id_ttl(self) -> str:
return str(self.auth_capability.get("secret_id_ttl", self.token_ttl))
@property
def secret_id_num_uses(self) -> int:
return int(self.auth_capability.get("secret_id_num_uses", 0))
@property
def token_num_uses(self) -> int:
return int(self.auth_capability.get("token_num_uses", 0))
def approval_required(self) -> bool:
return self.approval.get("model") != "bootstrap-only"
def stores_kv_value(self) -> bool:
return self.kind == "kv"
def validate_entry(data: dict[str, Any], *, source: str = "<memory>") -> CatalogEntry:
"""Validate a raw mapping and return a CatalogEntry, or raise CatalogError."""
if not isinstance(data, dict):
raise CatalogError(f"{source}: catalog entry must be a mapping")
data = dict(data)
data.setdefault("kind", "kv")
if data["kind"] not in VALID_KINDS:
raise CatalogError(
f"{source}: kind '{data['kind']}' invalid; must be one of {VALID_KINDS}"
)
if data["kind"] == "auth-capability":
data.setdefault("fields", [])
else:
data.setdefault("auth_capability", {})
missing = [k for k in REQUIRED_FIELDS if k not in data or data[k] in (None, "", [], {})]
if data["kind"] == "kv" and not data.get("fields"):
missing.append("fields")
if missing:
raise CatalogError(f"{source}: missing required fields: {', '.join(missing)}")
@ -120,6 +183,8 @@ def validate_entry(data: dict[str, Any], *, source: str = "<memory>") -> Catalog
fields = data["fields"]
if not isinstance(fields, list) or not all(isinstance(f, str) for f in fields):
raise CatalogError(f"{source}: fields must be a list of strings")
if data["kind"] == "auth-capability" and fields:
raise CatalogError(f"{source}: auth-capability lanes must not declare KV fields")
consumers = data["consumers"]
if not isinstance(consumers, list) or not consumers:
@ -153,6 +218,35 @@ def validate_entry(data: dict[str, Any], *, source: str = "<memory>") -> Catalog
f"{source}: delivery_config.npm.registry must be an http(s) URL"
)
if data["kind"] == "auth-capability":
cfg = data.get("auth_capability")
if not isinstance(cfg, dict) or not cfg.get("allowed_paths"):
raise CatalogError(
f"{source}: auth-capability lanes require auth_capability.allowed_paths"
)
if "approle-login" not in modes:
raise CatalogError(
f"{source}: auth-capability lanes require approle-login delivery mode"
)
allowed = cfg["allowed_paths"]
if not isinstance(allowed, list) or not allowed:
raise CatalogError(f"{source}: auth_capability.allowed_paths must be a list")
for item in allowed:
if not isinstance(item, dict) or "path" not in item or "capabilities" not in item:
raise CatalogError(
f"{source}: each allowed path needs path and capabilities"
)
caps = item["capabilities"]
if not isinstance(caps, list) or not all(isinstance(c, str) for c in caps):
raise CatalogError(
f"{source}: auth_capability.allowed_paths capabilities must be strings"
)
denied = cfg.get("denied_probe_paths", [])
if not isinstance(denied, list) or not all(isinstance(p, str) for p in denied):
raise CatalogError(
f"{source}: auth_capability.denied_probe_paths must be a list of strings"
)
# A path must never leak a value through a field name suggesting inline secrets.
if any(looks_secret(k) and data.get(k) for k in ("value", "secret", "token", "password")):
raise CatalogError(f"{source}: catalog entries must not contain secret values")

View file

@ -7,7 +7,8 @@ Command surface (FR7):
plan <decision-or-ref> --stage <stage>
apply <decision-or-ref> --stage <stage> [--dry-run] [--bootstrap-token-file F]
provision <catalog-id> --stage <stage> (--from-file F | --generate) --field NAME
verify <catalog-id> [--positive] [--negative] --field NAME
verify <catalog-id> [--positive] [--negative] [--field NAME]
handoff <catalog-id> --stage <stage> --role-id-file F --secret-id-file F
exec --catalog <catalog-id> [--field NAME] [--mode auto|npm-config|exec-env] -- CMD...
route <catalog-id> [--json]
revoke <catalog-id>
@ -26,7 +27,7 @@ from secrets_engine.apply import apply_plan
from secrets_engine.catalog import get_entry, load_catalog
from secrets_engine.config import Config, repo_root
from secrets_engine.decisions import require_approved, resolve_decision
from secrets_engine.errors import SecretsEngineError
from secrets_engine.errors import DecisionError, SecretsEngineError
from secrets_engine.evidence import EvidenceWriter
from secrets_engine.openbao import OpenBaoClient
from secrets_engine.plan import build_plan
@ -71,16 +72,24 @@ def cmd_catalog_list(cfg: Config, args) -> int:
print("(no catalog entries)")
return 0
for e in entries.values():
print(f"{e.id:32s} stage={e.stage:5s} owner={e.owner:18s} {e.mount}/{e.path}")
print(
f"{e.id:32s} kind={e.kind:15s} stage={e.stage:5s} "
f"owner={e.owner:18s} {e.mount}/{e.path}"
)
return 0
def cmd_catalog_show(cfg: Config, args) -> int:
e = get_entry(cfg.catalog_dir, args.catalog_id)
print(f"id: {e.id}")
print(f"kind: {e.kind}")
print(f"owner: {e.owner}")
print(f"stage: {e.stage}")
print(f"openbao: {e.mount}/{e.path} fields={e.fields}")
if e.kind == "auth-capability":
print(f"openbao: mount={e.mount} allowed={sorted(e.auth_allowed_paths)}")
print(f"approle: {e.role_name} policy={e.policy_name}")
else:
print(f"openbao: {e.mount}/{e.path} fields={e.fields}")
print(f"consumers: {[c['name'] for c in e.consumers]}")
print(f"delivery: {e.delivery_modes}")
print(f"approval: {e.approval.get('model')} ref={e.approval.get('decision_ref','')}")
@ -111,11 +120,13 @@ def cmd_plan(cfg: Config, args) -> int:
entry = _resolve_lane_and_decision(cfg, args.ref, args.stage)
decision = None
if entry.approval_required():
decision = resolve_decision(
hub_url=cfg.hub_url, repo_root=repo_root(),
decision_ref=entry.approval.get("decision_ref", args.ref),
)
require_approved(entry, decision)
try:
decision = resolve_decision(
hub_url=cfg.hub_url, repo_root=repo_root(),
decision_ref=entry.approval.get("decision_ref", args.ref),
)
except DecisionError:
decision = None
plan = build_plan(entry, args.stage, decision_id=decision.id if decision else "")
print(plan.render())
_writer(cfg).record(
@ -129,11 +140,16 @@ def cmd_apply(cfg: Config, args) -> int:
entry = _resolve_lane_and_decision(cfg, args.ref, args.stage)
decision = None
if entry.approval_required():
decision = resolve_decision(
hub_url=cfg.hub_url, repo_root=repo_root(),
decision_ref=entry.approval.get("decision_ref", args.ref),
)
require_approved(entry, decision)
try:
decision = resolve_decision(
hub_url=cfg.hub_url, repo_root=repo_root(),
decision_ref=entry.approval.get("decision_ref", args.ref),
)
except DecisionError:
if not args.dry_run:
raise
if not args.dry_run:
require_approved(entry, decision)
plan = build_plan(entry, args.stage, decision_id=decision.id if decision else "")
w = _writer(cfg)
if args.dry_run:
@ -174,6 +190,9 @@ def cmd_verify(cfg: Config, args) -> int:
entry = get_entry(cfg.catalog_dir, args.catalog_id)
client = OpenBaoClient.resolve(cfg.bao_addr, bootstrap_token_file=args.bootstrap_token_file)
field = args.field or (entry.fields[0] if entry.fields else "")
if entry.stores_kv_value() and not field:
from secrets_engine.errors import VerificationError
raise VerificationError(f"lane '{entry.id}' has no field to verify")
positive = args.positive or not args.negative
negative = args.negative or not args.positive
results = run_verification(client, entry, field, positive=positive, negative=negative)
@ -187,6 +206,53 @@ def cmd_verify(cfg: Config, args) -> int:
return rc
def cmd_handoff(cfg: Config, args) -> int:
from secrets_engine.errors import ProvisioningError
from secrets_engine.handoff import write_approle_handoff
entry = get_entry(cfg.catalog_dir, args.catalog_id)
if args.stage != entry.stage:
raise ProvisioningError(f"lane '{entry.id}' is stage '{entry.stage}', not '{args.stage}'")
if entry.kind != "auth-capability":
raise ProvisioningError(f"lane '{entry.id}' is {entry.kind}; handoff needs auth-capability")
decision = None
if entry.approval_required():
decision = resolve_decision(
hub_url=cfg.hub_url, repo_root=repo_root(),
decision_ref=entry.approval.get("decision_ref", entry.id),
)
require_approved(entry, decision)
client = OpenBaoClient.resolve(cfg.bao_addr, bootstrap_token_file=args.bootstrap_token_file)
result = write_approle_handoff(
client,
entry,
role_id_file=Path(args.role_id_file),
secret_id_file=Path(args.secret_id_file),
)
print(f"wrote AppRole handoff material for lane '{entry.id}' — secret_id not displayed")
print(f" role: {result.role_name}")
print(f" role_id_file: {result.role_id_file}")
print(f" secret_id_file: {result.secret_id_file}")
print(f" token_ttl: {result.token_ttl}")
print(f" secret_id_ttl: {result.secret_id_ttl}")
_writer(cfg).record(
"handoff",
result="secret-id-written",
catalog_id=entry.id,
stage=entry.stage,
decision_id=decision.id if decision else "",
detail={
"role": result.role_name,
"role_id_file": result.role_id_file,
"secret_id_file": result.secret_id_file,
"token_ttl": result.token_ttl,
"secret_id_ttl": result.secret_id_ttl,
"secret_id_num_uses": result.secret_id_num_uses,
},
)
return 0
def cmd_exec(cfg: Config, args) -> int:
from secrets_engine.exec_delivery import exec_with_secret
entry = get_entry(cfg.catalog_dir, args.catalog)
@ -240,9 +306,13 @@ def cmd_route(cfg: Config, args) -> int:
import json
print(json.dumps(result.to_json(), indent=2))
else:
print(f"lane: {result.catalog_id} (owner={result.owner}, stage={result.stage})")
print(
f"lane: {result.catalog_id} "
f"(kind={result.kind}, owner={result.owner}, stage={result.stage})"
)
print(f"decision: {result.decision_status} ref={result.decision_ref}")
print(f"applied: {result.metadata_applied} value_present: {result.value_present}")
material_label = "handoff_ready" if result.kind == "auth-capability" else "value_present"
print(f"applied: {result.metadata_applied} {material_label}: {result.value_present}")
print(f"ready: {result.ready}")
if result.missing:
print(f"missing: {result.missing}")
@ -253,6 +323,21 @@ def cmd_route(cfg: Config, args) -> int:
def cmd_revoke(cfg: Config, args) -> int:
entry = get_entry(cfg.catalog_dir, args.catalog_id)
client = OpenBaoClient.resolve(cfg.bao_addr, bootstrap_token_file=args.bootstrap_token_file)
if entry.kind == "auth-capability":
if args.dry_run:
print(
f"(dry-run) would delete approle {entry.role_name} "
f"and policy {entry.policy_name}"
)
return 0
client.delete_approle(entry.role_name)
client.delete_policy(entry.policy_name)
print(
f"revoked lane '{entry.id}': deleted approle {entry.role_name} "
f"and policy {entry.policy_name}"
)
_writer(cfg).record("revoke", result="auth-capability-deactivated", catalog_id=entry.id, stage=entry.stage)
return 0
if args.dry_run:
print(f"(dry-run) would delete KV metadata {entry.mount}/{entry.path} "
f"and approle {entry.role_name}")
@ -318,6 +403,14 @@ def build_parser() -> argparse.ArgumentParser:
add_token_arg(ve)
ve.set_defaults(func=cmd_verify)
ha = sub.add_parser("handoff", help="write AppRole role_id/secret_id handoff files")
ha.add_argument("catalog_id")
ha.add_argument("--stage", required=True, choices=("build", "test", "prod"))
ha.add_argument("--role-id-file", required=True)
ha.add_argument("--secret-id-file", required=True)
add_token_arg(ha)
ha.set_defaults(func=cmd_handoff)
ex = sub.add_parser("exec", help="run a command with the secret injected for the child only")
ex.add_argument("--catalog", required=True)
ex.add_argument("--field", default=None)

View file

@ -0,0 +1,96 @@
"""Out-of-band AppRole handoff helpers for auth-capability lanes.
The secret_id is secret material. It is minted only after output paths have been
validated, written to a mode-0600 file outside any Git worktree, and never
printed or recorded in evidence.
"""
from __future__ import annotations
import os
from dataclasses import dataclass
from pathlib import Path
from secrets_engine.catalog import CatalogEntry
from secrets_engine.errors import ProvisioningError
from secrets_engine.openbao import OpenBaoClient
@dataclass(frozen=True)
class HandoffResult:
role_name: str
role_id_file: str
secret_id_file: str
token_ttl: str
secret_id_ttl: str
secret_id_num_uses: int
def _assert_outside_git_worktree(path: Path) -> Path:
resolved = path.expanduser().resolve()
for parent in (resolved.parent, *resolved.parent.parents):
if (parent / ".git").exists():
raise ProvisioningError(
f"handoff file {resolved} is inside a Git worktree ({parent}); "
"keep AppRole material outside repos"
)
return resolved
def _validate_output_path(path: Path) -> Path:
resolved = _assert_outside_git_worktree(path)
if resolved.exists() and resolved.stat().st_mode & 0o077:
raise ProvisioningError(
f"handoff file {resolved} is group/other-accessible "
f"(mode {oct(resolved.stat().st_mode & 0o777)}); must be 0600"
)
return resolved
def _write_mode_0600(path: Path, value: str) -> None:
path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
fd: int | None = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
try:
os.fchmod(fd, 0o600)
with os.fdopen(fd, "w", encoding="utf-8") as fh:
fd = None
fh.write(value)
fh.write("\n")
finally:
if fd is not None:
os.close(fd)
def write_approle_handoff(
client: OpenBaoClient,
entry: CatalogEntry,
*,
role_id_file: Path,
secret_id_file: Path,
) -> HandoffResult:
"""Mint and write AppRole handoff material without printing the secret_id."""
if entry.kind != "auth-capability":
raise ProvisioningError(
f"lane '{entry.id}' is {entry.kind}; handoff is only for auth-capability lanes"
)
role_path = _validate_output_path(role_id_file)
secret_path = _validate_output_path(secret_id_file)
if role_path == secret_path:
raise ProvisioningError("role_id_file and secret_id_file must be different")
role_id = client.read_approle_role_id(entry.role_name)
secret_id = client.create_approle_secret_id(entry.role_name)
try:
_write_mode_0600(role_path, role_id)
_write_mode_0600(secret_path, secret_id)
finally:
del secret_id
return HandoffResult(
role_name=entry.role_name,
role_id_file=str(role_path),
secret_id_file=str(secret_path),
token_ttl=entry.token_ttl,
secret_id_ttl=entry.secret_id_ttl,
secret_id_num_uses=entry.secret_id_num_uses,
)

View file

@ -143,28 +143,49 @@ class OpenBaoClient:
if enable.returncode != 0 and "already in use" not in enable.stderr:
raise BackendError(f"could not enable approle: {enable.stderr.strip()}")
def write_approle(self, role_name: str, policies: list[str], ttl: str = "30m") -> None:
self._run_ok(
[
"write",
f"auth/approle/role/{role_name}",
f"token_policies={','.join(policies)}",
f"token_ttl={ttl}",
f"token_max_ttl={ttl}",
"secret_id_num_uses=0",
"token_num_uses=0",
]
)
def write_approle(
self,
role_name: str,
policies: list[str],
ttl: str = "30m",
*,
max_ttl: str | None = None,
secret_id_ttl: str | None = None,
secret_id_num_uses: int = 0,
token_num_uses: int = 0,
) -> None:
args = [
"write",
f"auth/approle/role/{role_name}",
f"token_policies={','.join(policies)}",
f"token_ttl={ttl}",
f"token_max_ttl={max_ttl or ttl}",
f"secret_id_num_uses={secret_id_num_uses}",
f"token_num_uses={token_num_uses}",
]
if secret_id_ttl:
args.append(f"secret_id_ttl={secret_id_ttl}")
self._run_ok(args)
def approle_exists(self, role_name: str) -> bool:
proc = self._run(["read", f"auth/approle/role/{role_name}"])
return proc.returncode == 0
def read_approle_role_id(self, role_name: str) -> str:
return self._run_ok(
["read", "-field=role_id", f"auth/approle/role/{role_name}/role-id"]
).strip()
def create_approle_secret_id(self, role_name: str) -> str:
return self._run_ok(
["write", "-field=secret_id", "-f", f"auth/approle/role/{role_name}/secret-id"]
).strip()
def approle_login_token(self, role_name: str) -> str:
"""Login as the approle and return a scoped child token. Used only for
verification / exec delivery; never logged."""
role_id = self._run_ok(
["read", "-field=role_id", f"auth/approle/role/{role_name}/role-id"]
).strip()
secret_id = self._run_ok(
["write", "-field=secret_id", "-f", f"auth/approle/role/{role_name}/secret-id"]
).strip()
role_id = self.read_approle_role_id(role_name)
secret_id = self.create_approle_secret_id(role_name)
token = self._run_ok(
[
"write",
@ -176,6 +197,28 @@ class OpenBaoClient:
).strip()
return token
def token_capabilities(self, path: str, *, token: str) -> list[str]:
"""Return token capabilities for a path without returning any secret value."""
client = OpenBaoClient(addr=self.addr, token=token, bao_bin=self.bao_bin)
out = client._run_ok(["token", "capabilities", "-format=json", path])
try:
data = json.loads(out)
except json.JSONDecodeError:
return [line.strip() for line in out.splitlines() if line.strip()]
if isinstance(data, list):
return [str(item) for item in data]
if isinstance(data, dict):
caps = data.get("capabilities", [])
if isinstance(caps, list):
return [str(item) for item in caps]
return []
def delete_policy(self, name: str) -> None:
self._run_ok(["policy", "delete", name])
def delete_approle(self, role_name: str) -> None:
self._run_ok(["delete", f"auth/approle/role/{role_name}"])
# -- KV v2 -------------------------------------------------------------
def kv_mount_exists(self, mount: str) -> bool:

View file

@ -14,6 +14,7 @@ from secrets_engine.errors import PolicyGuardError
from secrets_engine.roles import (
StageRole,
assert_path_in_stage,
auth_capability_policy_for,
consumer_policy_for,
)
@ -63,22 +64,42 @@ def build_plan(entry: CatalogEntry, stage: str, *, decision_id: str = "") -> Pla
f"refusing to apply as '{stage}'"
)
StageRole.for_stage(stage) # validates stage name
assert_path_in_stage(entry) # path must be in-stage, no wildcards
policy_name, policy_hcl = consumer_policy_for(entry) # runs assert_policy_safe
if entry.kind == "auth-capability":
policy_name, policy_hcl = auth_capability_policy_for(entry)
actions = [
PlanAction(
"policy",
policy_name,
{"paths": ",".join(entry.auth_allowed_paths)},
),
PlanAction(
"approle",
entry.role_name,
{
"token_policies": policy_name,
"auth": "approle",
"token_ttl": entry.token_ttl,
"secret_id_num_uses": entry.secret_id_num_uses,
},
),
]
else:
assert_path_in_stage(entry) # path must be in-stage, no wildcards
policy_name, policy_hcl = consumer_policy_for(entry) # runs assert_policy_safe
actions = [
PlanAction("kv-mount", entry.mount, {"type": "kv-v2"}),
PlanAction(
"policy",
policy_name,
{"paths": f"{entry.mount}/data/{entry.path}"},
),
PlanAction(
"approle",
entry.role_name,
{"token_policies": policy_name, "auth": "approle"},
),
]
actions = [
PlanAction("kv-mount", entry.mount, {"type": "kv-v2"}),
PlanAction(
"policy",
policy_name,
{"paths": f"{entry.mount}/data/{entry.path}"},
),
PlanAction(
"approle",
entry.role_name,
{"token_policies": policy_name, "auth": "approle"},
),
]
return Plan(
catalog_id=entry.id,
stage=stage,

View file

@ -45,6 +45,10 @@ def provision_from_file(
client: OpenBaoClient, entry: CatalogEntry, field: str, file_path: Path
) -> str:
"""Import a value from a strict-permission file. Returns the field name only."""
if not entry.stores_kv_value():
raise ProvisioningError(
f"lane '{entry.id}' is {entry.kind}; it has no KV value to provision"
)
if field not in entry.fields:
raise ProvisioningError(
f"field '{field}' not declared in lane '{entry.id}' fields {entry.fields}"
@ -58,6 +62,10 @@ def provision_from_file(
def provision_generated(client: OpenBaoClient, entry: CatalogEntry, field: str) -> str:
"""Generate a random NON-PRODUCTION value for build/test lanes only."""
if not entry.stores_kv_value():
raise ProvisioningError(
f"lane '{entry.id}' is {entry.kind}; it has no KV value to provision"
)
if entry.stage == "prod":
raise ProvisioningError(
f"refusing to generate a value for prod lane '{entry.id}'; "

View file

@ -26,6 +26,10 @@ STAGE_PREFIX = {
# Capabilities a stage role's *own* policy may carry. Anything else is broad.
ALLOWED_CAPABILITIES = {"create", "read", "update", "delete", "list"}
# Auth-capability lanes grant an operational action, not KV value access. Keep
# this intentionally smaller than the stage-role capability set.
AUTH_CAPABILITY_ALLOWED_CAPABILITIES = {"update"}
# Substrings that, if they appear in a policy path, mean the plan is too broad.
FORBIDDEN_PATH_MARKERS = (
"sys/",
@ -110,6 +114,63 @@ def assert_policy_safe(policy_name: str, paths: dict[str, list[str]]) -> None:
)
def assert_auth_capability_safe(
policy_name: str,
mount: str,
paths: dict[str, list[str]],
denied_probe_paths: list[str] | None = None,
) -> None:
"""Reject auth-capability grants that are broader than exact operations."""
lowered = policy_name.lower()
for marker in FORBIDDEN_NAME_MARKERS:
if marker in lowered:
raise PolicyGuardError(
f"policy name '{policy_name}' resembles broad admin (marker '{marker}')"
)
if not paths:
raise PolicyGuardError(f"policy '{policy_name}': no auth-capability paths")
mount_prefix = f"{mount}/"
for path, caps in paths.items():
if "*" in path or "+" in path or path.strip() in ("*", "/", "+"):
raise PolicyGuardError(
f"policy '{policy_name}': wildcard auth path '{path}' not allowed"
)
for marker in FORBIDDEN_PATH_MARKERS:
if marker in path:
raise PolicyGuardError(
f"policy '{policy_name}': path '{path}' is out of bounds "
f"(marker '{marker}')"
)
if not path.startswith(mount_prefix):
raise PolicyGuardError(
f"policy '{policy_name}': auth path '{path}' is outside mount '{mount}'"
)
if mount == "ssh":
parts = path.split("/")
if len(parts) != 3 or parts[1] != "sign" or not parts[2]:
raise PolicyGuardError(
f"policy '{policy_name}': ssh auth-capability path '{path}' "
"must be exactly ssh/sign/<role>"
)
bad_caps = set(caps) - AUTH_CAPABILITY_ALLOWED_CAPABILITIES
if bad_caps:
raise PolicyGuardError(
f"policy '{policy_name}': capabilities {sorted(bad_caps)} not allowed "
f"for auth-capability (allowed {sorted(AUTH_CAPABILITY_ALLOWED_CAPABILITIES)})"
)
for path in denied_probe_paths or []:
if path in paths:
raise PolicyGuardError(
f"policy '{policy_name}': denied probe '{path}' is also allowed"
)
if "*" in path or "+" in path:
raise PolicyGuardError(
f"policy '{policy_name}': denied probe '{path}' must be exact"
)
def lane_policy_paths(entry: CatalogEntry) -> dict[str, list[str]]:
"""The minimal KV v2 paths + capabilities a consumer policy needs for a lane."""
data_path = f"{entry.mount}/data/{entry.path}"
@ -136,3 +197,43 @@ def consumer_policy_for(entry: CatalogEntry) -> tuple[str, str]:
paths = lane_policy_paths(entry)
name = entry.policy_name
return name, render_policy_hcl(name, paths)
def auth_capability_policy_paths(entry: CatalogEntry) -> dict[str, list[str]]:
"""The exact operational paths an auth-capability lane may exercise."""
paths = entry.auth_allowed_paths
assert_auth_capability_safe(
entry.policy_name, entry.mount, paths, entry.auth_denied_probe_paths
)
return paths
def render_auth_capability_policy_hcl(
policy_name: str,
mount: str,
paths: dict[str, list[str]],
denied_probe_paths: list[str] | None = None,
) -> str:
"""Render a narrow auth-capability ACL policy in HCL."""
denied_probe_paths = denied_probe_paths or []
assert_auth_capability_safe(policy_name, mount, paths, denied_probe_paths)
blocks = [
f'# Generated by secrets-engine for auth-capability policy "{policy_name}"',
"# Exact allowlist below. Every path outside it is denied by OpenBao default.",
]
if denied_probe_paths:
blocks.append("# Denial probes expected to lack update capability:")
blocks.extend(f"# - {path}" for path in denied_probe_paths)
for path, caps in paths.items():
cap_list = ", ".join(f'"{c}"' for c in caps)
blocks.append(f'path "{path}" {{\n capabilities = [{cap_list}]\n}}')
return "\n\n".join(blocks) + "\n"
def auth_capability_policy_for(entry: CatalogEntry) -> tuple[str, str]:
"""Return (policy_name, hcl) for a non-KV auth-capability lane."""
paths = auth_capability_policy_paths(entry)
name = entry.policy_name
return name, render_auth_capability_policy_hcl(
name, entry.mount, paths, entry.auth_denied_probe_paths
)

View file

@ -19,6 +19,7 @@ from secrets_engine.openbao import OpenBaoClient
@dataclass
class RouteResult:
catalog_id: str
kind: str
owner: str
stage: str
decision_status: str
@ -59,13 +60,20 @@ def route_lane(
metadata_applied = False
value_present = False
if client is not None and client.is_reachable():
metadata_applied = client.read_policy(entry.policy_name) is not None
# Presence check uses the engine's own token; reports boolean only.
field = entry.fields[0] if entry.fields else ""
if field:
value_present = client.kv_field_present(entry.mount, entry.path, field)
policy_applied = client.read_policy(entry.policy_name) is not None
role_applied = client.approle_exists(entry.role_name)
metadata_applied = policy_applied and role_applied
if entry.stores_kv_value():
# Presence check uses the engine's own token; reports boolean only.
field = entry.fields[0] if entry.fields else ""
if field:
value_present = client.kv_field_present(entry.mount, entry.path, field)
else:
# Auth-capability lanes have no stored value; a fresh secret_id is minted
# on demand through the handoff command once metadata exists.
value_present = metadata_applied
approved = decision is None or decision.is_approved()
approved = not entry.approval_required() or (decision is not None and decision.is_approved())
ready = approved and metadata_applied and value_present
if not approved:
@ -74,6 +82,12 @@ def route_lane(
elif not metadata_applied:
missing = "OpenBao policy/role apply"
next_command = f"secrets-engine apply {decision_ref or entry.id} --stage {entry.stage}"
elif entry.kind == "auth-capability":
missing = ""
next_command = (
f"secrets-engine handoff {entry.id} --stage {entry.stage} "
"--role-id-file <path> --secret-id-file <path>"
)
elif not value_present:
missing = "provisioned secret value"
next_command = (
@ -86,6 +100,7 @@ def route_lane(
return RouteResult(
catalog_id=entry.id,
kind=entry.kind,
owner=entry.owner,
stage=entry.stage,
decision_status=decision_status,

View file

@ -68,14 +68,94 @@ def verify_negative(client: OpenBaoClient, entry: CatalogEntry) -> VerifyResult:
)
def verify_auth_capability_positive(client: OpenBaoClient, entry: CatalogEntry) -> VerifyResult:
"""Approved AppRole token must carry update on every exact allowed path."""
try:
token = client.approle_login_token(entry.role_name)
except Exception as e:
return VerifyResult(
"positive",
False,
{"reason": f"could not obtain approle token: {e}", "role": entry.role_name},
)
missing: list[str] = []
for path in entry.auth_allowed_paths:
try:
caps = client.token_capabilities(path, token=token)
except Exception as e:
return VerifyResult(
"positive",
False,
{"reason": f"could not inspect capabilities: {e}", "path": path},
)
if "update" not in caps:
missing.append(path)
passed = not missing
return VerifyResult(
"positive",
passed,
{
"reason": "approle token can update every allowlisted path"
if passed
else "approle token lacks update on allowlisted paths",
"role": entry.role_name,
"allowed_paths": sorted(entry.auth_allowed_paths),
"missing_update": missing,
},
)
def verify_auth_capability_negative(client: OpenBaoClient, entry: CatalogEntry) -> VerifyResult:
"""Approved AppRole token must not gain update outside denial probes."""
try:
token = client.approle_login_token(entry.role_name)
except Exception as e:
return VerifyResult(
"negative",
False,
{"reason": f"could not obtain approle token: {e}", "role": entry.role_name},
)
leaks: list[str] = []
for path in entry.auth_denied_probe_paths:
try:
caps = client.token_capabilities(path, token=token)
except Exception as e:
return VerifyResult(
"negative",
False,
{"reason": f"could not inspect capabilities: {e}", "path": path},
)
if "update" in caps or "sudo" in caps or "root" in caps:
leaks.append(path)
passed = not leaks
return VerifyResult(
"negative",
passed,
{
"reason": "denial probes lack update capability"
if passed
else "approle token can update outside the allowlist",
"role": entry.role_name,
"denied_probe_paths": entry.auth_denied_probe_paths,
"leaks": leaks,
},
)
def run_verification(
client: OpenBaoClient, entry: CatalogEntry, field: str, *, positive: bool, negative: bool
) -> list[VerifyResult]:
results: list[VerifyResult] = []
if positive:
results.append(verify_positive(client, entry, field))
if negative:
results.append(verify_negative(client, entry))
if entry.kind == "auth-capability":
if positive:
results.append(verify_auth_capability_positive(client, entry))
if negative:
results.append(verify_auth_capability_negative(client, entry))
else:
if positive:
results.append(verify_positive(client, entry, field))
if negative:
results.append(verify_negative(client, entry))
if not results:
raise VerificationError("no verification check selected (use --positive/--negative)")
return results