coulomb/ops-mason
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

MASON-WP-0010-T03: ops-mason build --case; Forgejo case bundle built

Browse source 
Builds from an accepted case and its reviewed, undrifted plan into CCR drafts
for railiance-platform's applier, policy HCL, catalog proposal, exact rollback
and manifest; the local audit log records case, type and acceptance. Drafts
pass the platform's own CCR validator. 6 tests; suite 175 passed.

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:
tegwick 2026-10-09 19:52:33 +02:00
parent a97d1931dd
commit 3e49ed4c0c
11 changed files with 633 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

213
src/ops_mason/case_build.py Normal file
View file

@ -0,0 +1,213 @@
"""Build from an accepted SecurityCase (MASON-WP-0010-T03).
`ops-mason build --case` turns a reviewed, case-derived construction plan into
the inputs the platform's delegated applier executes. ops-mason never writes to
OpenBao: the founder applies through railiance-platform (`credential-change.py
applier-apply`), as for every other lane. What changes for a KnownSecurityCase
is the gate: the case acceptance replaces a fresh executive summary, so the
build needs no separate approval in ops-mason, only
- an accepted case (validated by net-kingdom's validator),
- a plan rendered from it with status `reviewed` (live survey, no collision),
- and the plan's render digest equal to a fresh render, so nothing drifted.
The bundle names the case, the type version and the acceptance in every file,
and its rollback lists exactly what this case creates, in reverse order.
"""
from __future__ import annotations
import hashlib
import json
import os
import subprocess
from datetime import date
from pathlib import Path
import yaml
from ops_mason import case_plan
from ops_mason.audit import record_build
from ops_mason.plan import ConstructionPlan, PlanError
RAILIANCE_PLATFORM = Path.home() / "railiance-platform"
class CaseBuildRefused(RuntimeError):
"""The case cannot be built yet; nothing was written."""
def _plan_front(path: Path) -> dict:
text = Path(path).read_text()
if not text.startswith("---\n"):
raise CaseBuildRefused(f"{path} has no frontmatter")
return yaml.safe_load(text.split("---\n", 2)[1]) or {}
def _items(rendered: dict, kind: str) -> list[dict]:
return [i for i in rendered["materialization"] if i["kind"] == kind]
def _provenance(case: dict, rendered: dict) -> dict:
acc = case_plan.acceptance_ref(case)
return {"case": case["spec"]["id"], "type": rendered["type"], "render_digest": rendered["digest"],
"acceptance_record": acc["record"], "accepted_at": acc["at"], "accepted_by": acc["by"]}
def _ccr(case: dict, rendered: dict, policy: dict, role: dict | None, kv_path: str, prov: dict) -> dict:
"""A workload-kv-read CCR draft; railiance-platform assigns the id when it files it."""
params = case["spec"]["parameters"]
role_body = yaml.safe_load(role["body"]) if role else None
today = date.today().isoformat()
openbao = {
"mount": kv_path.split("/", 1)[0],
"kv_path": kv_path,
"fields": ["CLIENT_SECRET"],
"policy_name": policy["template"],
"policy_file": f"openbao/policies/{policy['template']}.hcl",
"metadata_read": False,
"token_self_lifecycle": True,
}
if role_body:
openbao["auth"] = {
"method": role_body.get("auth", "kubernetes"),
"mount": role_body.get("auth", "kubernetes"),
"role": role["template"],
"audience": "openbao",
"ttl": role_body.get("token_ttl", "15m"),
"token_max_ttl": role_body.get("token_max_ttl", "15m"),
"token_explicit_max_ttl": role_body.get("token_max_ttl", "15m"),
"token_no_default_policy": bool(role_body.get("token_no_default_policy", True)),
"policies": list(role_body.get("token_policies") or [policy["template"]]),
"bound_claims": {"service_account_names": list(role_body.get("bound_service_account_names") or []),
"service_account_namespaces": list(role_body.get("bound_service_account_namespaces") or [])},
"bound_claims_confirmed": False,
}
reader = "eso" if role else params["consuming_service"]
doc = {
"id": f"CCR-DRAFT-{prov['case']}-{reader}".upper(),
"kind": "credential-change-request",
"schema_version": 1,
"request_type": "workload-kv-read",
"title": f"{prov['case']}: {policy['id']} ({prov['type']})",
"status": "proposed" if role else "in_flight",
"created": today,
"updated": today,
"requester": {"agent": "ops-mason", "reason": (
f"MASON-WP-0010-T03 build of SecurityCase {prov['case']} ({prov['type']}), render "
f"{prov['render_digest']}. Accepted by {', '.join(prov['accepted_by'])} on {prov['accepted_at']}: "
f"{prov['acceptance_record']}. The case acceptance replaces a fresh executive summary.")},
"review": {"required": True, "required_approvers": ["platform-owner"], "comments": []},
"target": {"domain": "infotech", "tenant": "platform", "workload": "keycape", "environment": "production",
"purpose": policy.get("description") or policy["id"]},
"openbao": openbao,
"access_frontdoor": {"type": "external-secrets" if role else "none",
"catalog_id": f"{params['consuming_service']}-keycape-client",
"readiness": "pending-review", "resolvable": False},
"risk": {"classification": "high", "notes": [
"Rendered from an acknowledged SecurityCaseType; shape reviewed with the type, names checked against live structure.",
"Exact read on one KV path; no metadata, list, parent or sibling access."]},
"verification": {
"positive": [f"policy {policy['template']} grants read on {kv_path.replace('/', '/data/', 1)} only (capabilities check)"],
"negative": ["a default-only identity has no read capability on the path", "sibling GET and parent LIST are denied"],
"activation_conditions": ["case acceptance present", "bound claims confirmed by the platform owner"]},
"lifecycle": {"deactivate": f"delete the role, then policy {policy['template']} (case rollback)",
"rotate": "roll-forward through the credential broker (mint-secret); never a read of the old value",
"compromised": "owner incident process; no automatic read, reseed or cleanup"},
"security_case": prov,
}
if not role:
doc["in_flight"] = {"missing_fields": ["openbao.auth"], "owner": params["consumer_owner_repo"],
"blocking_reason": (f"{params['consumer_owner_repo']} binds this reader to "
f"{params['consuming_service']}'s own identity in its deployment path (hand-over).")}
return doc
def _policy_hcl(policy: dict, prov: dict) -> str:
head = (f"# {policy['template']}: rendered by ops-mason from SecurityCase {prov['case']} "
f"({prov['type']}, render {prov['render_digest']}).\n# Do not edit: re-run ops-mason build --case.\n\n")
return head + policy["body"].rstrip() + "\n"
def validate_with_platform(ccr_paths: list[Path], platform: Path = RAILIANCE_PLATFORM) -> list[str]:
"""Run railiance-platform's own CCR validator, read-only; return its FAIL lines.
A policy file that exists in the bundle but not yet in railiance-platform is
expected (the platform files it with the CCR) and is not counted.
"""
script = platform / "scripts" / "credential-change.py"
if not script.is_file():
return [f"platform validator not found at {script}"]
# Run in the platform's environment, not ops-mason's virtualenv.
venv = os.environ.get("VIRTUAL_ENV")
env = {k: v for k, v in os.environ.items() if k not in ("VIRTUAL_ENV", "UV_PROJECT_ENVIRONMENT", "PYTHONPATH")}
if venv:
env["PATH"] = os.pathsep.join(d for d in env.get("PATH", "").split(os.pathsep) if not d.startswith(venv))
ccr_paths = [Path(p).resolve() for p in ccr_paths]
r = subprocess.run(["python3", str(script), "validate", *map(str, ccr_paths)],
capture_output=True, text=True, cwd=platform, env=env)
bundle_policies = {f"openbao/policies/{q.name}" for p in ccr_paths
for q in (p.parent.parent / "openbao" / "policies").glob("*.hcl")}
fails = [l for l in (r.stdout + r.stderr).splitlines() if l.startswith("[FAIL]")]
fails = [l for l in fails if not any(l.endswith(f"policy_file does not exist: {b}") for b in bundle_policies)]
if r.returncode != 0 and not fails and "[FAIL]" not in r.stdout + r.stderr:
fails = [f"validator did not run: {(r.stderr or r.stdout).strip().splitlines()[-1:]}"]
return fails
def build_from_case(case_path: Path, plan_path: Path, out_dir: Path, *, nk_root: Path = case_plan.DEFAULT_NK_ROOT,
allow_unimplemented: bool = False, audit_log: Path | None = None,
types_root: Path | None = None) -> dict:
case, _type, rendered = case_plan.render(case_path, nk_root, types_root, allow_unimplemented)
acc = case_plan.acceptance_ref(case)
if case["spec"].get("status") != "accepted" or not acc:
raise CaseBuildRefused("case is not accepted: a build needs the case acceptance record")
front = _plan_front(plan_path)
if front.get("status") != "reviewed":
raise CaseBuildRefused(f"plan {plan_path} is {front.get('status')!r}, not 'reviewed' (re-run plan --case live)")
if front.get("render_digest") != rendered["digest"]:
raise CaseBuildRefused("plan render digest differs from a fresh render: the case or type changed since review")
prov = _provenance(case, rendered)
kv = _items(rendered, "kv_path")
if len(kv) != 1:
raise CaseBuildRefused(f"expected exactly one kv_path item, got {len(kv)}")
kv_path = kv[0]["template"]
roles = {yaml.safe_load(r["body"]).get("token_policies", [None])[0]: r for r in _items(rendered, "openbao_role")}
out_dir.mkdir(parents=True, exist_ok=False)
files: dict[str, str] = {}
def write(rel: str, text: str) -> Path:
path = out_dir / rel
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(text)
files[rel] = hashlib.sha256(text.encode()).hexdigest()
return path
ccrs = []
for policy in _items(rendered, "openbao_policy"):
write(f"openbao/policies/{policy['template']}.hcl", _policy_hcl(policy, prov))
doc = _ccr(case, rendered, policy, roles.get(policy["template"]), kv_path, prov)
ccrs.append(write(f"credential-change-requests/{doc['id']}.yaml", yaml.safe_dump(doc, sort_keys=False, width=110)))
for cat in _items(rendered, "catalog_entry"):
write("catalog-proposal.yaml", yaml.safe_dump({"proposal_for": "ops-warden", "security_case": prov,
"entry": cat["body"]}, sort_keys=False, width=110))
created = [{"kind": i["kind"], "name": i["template"]} for i in rendered["materialization"]]
write("rollback.yaml", yaml.safe_dump({
"security_case": prov,
"removes_exactly": list(reversed(created)),
"steps": [r["action"] for r in rendered["rollback"]],
"note": "Undo removes only what this case created; shared groups are revoked only if no other case uses them.",
}, sort_keys=False, width=110))
manifest = {"security_case": prov, "kv_path": kv_path, "files": files,
"applier": "railiance-platform credential-change.py applier-apply (founder), after it files the CCR drafts",
"not_ops_mason": ["mint-secret (credential broker lane keycape-client-secret-generate)",
"catalog activation (ops-warden)", "client registration (net-kingdom, attended)"]}
write("manifest.json", json.dumps(manifest, indent=1, sort_keys=True) + "\n")
record_build(plan_id=front.get("id", f"case-{prov['case']}"), plan_path=str(plan_path),
approved_by="case acceptance: " + ", ".join(acc["by"]), approved_at=str(acc["at"]),
objects={"event": "case-build-prepared", **prov, "bundle": str(out_dir),
"objects": created}, log_path=audit_log)
return {"bundle": str(out_dir), "ccrs": [str(p) for p in ccrs], **prov}

26
src/ops_mason/cli.py
View file

@ -48,9 +48,33 @@ def _parser() -> argparse.ArgumentParser:
plan.add_argument("--netkingdom-root", type=Path, default=None)
plan.add_argument("--allow-unimplemented", action="store_true",
help="reference case only: render against an acknowledged type")
build = families.add_parser("build", help="build bundle from an accepted SecurityCase and its reviewed plan")
build.add_argument("--case", required=True, type=Path)
build.add_argument("--plan", required=True, type=Path, help="the reviewed plan from plan --case")
build.add_argument("--out", type=Path, help="bundle directory (default: builds/case-<id>)")
build.add_argument("--allow-unimplemented", action="store_true",
help="reference case only: build against an acknowledged type")
build.add_argument("--netkingdom-root", type=Path, default=None)
return parser
def _build(args) -> int:
from ops_mason import case_build, case_plan
nk_root = args.netkingdom_root or case_plan.DEFAULT_NK_ROOT
try:
case_id = (case_plan.yaml.safe_load(args.case.read_text()) or {}).get("spec", {}).get("id", "unknown")
out = args.out or Path("builds") / f"case-{case_id}"
result = case_build.build_from_case(args.case, args.plan, out, nk_root=nk_root,
allow_unimplemented=args.allow_unimplemented)
except (case_plan.CasePlanError, case_build.CaseBuildRefused, FileExistsError, OSError) as exc:
print(json.dumps({"ok": False, "error": str(exc), "issues": getattr(exc, "issues", [])}), file=sys.stderr)
return 2
fails = case_build.validate_with_platform([Path(p) for p in result["ccrs"]])
print(json.dumps({"ok": not fails, "platform_validation": fails or "pass", **result}, indent=2))
return 1 if fails else 0
def _plan(args) -> int:
from ops_mason import case_plan
@ -74,6 +98,8 @@ def main(argv: Sequence[str] | None = None) -> int:
args = _parser().parse_args(argv)
if args.family == "plan":
return _plan(args)
if args.family == "build":
return _build(args)
try:
bundle = PlaneBundle.load(args.bundle)
if args.command == "render":