Add credential lane readiness proposals

This commit is contained in:
2026-06-27 23:30:29 +02:00
parent 815b124ab1
commit aee0dcefad
13 changed files with 425 additions and 25 deletions

View File

@@ -46,6 +46,15 @@ SECRET_MARKERS = [
"sk-",
]
DISALLOWED_POLICY_NAMES = {"root", "platform-admin"}
FRONTDOOR_READINESS = {
"template",
"pending-review",
"approved-pending-apply",
"applied-pending-verify",
"ready",
"disabled",
"compromised",
}
SAFE_ID_RE = re.compile(r"^[A-Z0-9][A-Z0-9_.-]*$")
TTL_RE = re.compile(r"^[1-9][0-9]*[smhd]$")
@@ -177,6 +186,19 @@ def validate_workload_kv_read(ccr: dict[str, Any], errors: list[str], warnings:
frontdoor = require_object(ccr.get("access_frontdoor"), "access_frontdoor", errors)
require_string(frontdoor.get("type"), "access_frontdoor.type", errors)
require_string(frontdoor.get("catalog_id"), "access_frontdoor.catalog_id", errors)
readiness = require_string(frontdoor.get("readiness"), "access_frontdoor.readiness", errors)
if readiness and readiness not in FRONTDOOR_READINESS:
errors.append(
f"access_frontdoor.readiness must be one of {sorted(FRONTDOOR_READINESS)}"
)
resolvable = frontdoor.get("resolvable")
if not isinstance(resolvable, bool):
errors.append("access_frontdoor.resolvable must be a boolean")
if resolvable is True and ccr.get("status") != "active":
errors.append("access_frontdoor.resolvable=true requires status active")
command = frontdoor.get("command")
if command is not None and not isinstance(command, str):
errors.append("access_frontdoor.command must be a string when present")
risk = require_object(ccr.get("risk"), "risk", errors)
require_string(risk.get("classification"), "risk.classification", errors)
@@ -258,8 +280,11 @@ def render_summary(ccr: dict[str, Any], warnings: list[str]) -> str:
f" confirmed: {auth.get('bound_claims_confirmed') is True}",
"Access front door:",
f" {frontdoor['type']} {frontdoor['catalog_id']}",
f"Risk: {risk['classification']}",
f" readiness: {frontdoor.get('readiness')} resolvable={frontdoor.get('resolvable') is True}",
]
if frontdoor.get("command"):
lines.append(f" command: {frontdoor['command']}")
lines.append(f"Risk: {risk['classification']}")
for note in risk.get("notes", []):
lines.append(f" - {note}")
lines.append("Checks:")
@@ -298,8 +323,19 @@ def generated_policy_hcl(ccr: dict[str, Any]) -> str:
def auth_payload(ccr: dict[str, Any]) -> dict[str, Any]:
auth = ccr["openbao"]["auth"]
if auth["method"] == "kubernetes":
claims = auth["bound_claims"]
return {
"bound_service_account_names": claims.get("service_account_names", []),
"bound_service_account_namespaces": claims.get(
"service_account_namespaces", []
),
"policies": ",".join(auth["policies"]),
"ttl": auth.get("ttl", "15m"),
}
payload: dict[str, Any] = {
"role_type": "oidc" if auth["method"] == "oidc" else "jwt",
"role_type": "oidc",
"user_claim": auth.get("user_claim", "sub"),
"policies": ",".join(auth["policies"]),
"ttl": auth.get("ttl", "15m"),
@@ -347,6 +383,91 @@ def validate_or_exit(path: Path) -> tuple[dict[str, Any], list[str]]:
return ccr, warnings
def apply_blockers(ccr: dict[str, Any]) -> list[str]:
blockers: list[str] = []
if ccr.get("status") not in APPLY_ALLOWED_STATUSES:
blockers.append(f"apply requires status approved, got {ccr.get('status')}")
if ccr["openbao"]["auth"].get("bound_claims_confirmed") is not True:
blockers.append("apply requires confirmed OpenBao auth binding")
return blockers
def frontdoor_blockers(ccr: dict[str, Any]) -> list[str]:
frontdoor = ccr["access_frontdoor"]
blockers: list[str] = []
if ccr.get("status") != "active":
blockers.append(f"front door requires CCR status active, got {ccr.get('status')}")
if frontdoor.get("readiness") != "ready":
blockers.append(
f"front door readiness must be ready, got {frontdoor.get('readiness')}"
)
if frontdoor.get("resolvable") is not True:
blockers.append("front door is marked resolvable=false")
return blockers
def status_payload(ccr: dict[str, Any], warnings: list[str]) -> dict[str, Any]:
apply_blocked_by = apply_blockers(ccr)
frontdoor_blocked_by = frontdoor_blockers(ccr)
frontdoor = ccr["access_frontdoor"]
openbao = ccr["openbao"]
auth = openbao["auth"]
return {
"id": ccr["id"],
"title": ccr["title"],
"status": ccr["status"],
"request_type": ccr["request_type"],
"apply_allowed": not apply_blocked_by,
"apply_blockers": apply_blocked_by,
"frontdoor_resolvable": not frontdoor_blocked_by,
"frontdoor_blockers": frontdoor_blocked_by,
"warnings": warnings,
"openbao": {
"mount": openbao["mount"],
"kv_path": openbao["kv_path"],
"fields": openbao["fields"],
"policy_name": openbao["policy_name"],
"auth_mount": auth["mount"],
"auth_method": auth["method"],
"auth_role": auth["role"],
"bound_claims_confirmed": auth.get("bound_claims_confirmed") is True,
},
"access_frontdoor": {
"type": frontdoor["type"],
"catalog_id": frontdoor["catalog_id"],
"readiness": frontdoor.get("readiness"),
"resolvable": frontdoor.get("resolvable") is True,
"command": frontdoor.get("command"),
},
}
def render_status(payload: dict[str, Any]) -> str:
lines = [
f"CCR: {payload['id']} ({payload['status']})",
f"Catalog: {payload['access_frontdoor']['catalog_id']}",
f"Readiness: {payload['access_frontdoor']['readiness']}",
f"Resolvable: {payload['frontdoor_resolvable']}",
f"Apply allowed: {payload['apply_allowed']}",
]
command = payload["access_frontdoor"].get("command")
if command:
lines.append(f"Command: {command}")
if payload["apply_blockers"]:
lines.append("Apply blockers:")
for blocker in payload["apply_blockers"]:
lines.append(f" - {blocker}")
if payload["frontdoor_blockers"]:
lines.append("Front-door blockers:")
for blocker in payload["frontdoor_blockers"]:
lines.append(f" - {blocker}")
if payload["warnings"]:
lines.append("Warnings:")
for warning in payload["warnings"]:
lines.append(f" - {warning}")
return "\n".join(lines)
def append_decision(path: Path, status: str, reviewer: str, comment: str) -> None:
ccr, _warnings = validate_or_exit(path)
review = ccr.setdefault("review", {})
@@ -399,6 +520,21 @@ def command_plan(args: argparse.Namespace) -> int:
return 0
def command_status(args: argparse.Namespace) -> int:
path = resolve_ccr(args.ref)
ccr, errors, warnings = validate_ccr(path)
if errors:
for error in errors:
print(f"[FAIL] {path.name}: {error}", file=sys.stderr)
return 1
payload = status_payload(ccr, warnings)
if args.json:
print(json.dumps(payload, indent=2, sort_keys=True))
else:
print(render_status(payload))
return 0
def command_apply_plan(args: argparse.Namespace) -> int:
path = resolve_ccr(args.ref)
ccr, _warnings = validate_or_exit(path)
@@ -436,6 +572,11 @@ def build_parser() -> argparse.ArgumentParser:
plan.add_argument("ref")
plan.set_defaults(func=command_plan)
status = sub.add_parser("status", help="Render machine-readable readiness status")
status.add_argument("ref")
status.add_argument("--json", action="store_true")
status.set_defaults(func=command_status)
apply_plan = sub.add_parser(
"apply-plan", help="Render an operator apply plan only for approved CCRs"
)