0b2e7ddb5f
Package the reviewed keycape-services JWT mount, secrets-engine-openbao role and secrets-engine-login-self policy as CCR-2026-0038 and teach the delegated applier a narrow service-jwt-login request type: fixed mount, generated login-self policy, allow-listed role/config keys, bounded TTL/uses, refusal on a foreign mount type and a value-free readback compare. The policy now denies the client-secret data/metadata paths explicitly (MASON-WP-0009 invariant ii). Proposed only; founder approval and delegated apply pending. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 327203@bnt-lap001 Assistant-Session: 55028ae3-4914-4bd1-8f5c-815f3527aef0
327 lines
16 KiB
Python
327 lines
16 KiB
Python
"""service-jwt-login CCRs: an isolated JWT auth mount, one role, one login-only policy.
|
|
|
|
RPF-WP-0035-T02. Loaded by scripts/credential-change.py for request_type
|
|
service-jwt-login. Deliberately narrow: the lane is an identity login, never a
|
|
lane applier, so the policy may hold only the three self endpoints plus explicit
|
|
denies on guarded paths (ops-mason MASON-WP-0009 invariant (ii)). Everything
|
|
written to OpenBao comes from reviewed repo files; nothing here reads, prints or
|
|
accepts a secret value.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import subprocess
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
REQUEST_TYPE = "service-jwt-login"
|
|
ALLOWED_AUTH_MOUNTS = {"keycape-services"}
|
|
SELF_GRANTS = (
|
|
("auth/token/lookup-self", "read"),
|
|
("sys/capabilities-self", "update"),
|
|
("auth/token/revoke-self", "update"),
|
|
)
|
|
POLICY_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9-]*-login-self$")
|
|
ROLE_RE = re.compile(r"^[a-z0-9][a-z0-9-]*$")
|
|
TTL_RE = re.compile(r"^([1-9][0-9]*)([sm])$")
|
|
MAX_TTL_SECONDS = 15 * 60
|
|
MAX_TOKEN_USES = 16
|
|
CONFIG_KEYS = {"jwks_url", "bound_issuer", "jwt_supported_algs"}
|
|
ROLE_KEYS = {
|
|
"role_type", "user_claim", "bound_audiences", "bound_subject",
|
|
"bound_claims_type", "bound_claims", "token_policies",
|
|
"token_no_default_policy", "token_type", "token_ttl", "token_max_ttl",
|
|
"token_explicit_max_ttl", "token_num_uses", "verbose_oidc_logging",
|
|
}
|
|
DURATION_KEYS = ("token_ttl", "token_max_ttl", "token_explicit_max_ttl")
|
|
DISALLOWED_POLICIES = {"root", "platform-admin", "default"}
|
|
|
|
|
|
def ttl_seconds(value: Any) -> int | None:
|
|
if isinstance(value, int) and not isinstance(value, bool):
|
|
return value
|
|
match = TTL_RE.match(str(value))
|
|
if not match:
|
|
return None
|
|
return int(match.group(1)) * (60 if match.group(2) == "m" else 1)
|
|
|
|
|
|
def generated_policy_hcl(ccr: dict[str, Any]) -> str:
|
|
openbao = ccr["openbao"]
|
|
lines = [f"# {ccr['id']} (RPF-WP-0035-T02): login-only service policy; not a lane applier."]
|
|
for path, capability in SELF_GRANTS:
|
|
lines += [f'path "{path}" {{', f' capabilities = ["{capability}"]', "}"]
|
|
for path in openbao.get("guarded_paths") or []:
|
|
lines += [f'path "{path}" {{', ' capabilities = ["deny"]', "}"]
|
|
return "\n".join(lines) + "\n"
|
|
|
|
|
|
def normalized(text: str) -> str:
|
|
return "\n".join(
|
|
line.rstrip() for line in text.strip().splitlines() if not line.lstrip().startswith("#")
|
|
)
|
|
|
|
|
|
def _load_json(repo_dir: Path, rel: str, field: str, errors: list[str]) -> dict[str, Any]:
|
|
path = (repo_dir / rel).resolve()
|
|
try:
|
|
path.relative_to((repo_dir / "openbao" / "auth").resolve())
|
|
except ValueError:
|
|
errors.append(f"{field} must stay under openbao/auth")
|
|
return {}
|
|
if not path.exists():
|
|
errors.append(f"{field} does not exist: {rel}")
|
|
return {}
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
if not isinstance(data, dict):
|
|
errors.append(f"{field} must be a JSON object")
|
|
return {}
|
|
return data
|
|
|
|
|
|
def role_payload(ccr: dict[str, Any], repo_dir: Path) -> dict[str, Any]:
|
|
return _load_json(repo_dir, ccr["openbao"]["auth"]["role_file"], "openbao.auth.role_file", [])
|
|
|
|
|
|
def config_payload(ccr: dict[str, Any], repo_dir: Path) -> dict[str, Any]:
|
|
return _load_json(repo_dir, ccr["openbao"]["auth"]["config_file"], "openbao.auth.config_file", [])
|
|
|
|
|
|
def validate(ccr: dict[str, Any], repo_dir: Path) -> list[str]:
|
|
"""Schema plus every boundary the delegated applier relies on."""
|
|
errors: list[str] = []
|
|
openbao = ccr.get("openbao")
|
|
if not isinstance(openbao, dict):
|
|
return ["openbao must be an object"]
|
|
auth = openbao.get("auth")
|
|
if not isinstance(auth, dict):
|
|
return ["openbao.auth must be an object"]
|
|
target = ccr.get("target") or {}
|
|
if target.get("environment") != "production":
|
|
errors.append("target.environment must be production")
|
|
review = ccr.get("review") or {}
|
|
if review.get("required") is not True or not review.get("required_approvers"):
|
|
errors.append("review.required must be true with required_approvers")
|
|
|
|
policy_name = str(openbao.get("policy_name") or "")
|
|
if not POLICY_NAME_RE.match(policy_name) or policy_name in DISALLOWED_POLICIES:
|
|
errors.append("openbao.policy_name must be a *-login-self policy")
|
|
policy_file = str(openbao.get("policy_file") or "")
|
|
if policy_file != f"openbao/policies/{policy_name}.hcl":
|
|
errors.append("openbao.policy_file must be openbao/policies/<policy_name>.hcl")
|
|
guarded = openbao.get("guarded_paths")
|
|
if not isinstance(guarded, list) or not guarded:
|
|
errors.append("openbao.guarded_paths must list the paths this identity must never reach")
|
|
guarded = []
|
|
for path in guarded:
|
|
if not isinstance(path, str) or any(f in path for f in ("*", "+", "..", "//")) or not path.strip():
|
|
errors.append(f"openbao.guarded_paths entry must be an exact path: {path!r}")
|
|
policy_path = repo_dir / policy_file
|
|
if policy_file and policy_path.exists():
|
|
if normalized(policy_path.read_text(encoding="utf-8")) != normalized(generated_policy_hcl(ccr)):
|
|
errors.append("openbao.policy_file does not match the generated login-self policy")
|
|
elif policy_file:
|
|
errors.append(f"openbao.policy_file does not exist: {policy_file}")
|
|
|
|
if auth.get("method") != "jwt":
|
|
errors.append("openbao.auth.method must be jwt")
|
|
mount = str(auth.get("mount") or "")
|
|
if mount not in ALLOWED_AUTH_MOUNTS:
|
|
errors.append(f"openbao.auth.mount must be one of {sorted(ALLOWED_AUTH_MOUNTS)}")
|
|
role_name = str(auth.get("role") or "")
|
|
if not ROLE_RE.match(role_name):
|
|
errors.append("openbao.auth.role contains unsafe characters")
|
|
if not isinstance(auth.get("bound_claims_confirmed"), bool):
|
|
errors.append("openbao.auth.bound_claims_confirmed must be a boolean")
|
|
|
|
config = _load_json(repo_dir, str(auth.get("config_file") or ""), "openbao.auth.config_file", errors)
|
|
if config:
|
|
if set(config) != CONFIG_KEYS:
|
|
errors.append(f"auth config keys must be exactly {sorted(CONFIG_KEYS)}")
|
|
issuer = str(config.get("bound_issuer") or "")
|
|
jwks = str(config.get("jwks_url") or "")
|
|
if not issuer.startswith("https://") or not jwks.startswith(issuer.rstrip("/") + "/"):
|
|
errors.append("auth config needs an https bound_issuer and a jwks_url under it")
|
|
if config.get("jwt_supported_algs") != ["RS256"]:
|
|
errors.append("auth config must pin jwt_supported_algs to [RS256]")
|
|
|
|
role = _load_json(repo_dir, str(auth.get("role_file") or ""), "openbao.auth.role_file", errors)
|
|
if role:
|
|
extra = set(role) - ROLE_KEYS
|
|
if extra:
|
|
errors.append(f"role has keys outside the reviewed set: {sorted(extra)}")
|
|
if role.get("role_type") != "jwt" or role.get("user_claim") != "sub":
|
|
errors.append("role must be role_type jwt with user_claim sub")
|
|
if role.get("token_policies") != [policy_name]:
|
|
errors.append("role token_policies must be exactly [openbao.policy_name]")
|
|
if role.get("token_no_default_policy") is not True:
|
|
errors.append("role must set token_no_default_policy true")
|
|
if role.get("token_type") != "service":
|
|
errors.append("role token_type must be service")
|
|
if role.get("verbose_oidc_logging") is not False:
|
|
errors.append("role must disable verbose_oidc_logging")
|
|
audiences = role.get("bound_audiences")
|
|
if not isinstance(audiences, list) or len(audiences) != 1 or audiences[0] != role_name:
|
|
errors.append("role bound_audiences must be exactly [<role name>] (KeyCape aud = clientId)")
|
|
if not str(role.get("bound_subject") or "").startswith("service:"):
|
|
errors.append("role bound_subject must be an exact service: subject")
|
|
claims = role.get("bound_claims")
|
|
if role.get("bound_claims_type") != "string" or not isinstance(claims, dict):
|
|
errors.append("role bound_claims must be string-typed")
|
|
claims = {}
|
|
for key in ("principal_type", "tenant", "roles", "scope"):
|
|
value = claims.get(key)
|
|
if not isinstance(value, str) or not value or "*" in value or "," in value:
|
|
errors.append(f"role bound_claims.{key} must be one exact string")
|
|
if claims.get("principal_type") != "service":
|
|
errors.append("role bound_claims.principal_type must be service")
|
|
seconds = [ttl_seconds(role.get(key)) for key in DURATION_KEYS]
|
|
if any(s is None or s <= 0 or s > MAX_TTL_SECONDS for s in seconds):
|
|
errors.append("role token TTLs must be set and at most 15m")
|
|
uses = role.get("token_num_uses")
|
|
if not isinstance(uses, int) or isinstance(uses, bool) or not 0 < uses <= MAX_TOKEN_USES:
|
|
errors.append(f"role token_num_uses must be 1..{MAX_TOKEN_USES}")
|
|
return errors
|
|
|
|
|
|
def readiness_blockers(ccr: dict[str, Any], repo_dir: Path, allowed_statuses: set[str]) -> list[str]:
|
|
blockers: list[str] = []
|
|
if ccr.get("status") not in allowed_statuses:
|
|
blockers.append(
|
|
"applier requires approved, applied, verified, or active CCR status, "
|
|
f"got {ccr.get('status')}"
|
|
)
|
|
if ccr["openbao"]["auth"].get("bound_claims_confirmed") is not True:
|
|
blockers.append("applier requires confirmed OpenBao auth binding")
|
|
blockers.extend(validate(ccr, repo_dir))
|
|
return blockers
|
|
|
|
|
|
def dry_run_payload(ccr: dict[str, Any], repo_dir: Path, warnings: list[str]) -> dict[str, Any]:
|
|
openbao = ccr["openbao"]
|
|
auth = openbao["auth"]
|
|
return {
|
|
"id": ccr["id"],
|
|
"title": ccr["title"],
|
|
"status": ccr["status"],
|
|
"environment": ccr["target"]["environment"],
|
|
"warnings": warnings,
|
|
"source_artifacts": {},
|
|
"mutations": [
|
|
{"kind": "policy_write", "openbao_path": f"sys/policies/acl/{openbao['policy_name']}",
|
|
"policy_name": openbao["policy_name"], "source": openbao["policy_file"],
|
|
"body": generated_policy_hcl(ccr).rstrip()},
|
|
{"kind": "auth_enable_if_absent", "openbao_path": f"sys/auth/{auth['mount']}",
|
|
"payload": {"type": "jwt", "refuse_if_existing_type_differs": True}},
|
|
{"kind": "auth_config_write", "openbao_path": f"auth/{auth['mount']}/config",
|
|
"payload": config_payload(ccr, repo_dir)},
|
|
{"kind": "auth_role_write", "openbao_path": f"auth/{auth['mount']}/role/{auth['role']}",
|
|
"payload": role_payload(ccr, repo_dir)},
|
|
{"kind": "readback_compare", "openbao_path": "policy, auth mount type, config, role",
|
|
"payload": {"prints": "booleans only"}},
|
|
],
|
|
"out_of_scope": [
|
|
"secret value reads or writes",
|
|
"token creation or login",
|
|
"changes to any other auth mount, role or policy",
|
|
"front-door activation before the login proof",
|
|
],
|
|
"required_evidence": [
|
|
"CCR id and approval reference",
|
|
"readback equality for policy, mount type, config and role",
|
|
"login proof from a real KeyCape service JWT (separate, value-free)",
|
|
],
|
|
}
|
|
|
|
|
|
def apply_details(ccr: dict[str, Any], actor: str) -> list[str]:
|
|
openbao = ccr["openbao"]
|
|
auth = openbao["auth"]
|
|
return [
|
|
f"Delegated metadata applier ran as {actor} using local bao CLI ambient authority.",
|
|
f"Policy metadata write: sys/policies/acl/{openbao['policy_name']}",
|
|
f"JWT auth mount ensured: sys/auth/{auth['mount']} (type jwt)",
|
|
f"Auth config write: auth/{auth['mount']}/config",
|
|
f"Auth role metadata write: auth/{auth['mount']}/role/{auth['role']}",
|
|
"Readback matched source for policy, mount type, config and role.",
|
|
"No secret values were read, written, printed, or accepted in argv.",
|
|
]
|
|
|
|
|
|
class BaoError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def _run(bao_bin: str, args: list[str], repo_dir: Path, quiet: bool, capture: bool = False) -> str:
|
|
command = [bao_bin, *args]
|
|
if capture or quiet:
|
|
result = subprocess.run(command, cwd=repo_dir, capture_output=True, text=True)
|
|
if result.returncode != 0:
|
|
raise BaoError(f"bao {' '.join(args[:2])} failed with exit code {result.returncode}")
|
|
return result.stdout
|
|
subprocess.run(command, cwd=repo_dir, check=True)
|
|
return ""
|
|
|
|
|
|
def _read_json(bao_bin: str, args: list[str], repo_dir: Path) -> dict[str, Any]:
|
|
# "auth list -format=json" / "read -format=json <path>": flags follow the verb.
|
|
flagged = [*args, "-format=json"] if args[:2] == ["auth", "list"] else [args[0], "-format=json", *args[1:]]
|
|
out = _run(bao_bin, flagged, repo_dir, quiet=True, capture=True)
|
|
try:
|
|
return json.loads(out)
|
|
except json.JSONDecodeError as exc:
|
|
raise BaoError(f"bao {args[0]} returned non-JSON output") from exc
|
|
|
|
|
|
def readback_mismatches(ccr: dict[str, Any], repo_dir: Path, bao_bin: str) -> list[str]:
|
|
openbao = ccr["openbao"]
|
|
auth = openbao["auth"]
|
|
mismatches: list[str] = []
|
|
policy = _run(bao_bin, ["policy", "read", openbao["policy_name"]], repo_dir, quiet=True, capture=True)
|
|
if normalized(policy) != normalized(generated_policy_hcl(ccr)):
|
|
mismatches.append("policy text differs from source")
|
|
mounts = _read_json(bao_bin, ["auth", "list"], repo_dir)
|
|
mounts = mounts.get("data", mounts)
|
|
if (mounts.get(f"{auth['mount']}/") or {}).get("type") != "jwt":
|
|
mismatches.append("auth mount is not type jwt")
|
|
config = _read_json(bao_bin, ["read", f"auth/{auth['mount']}/config"], repo_dir).get("data", {})
|
|
for key, want in config_payload(ccr, repo_dir).items():
|
|
if config.get(key) != want:
|
|
mismatches.append(f"config {key} differs")
|
|
role = _read_json(bao_bin, ["read", f"auth/{auth['mount']}/role/{auth['role']}"], repo_dir).get("data", {})
|
|
for key, want in role_payload(ccr, repo_dir).items():
|
|
got = role.get(key)
|
|
if key in DURATION_KEYS:
|
|
same = ttl_seconds(got) == ttl_seconds(want)
|
|
else:
|
|
same = got == want
|
|
if not same:
|
|
mismatches.append(f"role {key} differs")
|
|
return mismatches
|
|
|
|
|
|
def run_apply(ccr: dict[str, Any], repo_dir: Path, bao_bin: str, quiet: bool = False) -> None:
|
|
openbao = ccr["openbao"]
|
|
auth = openbao["auth"]
|
|
mount = auth["mount"]
|
|
mounts = _read_json(bao_bin, ["auth", "list"], repo_dir)
|
|
mounts = mounts.get("data", mounts)
|
|
existing = mounts.get(f"{mount}/")
|
|
if existing is not None and existing.get("type") != "jwt":
|
|
raise BaoError(f"auth mount {mount}/ exists with a different type; review, do not overwrite")
|
|
_run(bao_bin, ["policy", "write", openbao["policy_name"], openbao["policy_file"]], repo_dir, quiet)
|
|
if existing is None:
|
|
_run(bao_bin, ["auth", "enable", f"-path={mount}",
|
|
f"-description={ccr['id']} KeyCape service JWT login", "jwt"], repo_dir, quiet)
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
config_path = Path(tmp) / "config.json"
|
|
role_path = Path(tmp) / "role.json"
|
|
config_path.write_text(json.dumps(config_payload(ccr, repo_dir), sort_keys=True) + "\n")
|
|
role_path.write_text(json.dumps(role_payload(ccr, repo_dir), sort_keys=True) + "\n")
|
|
_run(bao_bin, ["write", f"auth/{mount}/config", f"@{config_path}"], repo_dir, quiet)
|
|
_run(bao_bin, ["write", f"auth/{mount}/role/{auth['role']}", f"@{role_path}"], repo_dir, quiet)
|
|
mismatches = readback_mismatches(ccr, repo_dir, bao_bin)
|
|
if mismatches:
|
|
raise BaoError("readback mismatch: " + "; ".join(mismatches))
|