7e7ca8be51
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 306568@bnt-lap001 Assistant-Session: 8bdf8150-ea56-41e7-9132-13f04f6b0b91
474 lines
22 KiB
Python
474 lines
22 KiB
Python
"""Self-service CCR intake: submit, standard-lane review and id checks.
|
|
|
|
RPF-WP-0061-T02. Lets a client file a CCR and obtain the platform-owner review
|
|
for in-profile lanes without a railiance-platform worker session. The rules
|
|
live in registry/ccr-standard-profile.yaml; this module only evaluates them.
|
|
Nothing here touches OpenBao or grants approval: the CCR stays `proposed`
|
|
until the founder decision is synced.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import urllib.request
|
|
from collections import defaultdict
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
REPO_DIR = Path(__file__).resolve().parents[1]
|
|
DEFAULT_PROFILE = REPO_DIR / "registry/ccr-standard-profile.yaml"
|
|
CCR_ID_RE = re.compile(r"^CCR-(\d{4})-(\d{4})$")
|
|
CCR_FILE_RE = re.compile(r"^(CCR-\d{4}-\d{4})(-[a-z0-9-]+)?\.ya?ml$")
|
|
SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9-]*$")
|
|
TTL_UNITS = {"s": 1, "m": 60, "h": 3600, "d": 86400}
|
|
PLATFORM_OWNER = "railiance-platform-owner"
|
|
PLATFORM_TOPIC_ID = "ca369340-a64e-442e-98f1-a4fa7dc74a38"
|
|
|
|
|
|
def load_profile(path: Path = DEFAULT_PROFILE) -> tuple[dict[str, Any], str]:
|
|
raw = path.read_bytes()
|
|
return yaml.safe_load(raw), "sha256:" + hashlib.sha256(raw).hexdigest()
|
|
|
|
|
|
def ttl_seconds(value: Any) -> int | None:
|
|
if not isinstance(value, str) or not re.match(r"^[1-9][0-9]*[smhd]$", value):
|
|
return None
|
|
return int(value[:-1]) * TTL_UNITS[value[-1]]
|
|
|
|
|
|
def bound_list(auth: dict[str, Any], key: str) -> list[Any]:
|
|
claims = auth.get("bound_claims") or {}
|
|
return list(claims.get(key) or auth.get(f"bound_{key}") or [])
|
|
|
|
|
|
def ccr_files(directory: Path) -> list[Path]:
|
|
return sorted(p for p in directory.glob("CCR-*.y*ml") if CCR_FILE_RE.match(p.name))
|
|
|
|
|
|
def reader_conflicts(
|
|
ccr: dict[str, Any], directory: Path, profile: dict[str, Any], cc: Any
|
|
) -> list[str]:
|
|
kv_path = (ccr.get("openbao") or {}).get("kv_path")
|
|
statuses = set(profile["reader_conflict_statuses"])
|
|
conflicts = []
|
|
for path in ccr_files(directory):
|
|
other = cc.load_yaml(path)
|
|
if other.get("id") == ccr.get("id"):
|
|
continue
|
|
if (other.get("openbao") or {}).get("kv_path") == kv_path and other.get("status") in statuses:
|
|
conflicts.append(f"{other.get('id')} ({other.get('status')})")
|
|
return conflicts
|
|
|
|
|
|
def standard_violations(
|
|
ccr: dict[str, Any], profile: dict[str, Any], directory: Path, cc: Any
|
|
) -> list[str]:
|
|
"""Return every reason the CCR falls outside the standard lane profile."""
|
|
v: list[str] = []
|
|
if ccr.get("request_type") != profile["request_type"]:
|
|
return [f"request_type must be {profile['request_type']}"]
|
|
openbao = ccr.get("openbao") or {}
|
|
auth = openbao.get("auth") or {}
|
|
want = profile["auth"]
|
|
kv_path = openbao.get("kv_path") or ""
|
|
if openbao.get("mount") != profile["mount"]:
|
|
v.append(f"mount must be {profile['mount']}")
|
|
if not any(re.match(p, kv_path) for p in profile["kv_path_patterns"]):
|
|
v.append(f"kv_path {kv_path!r} is outside the standard workload/supplier prefixes")
|
|
if openbao.get("metadata_read", True) is not profile["metadata_read"]:
|
|
v.append("metadata_read must be false")
|
|
if bool(openbao.get("token_capabilities_self", False)) is not profile["token_capabilities_self"]:
|
|
v.append("token_capabilities_self is not standard")
|
|
policy_name = openbao.get("policy_name") or ""
|
|
if not policy_name.startswith(profile["policy_name_prefix"]):
|
|
v.append(f"policy_name must start with {profile['policy_name_prefix']}")
|
|
policy_file = openbao.get("policy_file")
|
|
if not policy_file:
|
|
v.append("policy_file is required")
|
|
else:
|
|
policy_path = REPO_DIR / policy_file
|
|
if not policy_path.exists():
|
|
v.append(f"policy_file {policy_file} does not exist")
|
|
elif cc.normalized_policy_body(policy_path.read_text()) != cc.normalized_policy_body(
|
|
cc.generated_policy_hcl(ccr)
|
|
):
|
|
v.append("policy_file differs from the generated policy")
|
|
if auth.get("method") != want["method"] or auth.get("mount") != want["mount"]:
|
|
v.append(f"auth must be {want['method']} on mount {want['mount']}")
|
|
if not str(auth.get("role") or "").startswith(want["role_prefix"]):
|
|
v.append(f"auth role must start with {want['role_prefix']}")
|
|
if auth.get("audience") != want["audience"]:
|
|
v.append(f"audience must be {want['audience']}")
|
|
for key in ("ttl", "token_max_ttl", "token_explicit_max_ttl"):
|
|
seconds = ttl_seconds(auth.get(key))
|
|
if seconds is None or seconds > want["max_ttl_seconds"]:
|
|
v.append(f"{key} must be set and at most {want['max_ttl_seconds']}s")
|
|
if auth.get("token_no_default_policy") is not want["token_no_default_policy"]:
|
|
v.append("token_no_default_policy must be true")
|
|
if auth.get("policies") != [policy_name]:
|
|
v.append("auth policies must be exactly [policy_name]")
|
|
accounts = bound_list(auth, "service_account_names")
|
|
namespaces = bound_list(auth, "service_account_namespaces")
|
|
if len(accounts) != want["service_accounts"] or "*" in accounts:
|
|
v.append(f"exactly {want['service_accounts']} exact service account required")
|
|
if len(namespaces) != want["namespaces"] or "*" in namespaces:
|
|
v.append(f"exactly {want['namespaces']} exact namespace required")
|
|
forbidden = sorted(set(namespaces) & set(want["forbidden_namespaces"]))
|
|
if forbidden:
|
|
v.append(f"namespace {forbidden} is reserved for platform components")
|
|
risk = (ccr.get("risk") or {}).get("classification")
|
|
if risk not in profile["allowed_risk_classifications"]:
|
|
v.append(f"risk classification {risk!r} needs a platform review")
|
|
plan_ref = ccr.get("plan_ref")
|
|
if plan_ref is not None and not (
|
|
isinstance(plan_ref, dict)
|
|
and all(isinstance(plan_ref.get(k), str) and plan_ref.get(k) for k in ("repo", "path", "commit"))
|
|
and re.match(r"^[0-9a-f]{7,40}$", plan_ref["commit"])
|
|
):
|
|
v.append("plan_ref must have non-empty repo, path and a hex commit")
|
|
conflicts = reader_conflicts(ccr, directory, profile, cc)
|
|
if conflicts:
|
|
v.append(f"kv_path already has a reader: {', '.join(conflicts)}")
|
|
return v
|
|
|
|
|
|
def duplicate_ids(directory: Path, cc: Any) -> dict[str, list[str]]:
|
|
seen: dict[str, list[str]] = defaultdict(list)
|
|
for path in ccr_files(directory):
|
|
seen[str(cc.load_yaml(path).get("id"))].append(path.name)
|
|
prefix = CCR_FILE_RE.match(path.name).group(1)
|
|
if prefix != cc.load_yaml(path).get("id"):
|
|
seen[f"{prefix} (file name vs id)"].append(path.name)
|
|
return {k: v for k, v in seen.items() if len(v) > 1 or "file name" in k}
|
|
|
|
|
|
def next_ccr_id(directory: Path, year: int) -> str:
|
|
numbers = [0]
|
|
for path in ccr_files(directory):
|
|
match = CCR_ID_RE.match(CCR_FILE_RE.match(path.name).group(1))
|
|
if match and int(match.group(1)) == year:
|
|
numbers.append(int(match.group(2)))
|
|
return f"CCR-{year}-{max(numbers) + 1:04d}"
|
|
|
|
|
|
def record_review(path: Path, ccr: dict[str, Any], violations: list[str], profile: dict[str, Any], digest: str, cc: Any) -> None:
|
|
label = f"{profile['profile']} v{profile['version']} {digest}"
|
|
if violations:
|
|
decision = "needs_platform_review"
|
|
comment = f"Outside {label}: " + "; ".join(violations)
|
|
else:
|
|
decision = "approved"
|
|
comment = (f"Standard lane: passes {label}. Platform-owner source review only; "
|
|
"founder approval and confirm-binding still required.")
|
|
comments = ccr.setdefault("review", {}).setdefault("comments", [])
|
|
comments.append({
|
|
"at": cc.utc_now(),
|
|
"reviewer": f"{PLATFORM_OWNER} (policy-as-code {profile['profile']} v{profile['version']})",
|
|
"decision": decision,
|
|
"comment": comment,
|
|
})
|
|
ccr["updated"] = datetime.now(timezone.utc).date().isoformat()
|
|
cc.dump_yaml(path, ccr)
|
|
|
|
|
|
def notify_platform(ccr: dict[str, Any], violations: list[str], base_url: str) -> None:
|
|
payload = {
|
|
"from_agent": (ccr.get("requester") or {}).get("agent", "ccr-intake"),
|
|
"to_agent": "railiance-platform",
|
|
"subject": f"{ccr['id']} needs platform review (outside standard lane profile)",
|
|
"body": f"{ccr['id']} {ccr.get('title')}: " + "; ".join(violations),
|
|
}
|
|
request = urllib.request.Request(
|
|
f"{base_url.rstrip('/')}/messages/", data=json.dumps(payload).encode(),
|
|
headers={"Content-Type": "application/json"}, method="POST")
|
|
with urllib.request.urlopen(request, timeout=15):
|
|
pass
|
|
|
|
|
|
def review_standard(path: Path, cc: Any, profile_path: Path = DEFAULT_PROFILE,
|
|
write: bool = True) -> list[str]:
|
|
ccr, _warnings = cc.validate_or_exit(path)
|
|
if ccr.get("status") != "proposed":
|
|
cc.fail(f"review-standard requires status proposed, got {ccr.get('status')}")
|
|
profile, digest = load_profile(profile_path)
|
|
violations = standard_violations(ccr, profile, path.parent, cc)
|
|
if write:
|
|
record_review(path, ccr, violations, profile, digest, cc)
|
|
return violations
|
|
|
|
|
|
def command_review_standard(args: argparse.Namespace, cc: Any) -> int:
|
|
path = cc.resolve_ccr(args.ref)
|
|
violations = review_standard(path, cc, Path(args.profile), write=not args.check)
|
|
if not violations:
|
|
print(f"[OK] {path.name}: standard lane; platform-owner review recorded"
|
|
if not args.check else f"[OK] {path.name}: standard lane")
|
|
return 0
|
|
for item in violations:
|
|
print(f"[NEEDS-PLATFORM] {path.name}: {item}")
|
|
if args.notify and not args.check:
|
|
notify_platform(cc.load_yaml(path), violations, args.state_hub_url)
|
|
print("[OK] railiance-platform notified")
|
|
return 2
|
|
|
|
|
|
def command_check_ids(args: argparse.Namespace, cc: Any) -> int:
|
|
duplicates = duplicate_ids(cc.ccr_dir(), cc)
|
|
for key, names in duplicates.items():
|
|
print(f"[ERROR] {key}: {names}")
|
|
if not duplicates:
|
|
print("[OK] CCR ids unique and matching file names")
|
|
return 1 if duplicates else 0
|
|
|
|
|
|
def tree_digests(root: Path) -> dict[str, str]:
|
|
return {
|
|
str(path.relative_to(root)): hashlib.sha256(path.read_bytes()).hexdigest()
|
|
for path in root.rglob("*")
|
|
if path.is_file() and ".git" not in path.relative_to(root).parts
|
|
and "__pycache__" not in path.parts
|
|
}
|
|
|
|
|
|
def branch_gate(main_root: Path, stem: str, cc: Any, profile_path: Path = DEFAULT_PROFILE) -> list[str]:
|
|
"""CI gate for a ccr/<stem> branch checked out at REPO_DIR (RPF-WP-0061-T03).
|
|
|
|
The branch may add exactly one new CCR and its new policy file, nothing
|
|
else; the CCR must be an in-profile submission from a listed submitter with
|
|
a policy-as-code review against the current profile digest.
|
|
"""
|
|
errors: list[str] = []
|
|
ccr_rel = f"credential-change-requests/{stem}.yaml"
|
|
path = REPO_DIR / ccr_rel
|
|
if not CCR_FILE_RE.match(path.name) or not path.exists():
|
|
return [f"branch ccr/{stem} must add {ccr_rel}"]
|
|
if (main_root / ccr_rel).exists():
|
|
errors.append(f"{ccr_rel} already exists on main")
|
|
ccr, validation_errors, _warnings = cc.validate_ccr(path)
|
|
errors.extend(f"validate: {item}" for item in validation_errors)
|
|
policy_rel = (ccr.get("openbao") or {}).get("policy_file")
|
|
allowed = {ccr_rel} | ({policy_rel} if policy_rel else set())
|
|
branch, main = tree_digests(REPO_DIR), tree_digests(main_root)
|
|
changed = {k for k in branch.keys() | main.keys() if branch.get(k) != main.get(k)}
|
|
for extra in sorted(changed - allowed):
|
|
errors.append(f"ccr branch may not change {extra}")
|
|
if policy_rel and policy_rel in main and main[policy_rel] != branch.get(policy_rel):
|
|
errors.append(f"{policy_rel} exists on main; a submission may not modify it")
|
|
if ccr.get("status") != "proposed":
|
|
errors.append(f"status must be proposed, got {ccr.get('status')}")
|
|
profile, digest = load_profile(profile_path)
|
|
if (ccr.get("requester") or {}).get("agent") not in profile["submitters"]:
|
|
errors.append(f"requester is not a listed submitter {profile['submitters']}")
|
|
errors.extend(standard_violations(ccr, profile, path.parent, cc))
|
|
comments = (ccr.get("review") or {}).get("comments") or []
|
|
if len(comments) != 1 or comments[0].get("decision") != "approved" or digest not in str(
|
|
comments[0].get("comment")
|
|
):
|
|
errors.append("review must hold exactly the policy-as-code approval for the current profile")
|
|
errors.extend(f"duplicate id: {k} {v}" for k, v in duplicate_ids(path.parent, cc).items())
|
|
return errors
|
|
|
|
|
|
def command_check_branch(args: argparse.Namespace, cc: Any) -> int:
|
|
stem = args.branch.removeprefix("refs/heads/").removeprefix("ccr/")
|
|
errors = branch_gate(Path(args.main_root).resolve(), stem, cc)
|
|
for item in errors:
|
|
print(f"[FAIL] ccr/{stem}: {item}")
|
|
if not errors:
|
|
print(f"[OK] ccr/{stem}: one in-profile CCR, nothing else changed")
|
|
return 1 if errors else 0
|
|
|
|
|
|
def decision_payload(ccr: dict[str, Any], cc: Any) -> dict[str, Any]:
|
|
openbao = ccr["openbao"]
|
|
auth = openbao["auth"]
|
|
review = ccr["review"]["comments"][-1]
|
|
description = "\n".join([
|
|
f"Founder decision for {ccr['id']}: {ccr.get('title')}.",
|
|
"",
|
|
"Non-secret request summary:",
|
|
f"- Requester: {(ccr.get('requester') or {}).get('agent')}",
|
|
f"- OpenBao path: {openbao.get('kv_path')} fields {openbao.get('fields')}",
|
|
f"- Policy: {openbao.get('policy_name')} ({openbao.get('policy_file')})",
|
|
f"- Auth: {auth.get('method')} role {auth.get('role')}, SA {bound_list(auth, 'service_account_names')}"
|
|
f" in {bound_list(auth, 'service_account_namespaces')}, ttl {auth.get('ttl')}",
|
|
f"- Risk: {(ccr.get('risk') or {}).get('classification')}",
|
|
f"- Platform review: {review.get('reviewer')}: {review.get('comment')}",
|
|
"",
|
|
# The link line would be a placeholder: the decision does not exist yet.
|
|
*[line for line in cc.render_decision_templates(ccr).splitlines()
|
|
if not line.startswith("Decision link:")],
|
|
"",
|
|
"Approval permits apply after confirm-binding; it exposes and stores no value.",
|
|
])
|
|
return {
|
|
"topic_id": PLATFORM_TOPIC_ID,
|
|
"title": f"Review {ccr['id']} {ccr.get('title')}",
|
|
"description": description,
|
|
"escalation_note": "Founder decision required before OpenBao credential lane apply.",
|
|
}
|
|
|
|
|
|
def request_decision(path: Path, cc: Any, base_url: str) -> dict[str, Any]:
|
|
"""Open the founder's State Hub decision for a platform-reviewed CCR (RPF-WP-0061-T04)."""
|
|
ccr, _warnings = cc.validate_or_exit(path)
|
|
if ccr.get("status") != "proposed":
|
|
cc.fail(f"request-decision requires status proposed, got {ccr.get('status')}")
|
|
if (ccr.get("state_hub") or {}).get("decision_id"):
|
|
cc.fail(f"{ccr['id']} already has State Hub decision {ccr['state_hub']['decision_id']}")
|
|
comments = (ccr.get("review") or {}).get("comments") or []
|
|
last = comments[-1] if comments else {}
|
|
if last.get("decision") != "approved" or not str(last.get("reviewer", "")).startswith(PLATFORM_OWNER):
|
|
cc.fail("request-decision requires a platform-owner approval as the latest review comment")
|
|
decision = cc.state_hub_post_json(base_url, "/decisions/", decision_payload(ccr, cc))
|
|
if not decision.get("id") or decision.get("status") not in (None, "open"):
|
|
cc.fail(f"State Hub returned an unexpected decision: {decision.get('id')} {decision.get('status')}")
|
|
state_hub = ccr.setdefault("state_hub", {})
|
|
state_hub["decision_id"] = decision["id"]
|
|
state_hub["decision_api_url"] = f"{base_url.rstrip('/')}/decisions/{decision['id']}"
|
|
ccr["updated"] = datetime.now(timezone.utc).date().isoformat()
|
|
cc.dump_yaml(path, ccr)
|
|
return decision
|
|
|
|
|
|
def command_request_decision(args: argparse.Namespace, cc: Any) -> int:
|
|
path = cc.resolve_ccr(args.ref)
|
|
decision = request_decision(path, cc, args.state_hub_url)
|
|
print(f"[OK] {path.name}: State Hub decision {decision['id']} opened; commit the CCR file")
|
|
return 0
|
|
|
|
|
|
def standard_policy_hcl(kv_path: str, cc: Any) -> str:
|
|
"""The exact HCL review-standard compares against, for a standard-lane path."""
|
|
return cc.generated_policy_hcl({
|
|
"request_type": "workload-kv-read",
|
|
"openbao": {"mount": kv_path.split("/", 1)[0], "kv_path": kv_path,
|
|
"metadata_read": False, "token_self_lifecycle": True},
|
|
})
|
|
|
|
|
|
def command_render_policy(args: argparse.Namespace, cc: Any) -> int:
|
|
profile, _digest = load_profile(Path(args.profile))
|
|
if not any(re.match(p, args.kv_path) for p in profile["kv_path_patterns"]):
|
|
cc.fail(f"kv_path {args.kv_path!r} is outside the standard workload/supplier prefixes")
|
|
sys.stdout.write(standard_policy_hcl(args.kv_path, cc))
|
|
return 0
|
|
|
|
|
|
def git(*argv: str) -> None:
|
|
subprocess.run(["git", "-C", str(REPO_DIR), *argv], check=True)
|
|
|
|
|
|
def submit(args: argparse.Namespace, cc: Any) -> tuple[Path, list[str]]:
|
|
profile, _digest = load_profile(Path(args.profile))
|
|
if args.requester not in profile["submitters"]:
|
|
cc.fail(f"requester {args.requester!r} may not self-submit; allowed: {profile['submitters']}")
|
|
if not SLUG_RE.match(args.slug):
|
|
cc.fail("slug must be lowercase letters, digits and dashes")
|
|
source = cc.resolve_repo_path(args.ccr)
|
|
errors: list[str] = []
|
|
cc.scan_for_secrets(source, errors)
|
|
if errors:
|
|
cc.fail("; ".join(errors))
|
|
ccr = cc.load_yaml(source)
|
|
directory = cc.ccr_dir()
|
|
today = datetime.now(timezone.utc).date()
|
|
ccr_id = next_ccr_id(directory, today.year)
|
|
ccr.update(id=ccr_id, status="proposed", created=today.isoformat(), updated=today.isoformat())
|
|
ccr.setdefault("requester", {})["agent"] = args.requester
|
|
ccr.setdefault("review", {})["comments"] = []
|
|
openbao = ccr.get("openbao") or {}
|
|
if args.policy:
|
|
policy_name = openbao.get("policy_name") or cc.fail("openbao.policy_name is required")
|
|
target = REPO_DIR / "openbao/policies" / f"{policy_name}.hcl"
|
|
policy_text = cc.resolve_repo_path(args.policy).read_text()
|
|
cc.reject_secret_text(policy_text, "policy file")
|
|
if target.exists() and target.read_text() != policy_text:
|
|
cc.fail(f"{target.relative_to(REPO_DIR)} exists with different content")
|
|
target.write_text(policy_text)
|
|
openbao["policy_file"] = str(target.relative_to(REPO_DIR))
|
|
destination = directory / f"{ccr_id}-{args.slug}.yaml"
|
|
cc.dump_yaml(destination, ccr)
|
|
violations = review_standard(destination, cc, Path(args.profile))
|
|
return destination, violations
|
|
|
|
|
|
def command_submit(args: argparse.Namespace, cc: Any) -> int:
|
|
destination, violations = submit(args, cc)
|
|
ccr = cc.load_yaml(destination)
|
|
print(f"[OK] filed {ccr['id']} -> {destination.relative_to(REPO_DIR)}")
|
|
for item in violations:
|
|
print(f"[NEEDS-PLATFORM] {item}")
|
|
if args.commit or args.push:
|
|
branch = f"ccr/{destination.stem}"
|
|
git("switch", "-c", branch)
|
|
paths = [str(destination.relative_to(REPO_DIR))]
|
|
if ccr["openbao"].get("policy_file"):
|
|
paths.append(ccr["openbao"]["policy_file"])
|
|
git("add", *paths)
|
|
git("commit", "-m", f"File {ccr['id']}: {ccr.get('title')}", "--", *paths)
|
|
if args.push:
|
|
git("push", "-u", "origin", branch)
|
|
if violations and args.notify:
|
|
notify_platform(ccr, violations, args.state_hub_url)
|
|
return 0
|
|
|
|
|
|
def add_parsers(sub: Any, cc: Any, state_hub_url: str) -> None:
|
|
review = sub.add_parser(
|
|
"review-standard",
|
|
help="Check a proposed CCR against the standard lane profile; record the platform-owner review",
|
|
)
|
|
review.add_argument("ref")
|
|
review.add_argument("--profile", default=str(DEFAULT_PROFILE))
|
|
review.add_argument("--check", action="store_true", help="report only; do not write the review")
|
|
review.add_argument("--notify", action="store_true", help="message railiance-platform when out of profile")
|
|
review.add_argument("--state-hub-url", default=state_hub_url)
|
|
review.set_defaults(func=lambda args: command_review_standard(args, cc))
|
|
|
|
check_ids = sub.add_parser("check-ids", help="Fail on duplicate CCR ids or id/file name mismatch")
|
|
check_ids.set_defaults(func=lambda args: command_check_ids(args, cc))
|
|
|
|
check_branch = sub.add_parser(
|
|
"check-branch", help="CI gate: a ccr/* branch adds exactly one in-profile CCR and its policy"
|
|
)
|
|
check_branch.add_argument("--branch", required=True, help="ccr/<id>-<slug> or refs/heads/ccr/...")
|
|
check_branch.add_argument("--main-root", required=True, help="tree of main to compare against")
|
|
check_branch.set_defaults(func=lambda args: command_check_branch(args, cc))
|
|
|
|
decision_p = sub.add_parser(
|
|
"request-decision", help="Open the founder State Hub decision for a platform-reviewed proposed CCR"
|
|
)
|
|
decision_p.add_argument("ref")
|
|
decision_p.add_argument("--state-hub-url", default=state_hub_url)
|
|
decision_p.set_defaults(func=lambda args: command_request_decision(args, cc))
|
|
|
|
render_policy = sub.add_parser(
|
|
"render-policy", help="Print the standard-lane policy HCL for a KV path (what review-standard expects)"
|
|
)
|
|
render_policy.add_argument("--kv-path", required=True)
|
|
render_policy.add_argument("--profile", default=str(DEFAULT_PROFILE))
|
|
render_policy.set_defaults(func=lambda args: command_render_policy(args, cc))
|
|
|
|
submit_p = sub.add_parser(
|
|
"submit",
|
|
help="File a client CCR draft under the next CCR id and run the standard review",
|
|
)
|
|
submit_p.add_argument("--ccr", required=True, help="client CCR draft (non-secret YAML)")
|
|
submit_p.add_argument("--policy", help="client policy HCL; copied to openbao/policies/<policy_name>.hcl")
|
|
submit_p.add_argument("--slug", required=True)
|
|
submit_p.add_argument("--requester", required=True)
|
|
submit_p.add_argument("--profile", default=str(DEFAULT_PROFILE))
|
|
submit_p.add_argument("--commit", action="store_true", help="commit on a new ccr/<id>-<slug> branch")
|
|
submit_p.add_argument("--push", action="store_true", help="commit and push the ccr/* branch")
|
|
submit_p.add_argument("--notify", action="store_true")
|
|
submit_p.add_argument("--state-hub-url", default=state_hub_url)
|
|
submit_p.set_defaults(func=lambda args: command_submit(args, cc))
|