bed91ceaf2
- New CCR request type workload-kv-operator-matrix (scripts/operator_matrix_lane.py): exact per-entry create/read/update, OIDC role on the existing netkingdom mount, and exact agent denies appended to the live agent-high-risk-boundary (live content preserved, partial presence refused). credential-change.py now dispatches lane request types through one module table. - CCR-2026-0039 (proposed): hall-of-helix four-entry matrix at tenant coulomb per the founder ruling of 2026-10-09; policy, role and overlay sources. - scripts/fluid_telegram_custody_move.py: silent CAS-0 move of the four undocumented infotech entries, deleting sources only after all verify. - scripts/kv_duplicate_retire.py: silent retirement of activity-core/default/llm-connect-provider-secrets once no live non-admin policy grants it. - Founder session commands recorded in the T03 note. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 469992@bnt-lap001 Assistant-Session: e4b8e420-df0a-441d-83ae-7f19130594d9
358 lines
18 KiB
Python
358 lines
18 KiB
Python
"""workload-kv-operator-matrix CCRs: one attended OIDC role, one per-path KV policy.
|
|
|
|
RPF-WP-0035-T03 (design docs/credential-lane-designs/fluid-telegram-operator-kv.md).
|
|
Loaded by scripts/credential-change.py for request_type workload-kv-operator-matrix.
|
|
Unlike workload-kv-read, the policy is a reviewed matrix of exact entries, each
|
|
with its own capabilities (read, create, update), and the apply also adds exact
|
|
coding-agent denies on every entry's data and metadata path to the shared
|
|
agent-high-risk-boundary policy. Live unrelated boundary content is preserved
|
|
verbatim. Nothing here reads, prints or accepts a secret value; the role is
|
|
written to an existing OIDC mount, never a new one.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import re
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
REQUEST_TYPE = "workload-kv-operator-matrix"
|
|
ALLOWED_AUTH_MOUNTS = {"netkingdom"}
|
|
ALLOWED_CAPABILITIES = ("create", "read", "update")
|
|
SELF_GRANTS = (("auth/token/revoke-self", "update"),)
|
|
BOUNDARY_POLICY = "agent-high-risk-boundary"
|
|
BOUNDARY_FILE = "openbao/policies/agent-high-risk-boundary.hcl"
|
|
CAMPAIGN_RE = re.compile(r"^platform/workloads/[a-z0-9-]+/[a-z0-9-]+/[a-z0-9-]+$")
|
|
ENTRY_RE = re.compile(r"^[a-z0-9][a-z0-9-]*$")
|
|
FIELD_RE = re.compile(r"^[a-z][a-z0-9_]*$")
|
|
POLICY_NAME_RE = re.compile(r"^workload-kv-[a-z0-9-]+-operator$")
|
|
ROLE_RE = re.compile(r"^[a-z0-9][a-z0-9-]*$")
|
|
GROUP_RE = re.compile(r"^[a-z0-9][a-z0-9-]*$")
|
|
OIDC_SCOPES = ["openid", "profile", "email", "groups"]
|
|
# Loopback CLI callbacks and the operator-tunnel UI callback; never a public listener.
|
|
ALLOWED_REDIRECT_URIS = {
|
|
"http://localhost:8250/oidc/callback",
|
|
"http://127.0.0.1:8250/oidc/callback",
|
|
"http://127.0.0.1:18200/ui/vault/auth/netkingdom/oidc/callback",
|
|
}
|
|
MAX_TTL_SECONDS = 15 * 60
|
|
MAX_EXPLICIT_TTL_SECONDS = 30 * 60
|
|
ROLE_KEYS = {
|
|
"role_type", "allowed_redirect_uris", "oidc_scopes", "user_claim", "groups_claim",
|
|
"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")
|
|
|
|
|
|
def _service_jwt_lane():
|
|
spec = importlib.util.spec_from_file_location(
|
|
"service_jwt_lane", Path(__file__).resolve().parent / "service_jwt_lane.py"
|
|
)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
_shared = _service_jwt_lane()
|
|
ttl_seconds = _shared.ttl_seconds
|
|
normalized = _shared.normalized
|
|
BaoError = _shared.BaoError
|
|
_run = _shared._run
|
|
_read_json = _shared._read_json
|
|
|
|
|
|
def _kv_paths(ccr: dict[str, Any], entry: str) -> tuple[str, str]:
|
|
campaign = ccr["openbao"]["campaign_path"]
|
|
mount, rest = campaign.split("/", 1)
|
|
return f"{mount}/data/{rest}/{entry}", f"{mount}/metadata/{rest}/{entry}"
|
|
|
|
|
|
def generated_policy_hcl(ccr: dict[str, Any]) -> str:
|
|
lines = [f"# {ccr['id']} (RPF-WP-0035-T03): exact per-entry operator matrix; no metadata, list or delete."]
|
|
for row in ccr["openbao"]["matrix"]:
|
|
data_path, _ = _kv_paths(ccr, row["entry"])
|
|
caps = ", ".join(f'"{c}"' for c in ALLOWED_CAPABILITIES if c in row["capabilities"])
|
|
lines += [f'path "{data_path}" {{', f" capabilities = [{caps}]", "}"]
|
|
for path, capability in SELF_GRANTS:
|
|
lines += [f'path "{path}" {{', f' capabilities = ["{capability}"]', "}"]
|
|
return "\n".join(lines) + "\n"
|
|
|
|
|
|
def boundary_stanzas(ccr: dict[str, Any]) -> list[str]:
|
|
stanzas = []
|
|
for row in ccr["openbao"]["matrix"]:
|
|
for path in _kv_paths(ccr, row["entry"]):
|
|
stanzas.append(f'path "{path}" {{ capabilities = ["deny"] }}')
|
|
return stanzas
|
|
|
|
|
|
def generated_boundary_overlay(ccr: dict[str, Any]) -> str:
|
|
header = f"# {ccr['id']}: coding-agent denies for {ccr['openbao']['campaign_path']} (data and metadata)."
|
|
return "\n".join([header, *boundary_stanzas(ccr)]) + "\n"
|
|
|
|
|
|
def _flat(text: str) -> set[str]:
|
|
return {" ".join(line.split()) for line in text.splitlines() if line.strip()}
|
|
|
|
|
|
def boundary_presence(ccr: dict[str, Any], policy_text: str) -> tuple[int, int]:
|
|
present = _flat(policy_text)
|
|
stanzas = boundary_stanzas(ccr)
|
|
return sum(1 for s in stanzas if " ".join(s.split()) in present), len(stanzas)
|
|
|
|
|
|
def role_payload(ccr: dict[str, Any], repo_dir: Path) -> dict[str, Any]:
|
|
return _shared._load_json(repo_dir, ccr["openbao"]["auth"]["role_file"], "openbao.auth.role_file", [])
|
|
|
|
|
|
def _validate_matrix(openbao: dict[str, Any], errors: list[str]) -> None:
|
|
matrix = openbao.get("matrix")
|
|
if not isinstance(matrix, list) or not matrix:
|
|
errors.append("openbao.matrix must list the exact entries")
|
|
return
|
|
seen: set[str] = set()
|
|
for row in matrix:
|
|
if not isinstance(row, dict):
|
|
errors.append("openbao.matrix rows must be objects")
|
|
continue
|
|
entry = str(row.get("entry") or "")
|
|
if not ENTRY_RE.match(entry):
|
|
errors.append(f"openbao.matrix entry must be one exact segment: {entry!r}")
|
|
if entry in seen:
|
|
errors.append(f"openbao.matrix entry is duplicated: {entry}")
|
|
seen.add(entry)
|
|
fields = row.get("fields")
|
|
if not isinstance(fields, list) or not fields or not all(FIELD_RE.match(str(f)) for f in fields):
|
|
errors.append(f"openbao.matrix {entry} fields must be safe field names")
|
|
caps = row.get("capabilities")
|
|
if not isinstance(caps, list) or not caps or len(set(caps)) != len(caps):
|
|
errors.append(f"openbao.matrix {entry} capabilities must be a non-empty unique list")
|
|
continue
|
|
unknown = sorted(set(caps) - set(ALLOWED_CAPABILITIES))
|
|
if unknown:
|
|
errors.append(f"openbao.matrix {entry} has unsupported capabilities: {unknown}")
|
|
if "read" not in caps:
|
|
errors.append(f"openbao.matrix {entry} must include read")
|
|
|
|
|
|
def _validate_role(role: dict[str, Any], policy_name: str, errors: list[str]) -> None:
|
|
extra = set(role) - ROLE_KEYS
|
|
if extra:
|
|
errors.append(f"role has keys outside the reviewed set: {sorted(extra)}")
|
|
if role.get("role_type") != "oidc" or role.get("user_claim") != "sub":
|
|
errors.append("role must be role_type oidc with user_claim sub")
|
|
if role.get("groups_claim") != "groups" or role.get("oidc_scopes") != OIDC_SCOPES:
|
|
errors.append(f"role must use groups_claim groups and oidc_scopes {OIDC_SCOPES}")
|
|
uris = role.get("allowed_redirect_uris")
|
|
if not isinstance(uris, list) or not uris or not set(uris) <= ALLOWED_REDIRECT_URIS:
|
|
errors.append("role allowed_redirect_uris must be a subset of the approved loopback/tunnel callbacks")
|
|
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")
|
|
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 = {}
|
|
if set(claims) != {"groups", "/assurance/level"}:
|
|
errors.append("role bound_claims must be exactly groups and /assurance/level")
|
|
groups = claims.get("groups")
|
|
if not isinstance(groups, list) or len(groups) != 1 or not GROUP_RE.match(str(groups[0])):
|
|
errors.append("role bound_claims.groups must be one exact group")
|
|
if claims.get("/assurance/level") != "aal2":
|
|
errors.append("role bound_claims./assurance/level must be aal2")
|
|
ttl = ttl_seconds(role.get("token_ttl"))
|
|
max_ttl = ttl_seconds(role.get("token_max_ttl"))
|
|
explicit = ttl_seconds(role.get("token_explicit_max_ttl"))
|
|
if ttl is None or not 0 < ttl <= MAX_TTL_SECONDS:
|
|
errors.append("role token_ttl must be set and at most 15m")
|
|
if any(v is None or not 0 < v <= MAX_EXPLICIT_TTL_SECONDS for v in (max_ttl, explicit)):
|
|
errors.append("role token_max_ttl and token_explicit_max_ttl must be set and at most 30m")
|
|
elif ttl is not None and ttl > min(max_ttl, explicit):
|
|
errors.append("role token_ttl must not exceed its maximums")
|
|
uses = role.get("token_num_uses")
|
|
if not isinstance(uses, int) or isinstance(uses, bool) or uses < 0:
|
|
errors.append("role token_num_uses must be a non-negative integer")
|
|
|
|
|
|
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"]
|
|
if (ccr.get("target") or {}).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")
|
|
|
|
campaign = str(openbao.get("campaign_path") or "")
|
|
if not CAMPAIGN_RE.match(campaign):
|
|
errors.append("openbao.campaign_path must be platform/workloads/<tenant>/<workload>/<campaign>")
|
|
_validate_matrix(openbao, errors)
|
|
if errors:
|
|
return errors
|
|
|
|
policy_name = str(openbao.get("policy_name") or "")
|
|
if not POLICY_NAME_RE.match(policy_name):
|
|
errors.append("openbao.policy_name must be workload-kv-<...>-operator")
|
|
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")
|
|
elif not (repo_dir / policy_file).exists():
|
|
errors.append(f"openbao.policy_file does not exist: {policy_file}")
|
|
elif normalized((repo_dir / policy_file).read_text(encoding="utf-8")) != normalized(generated_policy_hcl(ccr)):
|
|
errors.append("openbao.policy_file does not match the generated operator matrix policy")
|
|
|
|
overlay_file = str(openbao.get("boundary_overlay_file") or "")
|
|
overlay_path = (repo_dir / overlay_file).resolve()
|
|
if not overlay_file or not overlay_path.exists():
|
|
errors.append("openbao.boundary_overlay_file must name an existing file")
|
|
elif normalized(overlay_path.read_text(encoding="utf-8")) != normalized(generated_boundary_overlay(ccr)):
|
|
errors.append("openbao.boundary_overlay_file does not match the generated agent denies")
|
|
boundary_text = (repo_dir / BOUNDARY_FILE).read_text(encoding="utf-8")
|
|
present, total = boundary_presence(ccr, boundary_text)
|
|
if present != total:
|
|
errors.append(f"{BOUNDARY_FILE} must carry all {total} agent denies for this matrix (has {present})")
|
|
|
|
if auth.get("method") != "oidc":
|
|
errors.append("openbao.auth.method must be oidc")
|
|
if auth.get("mount") not in ALLOWED_AUTH_MOUNTS:
|
|
errors.append(f"openbao.auth.mount must be one of {sorted(ALLOWED_AUTH_MOUNTS)}")
|
|
if not ROLE_RE.match(str(auth.get("role") or "")):
|
|
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")
|
|
role = _shared._load_json(repo_dir, str(auth.get("role_file") or ""), "openbao.auth.role_file", errors)
|
|
if role:
|
|
_validate_role(role, policy_name, errors)
|
|
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": "boundary_append_if_absent", "openbao_path": f"sys/policies/acl/{BOUNDARY_POLICY}",
|
|
"source": openbao["boundary_overlay_file"],
|
|
"payload": {"preserves_live_policy": True, "refuses_partial_presence": True,
|
|
"denies": boundary_stanzas(ccr)}},
|
|
{"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_role_write", "openbao_path": f"auth/{auth['mount']}/role/{auth['role']}",
|
|
"payload": role_payload(ccr, repo_dir),
|
|
"precondition": f"auth/{auth['mount']} exists with type oidc (never enabled here)"},
|
|
{"kind": "readback_compare", "openbao_path": "boundary denies, policy, role",
|
|
"payload": {"prints": "booleans only"}},
|
|
],
|
|
"out_of_scope": [
|
|
"secret value reads or writes (custody moves use their own silent helper)",
|
|
"token creation or login",
|
|
"auth mount enable/config changes",
|
|
"changes to any other role or policy",
|
|
],
|
|
"required_evidence": [
|
|
"CCR id and approval reference",
|
|
"readback equality for boundary denies, policy and role",
|
|
"owner verification (ops-mason MASON-WP-0005-T05): positive matrix, sibling/agent/non-member denial, salt CAS 0",
|
|
],
|
|
}
|
|
|
|
|
|
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"Agent boundary denies ensured: sys/policies/acl/{BOUNDARY_POLICY} ({len(boundary_stanzas(ccr))} exact paths)",
|
|
f"Policy metadata write: sys/policies/acl/{openbao['policy_name']}",
|
|
f"OIDC role metadata write: auth/{auth['mount']}/role/{auth['role']}",
|
|
"Readback matched source for boundary denies, policy and role.",
|
|
"No secret values were read, written, printed, or accepted in argv.",
|
|
]
|
|
|
|
|
|
def readback_mismatches(ccr: dict[str, Any], repo_dir: Path, bao_bin: str) -> list[str]:
|
|
openbao = ccr["openbao"]
|
|
auth = openbao["auth"]
|
|
mismatches: list[str] = []
|
|
boundary = _run(bao_bin, ["policy", "read", BOUNDARY_POLICY], repo_dir, quiet=True, capture=True)
|
|
present, total = boundary_presence(ccr, boundary)
|
|
if present != total:
|
|
mismatches.append("boundary policy lacks agent denies")
|
|
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")
|
|
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)
|
|
elif key in ("allowed_redirect_uris", "oidc_scopes"):
|
|
same = sorted(got or []) == sorted(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"]
|
|
mounts = _read_json(bao_bin, ["auth", "list"], repo_dir)
|
|
mounts = mounts.get("data", mounts)
|
|
if (mounts.get(f"{auth['mount']}/") or {}).get("type") != "oidc":
|
|
raise BaoError(f"auth mount {auth['mount']}/ is absent or not oidc; this lane never enables it")
|
|
|
|
current = _run(bao_bin, ["policy", "read", BOUNDARY_POLICY], repo_dir, quiet=True, capture=True)
|
|
present, total = boundary_presence(ccr, current)
|
|
if 0 < present < total:
|
|
raise BaoError("boundary policy holds some but not all of this matrix's denies; review, do not merge")
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
if present == 0:
|
|
overlay = (repo_dir / openbao["boundary_overlay_file"]).read_text(encoding="utf-8")
|
|
boundary_path = Path(tmp) / "boundary.hcl"
|
|
# Preserve live unrelated policy verbatim; add only the reviewed denies.
|
|
boundary_path.write_text(current.rstrip("\n") + "\n\n" + overlay, encoding="utf-8")
|
|
_run(bao_bin, ["policy", "write", BOUNDARY_POLICY, str(boundary_path)], repo_dir, quiet)
|
|
_run(bao_bin, ["policy", "write", openbao["policy_name"], openbao["policy_file"]], repo_dir, quiet)
|
|
role_path = Path(tmp) / "role.json"
|
|
role_path.write_text(json.dumps(role_payload(ccr, repo_dir), sort_keys=True) + "\n")
|
|
_run(bao_bin, ["write", f"auth/{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))
|