RPF-WP-0035-T02: CCR-2026-0038 service-JWT login lane for secrets-engine-openbao
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s

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
This commit is contained in:
codex 2026-10-09 00:31:47 +02:00
parent 2e7701f155
commit 0b2e7ddb5f
8 changed files with 680 additions and 15 deletions

View file

@ -0,0 +1,98 @@
id: CCR-2026-0038
kind: credential-change-request
schema_version: 1
request_type: service-jwt-login
title: secrets-engine-openbao KeyCape service-JWT login (isolated mount, login-only policy)
status: proposed
created: '2026-10-09'
updated: '2026-10-09'
requester:
agent: railiance-platform
reason: RPF-WP-0035-T02. KEY-WP-0037 finished 2026-10-07 (registration live, issuance
verified 10/10, key-cape e0b5fae); CCR-2026-0033/0034 applied. Design
docs/credential-lane-designs/secrets-engine-service-jwt.md; source bundle prepared
2026-09-27. ops-mason MASON-WP-0009-T05 waits on this lane to recheck invariant (ii).
review:
required: true
required_approvers:
- platform-owner
- key-cape-owner
- secrets-engine-owner
comments: []
target:
domain: infotech
tenant: platform
workload: secrets-engine
environment: production
purpose: Let secrets-engine log in to OpenBao with a KeyCape-issued service JWT and
receive a 5-minute, 8-use token that can only inspect and revoke itself.
openbao:
policy_name: secrets-engine-login-self
policy_file: openbao/policies/secrets-engine-login-self.hcl
guarded_paths:
- platform/data/workloads/keycape/secrets-engine-openbao
- platform/metadata/workloads/keycape/secrets-engine-openbao
auth:
method: jwt
mount: keycape-services
config_file: openbao/auth/keycape-services-config.json
role: secrets-engine-openbao
role_file: openbao/auth/secrets-engine-jwt-role.json
bound_claims_confirmed: false
founder_confirmation_items:
- bound_claims_confirmed stays false until the founder confirms the binding against the
KeyCape registration (aud secrets-engine-openbao, sub service:secrets-engine,
principal_type service, tenant tenant:platform, roles secrets-engine, scope
openbao:login; issuer https://kc.coulomb.social, RS256). applier-dry-run refuses
until then.
- The isolated mount keycape-services is new (ops-mason survey 2026-10-07 found no
auth/jwt or keycape-services mount). The applier refuses if the path exists with a
type other than jwt.
access_frontdoor:
type: none
catalog_id: none
readiness: pending-review
resolvable: false
delivery:
surface: none
target: No value is delivered. The consumer contract file (mount, role, bound_issuer)
is published only after the login proof passes.
risk:
classification: medium
notes:
- Login identity only. The policy holds lookup-self, capabilities-self and revoke-self,
plus explicit deny on the client-secret data and metadata paths (MASON-WP-0009
invariant (ii)); no KV read, policy/auth write, SecretID or child token.
- A JWT may be replayed while valid (15m); residual access is bounded by the 5m/8-use
OpenBao token. In an incident, delete the role, revoke issued tokens; do not disable a
shared mount.
verification:
positive:
- Readback after apply matches source for the policy text, mount type jwt, config and
role (done by applier-apply; prints nothing).
- A secrets-engine --auth service-jwt run (SECRETS-WP-0012-T03 probe-openbao) logs in;
the token's policies are exactly [secrets-engine-login-self]; revoke-self succeeds.
negative:
- sys/capabilities-self for the service token returns deny on both guarded paths and
on sys/policies/acl/*, auth/*/role/*, auth/token/create.
- Wrong audience, subject, tenant, principal_type, role or scope, a coding-agent JWT
and an expired JWT are refused.
- A failed login never falls back to bootstrap/env providers.
activation_conditions:
- Founder approval of this CCR (three reviews) and bound_claims_confirmed true.
evidence: []
lifecycle:
deactivate: Delete auth/keycape-services/role/secrets-engine-openbao, then the policy;
removing the exclusively owned empty mount is a later review.
rotate: Not applicable; the client secret rotates through the CCR-2026-0033 custody child.
compromised: KeyCape disables the registration; platform deletes the role and revokes
issued tokens; tokens are 5m/8-use and self-revoked.
state_hub:
workplan_id: RPF-WP-0035
task_id: RPF-WP-0035-T02
related_workplan: KEY-WP-0037
related:
design: docs/credential-lane-designs/secrets-engine-service-jwt.md
sibling_ccrs:
- CCR-2026-0033
- CCR-2026-0034

View file

@ -1,6 +1,6 @@
# Secrets-engine service JWT login
Status: proposed, not provisioned. Owner: railiance-platform, RPF-WP-0035-T02 (design origin RPF-WP-0032).
Status: source complete as CCR-2026-0038 (proposed), not provisioned. Owner: railiance-platform, RPF-WP-0035-T02 (design origin RPF-WP-0032).
Demand: State Hub message `38b47122-07eb-4a7f-a5df-13a38f50e110`.
## Contract and ownership
@ -150,3 +150,14 @@ its source contract or local managed IDs.
`docs/provisioning-windows/2026-10-04-1000/dependency-requirements.md` names all
source/acceptance/reopen boundaries. RPF-WP-0035-T02 remains wait; its workplan
is blocked on those real dependencies and separate attended execution approval.
## Source complete — 2026-10-09
KEY-WP-0037 finished 2026-10-07 and CCR-2026-0033/0034 are applied, so the
"before writing" preconditions are met. The reviewed bundle is now packaged as
CCR-2026-0038 (`request_type: service-jwt-login`), applied by the delegated
applier (`scripts/service_jwt_lane.py`): policy, `auth enable` of
`keycape-services` only if absent (refuses a foreign type), config, role, then
a value-free readback compare. The self policy gained explicit denies on the
client-secret data/metadata paths (ops-mason MASON-WP-0009 invariant (ii)).
The founder command is recorded in RPF-WP-0035-T02.

View file

@ -1,4 +1,4 @@
# RPF-WP-0035-T02: proposed login-only service policy; not a lane applier.
# CCR-2026-0038 (RPF-WP-0035-T02): login-only service policy; not a lane applier.
path "auth/token/lookup-self" {
capabilities = ["read"]
}
@ -8,3 +8,9 @@ path "sys/capabilities-self" {
path "auth/token/revoke-self" {
capabilities = ["update"]
}
path "platform/data/workloads/keycape/secrets-engine-openbao" {
capabilities = ["deny"]
}
path "platform/metadata/workloads/keycape/secrets-engine-openbao" {
capabilities = ["deny"]
}

View file

@ -516,6 +516,8 @@ def validate_ccr(path: Path) -> tuple[dict[str, Any], list[str], list[str]]:
owner = importlib.util.module_from_spec(spec)
spec.loader.exec_module(owner)
errors.extend(owner.validate_contract(ccr))
elif request_type == "service-jwt-login":
errors.extend(service_jwt_lane().validate(ccr, REPO_DIR))
else:
errors.append("unsupported request_type")
return ccr, errors, warnings
@ -527,14 +529,14 @@ def render_summary(ccr: dict[str, Any], warnings: list[str]) -> str:
frontdoor = ccr["access_frontdoor"]
risk = ccr["risk"]
verification = ccr["verification"]
fields = ", ".join(openbao["fields"])
fields = ", ".join(openbao.get("fields", []))
lines = [
f"Request: {ccr['title']}",
f"CCR: {ccr['id']} ({ccr['status']})",
f"Type: {ccr['request_type']}",
f"Target: {ccr['target']['tenant']}/{ccr['target']['workload']} ({ccr['target']['environment']})",
"Mount/path/field:",
f" {openbao['kv_path']}",
f" {openbao.get('kv_path', '-')}",
f" {fields}",
"Policy:",
f" {openbao.get('policy_name', 'none (attended host delivery)')}",
@ -804,6 +806,19 @@ def string_values(value: Any) -> list[str]:
return []
def service_jwt_lane():
"""Lazy-load scripts/service_jwt_lane.py (request_type service-jwt-login)."""
import importlib.util
spec = importlib.util.spec_from_file_location("service_jwt_lane", REPO_DIR / "scripts/service_jwt_lane.py")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def is_service_jwt(ccr: dict[str, Any]) -> bool:
return ccr.get("request_type") == "service-jwt-login"
def applier_policy_violations(ccr: dict[str, Any]) -> list[str]:
violations: list[str] = []
if ccr.get("request_type") != "workload-kv-read":
@ -921,6 +936,8 @@ def applier_policy_violations(ccr: dict[str, Any]) -> list[str]:
def applier_readiness_blockers(ccr: dict[str, Any]) -> list[str]:
if is_service_jwt(ccr):
return service_jwt_lane().readiness_blockers(ccr, REPO_DIR, APPLIER_DRY_RUN_ALLOWED_STATUSES)
blockers: list[str] = []
status = ccr.get("status")
if status not in APPLIER_DRY_RUN_ALLOWED_STATUSES:
@ -935,6 +952,8 @@ def applier_readiness_blockers(ccr: dict[str, Any]) -> list[str]:
def applier_dry_run_payload(ccr: dict[str, Any], warnings: list[str]) -> dict[str, Any]:
if is_service_jwt(ccr):
return service_jwt_lane().dry_run_payload(ccr, REPO_DIR, warnings)
openbao = ccr["openbao"]
auth = openbao["auth"]
auth_path = f"auth/{auth['mount']}/role/{auth['role']}"
@ -1023,6 +1042,8 @@ def applier_confirmation_phrase(ccr: dict[str, Any]) -> str:
def delegated_apply_details(ccr: dict[str, Any], actor: str) -> list[str]:
if is_service_jwt(ccr):
return service_jwt_lane().apply_details(ccr, actor)
openbao = ccr["openbao"]
auth = openbao["auth"]
return [
@ -1183,6 +1204,13 @@ def render_runbook(payload: dict[str, Any]) -> str:
def run_bao_metadata_apply(ccr: dict[str, Any], bao_bin: str, quiet: bool = False) -> None:
if is_service_jwt(ccr):
lane = service_jwt_lane()
try:
lane.run_apply(ccr, REPO_DIR, bao_bin, quiet=quiet)
except (lane.BaoError, subprocess.CalledProcessError) as exc:
fail(str(exc))
return
openbao = ccr["openbao"]
auth = openbao["auth"]
auth_path = f"auth/{auth['mount']}/role/{auth['role']}"
@ -1272,7 +1300,7 @@ def record_evidence_state_hub(
openbao = ccr["openbao"]
summary = (
f"CCR {ccr['id']} evidence {kind}/{result} by {actor}: "
f"status={ccr['status']} path={openbao['kv_path']} "
f"status={ccr['status']} path={openbao.get('kv_path') or 'auth/' + openbao['auth']['mount']} "
f"policy={openbao['policy_name']}; "
+ "; ".join(details)
)
@ -1451,7 +1479,7 @@ def record_lifecycle_state_hub(
summary = (
f"CCR {ccr['id']} lifecycle {action} by {actor}: "
f"status={ccr['status']} readiness={frontdoor.get('readiness')} "
f"resolvable={frontdoor.get('resolvable') is True} path={openbao['kv_path']} "
f"resolvable={frontdoor.get('resolvable') is True} path={openbao.get('kv_path', '-')} "
f"policy={openbao['policy_name']}; reason={reason}; "
+ "; ".join(details)
)
@ -1687,9 +1715,9 @@ def status_payload(ccr: dict[str, Any], warnings: list[str]) -> dict[str, Any]:
"frontdoor_blockers": frontdoor_blocked_by,
"warnings": warnings,
"openbao": {
"mount": openbao["mount"],
"kv_path": openbao["kv_path"],
"fields": openbao["fields"],
"mount": openbao.get("mount"),
"kv_path": openbao.get("kv_path"),
"fields": openbao.get("fields", []),
"policy_name": openbao.get("policy_name"),
"auth_mount": auth.get("mount"),
"auth_method": auth.get("method"),
@ -1845,7 +1873,7 @@ def decision_template_context(ccr: dict[str, Any]) -> dict[str, str]:
state_hub = ccr.get("state_hub", {})
return {
"id": ccr["id"],
"kv_path": openbao["kv_path"],
"kv_path": openbao.get("kv_path") or f"auth/{auth['mount']}",
"policy_name": openbao.get("policy_name"),
"auth_role_path": f"auth/{auth['mount']}/role/{auth['role']}",
"decision_link": state_hub.get("decision_api_url")
@ -2159,7 +2187,7 @@ def command_decision(args: argparse.Namespace, status: str) -> int:
{
"summary": (
f"CCR {ccr['id']} decision {status} by {args.reviewer}: "
f"path={openbao['kv_path']} policy={openbao['policy_name']} "
f"path={openbao.get('kv_path', '-')} policy={openbao['policy_name']} "
f"fields={','.join(openbao['fields'])} "
f"auth_role=auth/{auth['mount']}/role/{auth['role']}; "
f"{args.comment}"

327
scripts/service_jwt_lane.py Normal file
View file

@ -0,0 +1,327 @@
"""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))

View file

@ -24,7 +24,9 @@ class ServiceJWTContract(unittest.TestCase):
def test_only_self_endpoints_and_pinned_signature_verifier(self):
policy = (ROOT / "openbao/policies/secrets-engine-login-self.hcl").read_text()
grants = re.findall(r'path "([^"]+)"\s*{\s*capabilities = \["([^"]+)"\]', policy)
self.assertEqual(grants, [("auth/token/lookup-self", "read"), ("sys/capabilities-self", "update"), ("auth/token/revoke-self", "update")])
self.assertEqual(grants, [("auth/token/lookup-self", "read"), ("sys/capabilities-self", "update"), ("auth/token/revoke-self", "update"),
("platform/data/workloads/keycape/secrets-engine-openbao", "deny"),
("platform/metadata/workloads/keycape/secrets-engine-openbao", "deny")])
self.assertNotIn("*", policy)
config = json.loads((ROOT / "openbao/auth/keycape-services-config.json").read_text())
self.assertEqual(config, dict(jwks_url="https://kc.coulomb.social/jwks", bound_issuer="https://kc.coulomb.social", jwt_supported_algs=["RS256"]))

View file

@ -0,0 +1,166 @@
"""CCR-2026-0038 service-jwt-login lane: validator, invariant (ii), fake-bao applier."""
from __future__ import annotations
import copy
import importlib.util
import json
import os
import re
import stat
import sys
import tempfile
import unittest
from pathlib import Path
import yaml
ROOT = Path(__file__).resolve().parents[1]
CCR_PATH = ROOT / "credential-change-requests/CCR-2026-0038-secrets-engine-service-jwt-login.yaml"
GUARDED = "platform/data/workloads/keycape/secrets-engine-openbao"
def load(name: str, rel: str):
spec = importlib.util.spec_from_file_location(name, ROOT / rel)
module = importlib.util.module_from_spec(spec)
sys.modules[name] = module
spec.loader.exec_module(module)
return module
lane = load("service_jwt_lane", "scripts/service_jwt_lane.py")
cc = load("credential_change_cli", "scripts/credential-change.py")
def ccr() -> dict:
return yaml.safe_load(CCR_PATH.read_text())
FAKE_BAO = r'''#!/usr/bin/env python3
import json, os, sys
state = json.load(open(os.environ["FAKE_BAO_STATE"]))
open(os.environ["FAKE_BAO_LOG"], "a").write(json.dumps(sys.argv[1:]) + "\n")
args = [a for a in sys.argv[1:] if a != "-format=json"]
if args[:2] == ["auth", "list"]:
print(json.dumps({k: {"type": v} for k, v in state["mounts"].items()}))
elif args[:2] == ["policy", "read"]:
print(state["policy"])
elif args[0] == "read" and args[1].endswith("/config"):
print(json.dumps({"data": state["config"]}))
elif args[0] == "read":
print(json.dumps({"data": state["role"]}))
'''
class ServiceJwtLaneTests(unittest.TestCase):
def test_ccr_validates_and_is_not_yet_applicable(self):
_ccr, errors, _w = cc.validate_ccr(CCR_PATH)
self.assertEqual(errors, [])
blockers = cc.applier_readiness_blockers(_ccr)
self.assertIn("applier requires confirmed OpenBao auth binding", blockers)
self.assertTrue(any("got proposed" in b for b in blockers))
def test_invariant_ii_policy_has_no_capability_on_the_client_secret_path(self):
policy = (ROOT / "openbao/policies/secrets-engine-login-self.hcl").read_text()
stanzas = re.findall(r'path "([^"]+)"\s*{\s*capabilities = \[([^\]]*)\]', policy)
for path, caps in stanzas:
if path.startswith("platform/"):
self.assertEqual(caps.strip(), '"deny"', path)
self.assertIn((GUARDED, '"deny"'), stanzas)
granted = {p for p, c in stanzas if c.strip() != '"deny"'}
self.assertEqual(granted, {p for p, _ in lane.SELF_GRANTS})
self.assertNotIn("*", policy)
def test_validator_rejects_widening(self):
cases = []
c = ccr(); c["openbao"]["guarded_paths"] = ["platform/data/workloads/*"]; cases.append(c)
c = ccr(); c["openbao"]["auth"]["mount"] = "netkingdom"; cases.append(c)
c = ccr(); c["openbao"]["policy_name"] = "platform-admin"; cases.append(c)
c = ccr(); c["target"]["environment"] = "staging"; cases.append(c)
for case in cases:
self.assertTrue(lane.validate(case, ROOT), case["openbao"])
def test_validator_rejects_bad_role_file(self):
with tempfile.TemporaryDirectory() as tmp:
repo = Path(tmp)
(repo / "openbao/auth").mkdir(parents=True)
(repo / "openbao/policies").mkdir(parents=True)
c = ccr()
(repo / c["openbao"]["policy_file"]).write_text(lane.generated_policy_hcl(c))
(repo / c["openbao"]["auth"]["config_file"]).write_text(
(ROOT / c["openbao"]["auth"]["config_file"]).read_text())
base = json.loads((ROOT / c["openbao"]["auth"]["role_file"]).read_text())
self.assertEqual(self._role_errors(repo, c, base), [])
for mutate in (
lambda r: r.update(token_ttl="1h"),
lambda r: r.update(token_no_default_policy=False),
lambda r: r.update(token_policies=["secrets-engine-login-self", "default"]),
lambda r: r.update(token_period="5m"),
lambda r: r["bound_claims"].update(roles="*"),
lambda r: r.update(bound_audiences=["other"]),
lambda r: r.update(token_num_uses=0),
):
role = copy.deepcopy(base); mutate(role)
self.assertTrue(self._role_errors(repo, c, role), role)
def _role_errors(self, repo, c, role):
(repo / c["openbao"]["auth"]["role_file"]).write_text(json.dumps(role))
return lane.validate(c, repo)
def _fake(self, tmp: Path, mounts: dict, policy: str | None = None, role_ttl=300):
c = ccr()
role = lane.role_payload(c, ROOT); role["token_ttl"] = role_ttl
role["token_max_ttl"] = role["token_explicit_max_ttl"] = 300
state = {"mounts": mounts, "policy": policy if policy is not None else lane.generated_policy_hcl(c),
"config": lane.config_payload(c, ROOT), "role": role}
(tmp / "state.json").write_text(json.dumps(state))
bao = tmp / "bao"; bao.write_text(FAKE_BAO); bao.chmod(bao.stat().st_mode | stat.S_IEXEC)
os.environ["FAKE_BAO_STATE"] = str(tmp / "state.json")
os.environ["FAKE_BAO_LOG"] = str(tmp / "log")
return c, str(bao)
def _log(self, tmp: Path):
return [json.loads(l) for l in (tmp / "log").read_text().splitlines()]
def test_apply_enables_absent_mount_writes_in_order_and_reads_back(self):
with tempfile.TemporaryDirectory() as t:
tmp = Path(t)
c, bao = self._fake(tmp, {"keycape-services/": "jwt"})
# first auth list sees no mount; later readback sees jwt
state = json.loads((tmp / "state.json").read_text())
state["mounts"] = {}
(tmp / "state.json").write_text(json.dumps(state))
with self.assertRaises(lane.BaoError): # readback: mount absent in fake
lane.run_apply(c, ROOT, bao, quiet=True)
verbs = [a[:2] for a in self._log(tmp)]
self.assertEqual(verbs[:5], [["auth", "list"], ["policy", "write"], ["auth", "enable"],
["write", "auth/keycape-services/config"],
["write", "auth/keycape-services/role/secrets-engine-openbao"]])
self.assertIn("jwt", self._log(tmp)[2])
def test_apply_on_existing_jwt_mount_skips_enable_and_passes_readback(self):
with tempfile.TemporaryDirectory() as t:
tmp = Path(t)
c, bao = self._fake(tmp, {"keycape-services/": "jwt"})
lane.run_apply(c, ROOT, bao, quiet=True)
verbs = [a[:2] for a in self._log(tmp)]
self.assertNotIn(["auth", "enable"], verbs)
for argv in self._log(tmp):
self.assertFalse(any("token" in a.lower() and "=" in a for a in argv), argv)
def test_apply_refuses_foreign_mount_type_before_any_write(self):
with tempfile.TemporaryDirectory() as t:
tmp = Path(t)
c, bao = self._fake(tmp, {"keycape-services/": "oidc"})
with self.assertRaises(lane.BaoError):
lane.run_apply(c, ROOT, bao, quiet=True)
self.assertEqual([a[:2] for a in self._log(tmp)], [["auth", "list"]])
def test_readback_detects_policy_drift(self):
with tempfile.TemporaryDirectory() as t:
tmp = Path(t)
c, bao = self._fake(tmp, {"keycape-services/": "jwt"}, policy='path "platform/data/*" {}')
self.assertIn("policy text differs from source", lane.readback_mismatches(c, ROOT, bao))
if __name__ == "__main__":
unittest.main()

View file

@ -8,7 +8,7 @@ status: blocked
flavor: implementation
owner: codex
created: "2026-09-05"
updated: "2026-10-04"
updated: "2026-10-09"
related:
- RPF-WP-0032
- RPF-WP-0033
@ -47,8 +47,7 @@ status: wait
needs_human: true
decision_id: 62aa95e4-3ccc-4867-8806-5e3db005b483
depends_on: [KEY-WP-0037-T02, KEY-WP-0037-T03, KEY-WP-0037-T04]
blocked_on: message-from:secrets-engine
blocking_reason: "KEY-WP-0037 now owns protected custody, real registration and native issuance. Platform issuer-reader source CCR-2026-0033 is prepared. secrets-engine must accept exact consumer host/file/delivery/rotation coordinates before the separate custody child is finalized; KeyCape exact-primary identity repair and separate live go remain."
blocking_reason: "Source ready (CCR-2026-0038, service-jwt-login applier type). Waits on founder approval + binding confirmation + delegated apply (one command, see 2026-10-09 note), then the login proof via a real KeyCape service JWT (SECRETS-WP-0012-T03 probe-openbao)."
priority: high
state_hub_task_id: "e0c82ea9-05a5-537e-8fbf-67534263572e"
```
@ -106,6 +105,34 @@ rotation/revocation ownership, live order and rollback. Attended custody child
founder decision id recorded in the approved CCR. Nothing executed, generated or
applied; nothing added to active GitOps. The founder approval it needs is T09.
2026-10-09: KEY-WP-0037 finished 2026-10-07 (registration live, issuance
verified 10/10, key-cape e0b5fae); CCR-2026-0033/0034 applied. Source for this
lane is now complete as **CCR-2026-0038** (proposed,
`credential-change-requests/CCR-2026-0038-secrets-engine-service-jwt-login.yaml`).
The delegated applier gained a narrow `service-jwt-login` request type
(`scripts/service_jwt_lane.py`): auth mount fixed to `keycape-services`, policy
must equal the generated login-self policy, role/config keys allow-listed,
TTLs at most 15m, uses 1..16, `token_no_default_policy`, refusal if the mount
exists with another type, and a value-free readback compare after the writes.
Policy `secrets-engine-login-self` now also carries explicit `deny` on
`platform/{data,metadata}/workloads/keycape/secrets-engine-openbao` so
ops-mason invariant (ii) (no capability on the client-secret path) holds even
if identity groups were ever attached. Tests: `tests/test_service_jwt_lane.py`
(8) plus the updated contract test; full suite unchanged apart from 5
pre-existing failures unrelated to this lane.
Founder apply (one line; approval, binding confirmation and silent apply):
```bash
cd /home/worsch/railiance-platform && python3 scripts/credential-change.py approve CCR-2026-0038 --reviewer "platform-owner, key-cape-owner, secrets-engine-owner (founder Bernd Worsch)" --comment "Approve service-JWT login lane: mount keycape-services, role secrets-engine-openbao, policy secrets-engine-login-self (self endpoints + deny on the client-secret path)." && python3 scripts/credential-change.py confirm-binding CCR-2026-0038 --reviewer founder --comment "Bound claims match KeyCape registration: aud secrets-engine-openbao, sub service:secrets-engine, principal_type service, tenant tenant:platform, roles secrets-engine, scope openbao:login; issuer https://kc.coulomb.social RS256." && warden access openbao-platform-admin-login --exec -- python3 /home/worsch/railiance-platform/scripts/credential-change.py applier-apply CCR-2026-0038 --confirm "DELEGATED APPLY CCR-2026-0038" --actor founder --record-state-hub --quiet
```
Done after that: login proof with a real KeyCape service JWT
(SECRETS-WP-0012-T03 probe-openbao; policies exactly
[secrets-engine-login-self], capabilities-self deny on the guarded paths,
revoke-self), negative claim checks, then publish the consumer contract file
and ops-mason MASON-WP-0009-T05 rechecks invariant (ii).
## Implement the platform operator-write CCR contract and Fluid lane
```task