coulomb/ops-mason
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
ops-mason/scripts/plan-survey.py
tegwick 6b177270d3 Replace ops-mason-build with read-only ops-mason-survey grant (MASON-IN-0004) 
Founder decision 2026-10-07, option (c)+(a): ops-mason surveys and verifies
only. Policy/role writes, secret-id minting and token creation go through the
founder-attended railiance-platform delegated applier. ops-mason-build retired
to policies/retired/ (never installed).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 132360@bnt-lap001
Assistant-Session: 769b1afd-4330-4c86-96b2-e510e26db42b
2026-10-07 10:44:24 +02:00

503 lines
22 KiB
Python
Executable file
Raw Blame History

#!/usr/bin/env python3
"""plan-survey — section 2 of a construction plan, generated from live metadata.
./scripts/bao-session.sh grant <task>
./scripts/plan-survey.py --kv-path platform/workloads/<consumer>/<name> \\
--policy <proposed-policy> [--approle <role>] [--k8s-role <role>] \\
--consumer <repo> [--keyword <word> ...] [--catalog <catalog.yaml>]
Prints a Markdown block to paste as "## 2. Existing-structure survey": the
relevant lines of `bao secrets list`, `bao auth list`, `bao policy list` and the
auth-role listings, KV metadata existence for the target path, matching
ops-warden routing catalog entries, and a draft answer to "does an existing lane
already satisfy this demand?".
Metadata only. It runs under the ops-mason-survey grant (~/.claude-bao-token),
refuses any other token, and every bao invocation passes `allowed_bao_call`, an
allowlist that rejects `kv get` and every `*/data/*` path before a process starts.
Policy bodies are read with `bao policy read` (policy text, never a secret value)
and only for name-matched candidates. The token itself is never printed.
"""
from __future__ import annotations
import argparse
import datetime as _dt
import json
import os
import re
import subprocess
import sys
from dataclasses import dataclass, field
from pathlib import Path
from typing import Callable
GRANT_FILE = "~/.claude-bao-token"
GRANT_POLICY = "ops-mason-survey"
DEFAULT_ADDR = "http://127.0.0.1:18200"
DEFAULT_CATALOG = "~/ops-warden/registry/routing/catalog.yaml"
MAX_POLICY_READS = 15
NOT_PERMITTED = "not permitted by ops-mason-survey"
# Path segments too generic to identify a consumer.
GENERIC = {"platform", "workloads", "operators", "secret", "data", "metadata",
"kv", "read", "workload", "role", "token", "auth", "policy"}
ROLE_FIELDS = ("token_policies", "policies", "token_ttl", "token_max_ttl",
"token_num_uses", "secret_id_ttl", "bound_service_account_names",
"bound_service_account_namespaces", "audience")
class SurveyError(RuntimeError):
"""The survey could not be completed; never report this as an empty store."""
@dataclass
class BaoResult:
status: str # ok | denied | missing | error
data: object = None
text: str = ""
Runner = Callable[[list[str], dict], tuple[int, str, str]]
# --- metadata-only guard -----------------------------------------------------
def allowed_bao_call(args: list[str]) -> bool:
"""Allowlist of the bao invocations the survey may make. Flags are ignored."""
pos = [a for a in args if not a.startswith("-")]
if any("/data/" in a or a.startswith("data/") or a.endswith("/data") for a in pos):
return False
if any(seg in ("role-id", "secret-id", "secret-id-accessor") for a in pos for seg in a.split("/")):
return False
shape = tuple(pos[:3])
if pos in (["token", "lookup"], ["secrets", "list"], ["auth", "list"], ["policy", "list"]):
return True
if len(pos) == 3 and shape[:2] == ("policy", "read"):
return True
if len(pos) == 3 and shape[:2] == ("kv", "list"):
return True
if len(pos) == 4 and shape == ("kv", "metadata", "get"):
return True
if len(pos) == 2 and pos[0] == "list":
parts = pos[1].strip("/").split("/")
return len(parts) == 3 and parts[0] == "auth" and parts[2] == "role"
if len(pos) == 2 and pos[0] == "read":
parts = pos[1].strip("/").split("/")
return len(parts) == 4 and parts[0] == "auth" and parts[2] == "role"
return False
def subprocess_runner(argv: list[str], env: dict) -> tuple[int, str, str]:
try:
p = subprocess.run(argv, capture_output=True, text=True, timeout=30, env=env)
except (OSError, subprocess.TimeoutExpired) as exc:
raise SurveyError("bao CLI unavailable or timed out; survey was not run") from exc
return p.returncode, p.stdout, p.stderr
class Bao:
def __init__(self, runner: Runner, env: dict):
self.runner = runner
self.env = env
def call(self, *args: str, as_json: bool = True) -> BaoResult:
args_l = list(args)
if not allowed_bao_call(args_l):
raise SurveyError(f"refused non-metadata bao call: bao {' '.join(args_l[:2])}")
# The CLI rejects flags after a positional argument: -format goes right
# after the subcommand words.
if as_json:
words = 3 if args_l[:2] == ["kv", "metadata"] else (1 if args_l[0] in ("list", "read") else 2)
args_l = args_l[:words] + ["-format=json"] + args_l[words:]
rc, out, err = self.runner(["bao", *args_l], self.env)
if rc != 0:
low = (err or "").lower()
if "permission denied" in low or "403" in low:
return BaoResult("denied")
if "no value found" in low or "404" in low or (not out.strip() and rc == 2):
return BaoResult("missing")
return BaoResult("error")
if not as_json:
return BaoResult("ok", text=out)
try:
return BaoResult("ok", data=json.loads(out) if out.strip() else None)
except json.JSONDecodeError:
return BaoResult("error")
def grant_env(environ: dict | None = None) -> dict:
env = dict(os.environ if environ is None else environ)
env.setdefault("BAO_ADDR", DEFAULT_ADDR)
grant = os.path.expanduser(GRANT_FILE)
if not env.get("BAO_TOKEN") and os.path.exists(grant):
with open(grant) as f:
env["BAO_TOKEN"] = f.read().strip()
if not env.get("BAO_TOKEN"):
raise SurveyError("no ops-mason-survey grant (~/.claude-bao-token absent and BAO_TOKEN unset); "
"ask the founder to run ./scripts/bao-session.sh grant <task>")
return env
# --- policy text -------------------------------------------------------------
_STANZA = re.compile(r'path\s+"([^"]+)"\s*\{(.*?)\}', re.S)
_CAPS = re.compile(r'capabilities\s*=\s*\[([^\]]*)\]')
def policy_stanzas(hcl: str) -> list[tuple[str, list[str]]]:
out = []
for path, body in _STANZA.findall(hcl):
m = _CAPS.search(body)
caps = re.findall(r'"([^"]+)"', m.group(1)) if m else []
out.append((path, caps))
return out
def path_matches(pattern: str, path: str) -> bool:
regex = ""
for i, ch in enumerate(pattern):
if ch == "*" and i == len(pattern) - 1:
regex += ".*"
elif ch == "+":
regex += "[^/]+"
else:
regex += re.escape(ch)
return re.fullmatch(regex, path) is not None
def kv_api_paths(kv_path: str, mount: str | None) -> tuple[str, str]:
mount = (mount or kv_path.split("/")[0]).strip("/")
rest = kv_path.strip("/")[len(mount):].strip("/")
return f"{mount}/data/{rest}", f"{mount}/metadata/{rest}"
# --- catalog -----------------------------------------------------------------
def load_catalog(path: Path) -> list[dict]:
if not path.exists():
raise SurveyError(f"ops-warden catalog not found at {path}; pass --catalog")
try:
import yaml # type: ignore
except ImportError as exc: # pragma: no cover - PyYAML is a project dependency
raise SurveyError("PyYAML unavailable; cannot read the ops-warden catalog") from exc
doc = yaml.safe_load(path.read_text())
entries = doc.get("entries") if isinstance(doc, dict) else None
if not isinstance(entries, list):
raise SurveyError(f"{path} has no entries list")
return [e for e in entries if isinstance(e, dict)]
def catalog_matches(entries: list[dict], terms: list[str]) -> list[tuple[dict, list[str]]]:
out = []
for e in entries:
hay = " ".join(str(e.get(k, "")) for k in
("id", "title", "owner_repo", "subsystem", "path_template", "policy_ref",
"auth_method", "canon_ref")).lower()
hay += " " + " ".join(str(k) for k in e.get("need_keywords") or []).lower()
hits = [t for t in terms if t.lower() in hay]
if hits:
out.append((e, hits))
return out
# --- survey ------------------------------------------------------------------
@dataclass
class Demand:
kv_path: str
policy: str
consumer: str
approle: str | None = None
k8s_role: str | None = None
keywords: list[str] = field(default_factory=list)
def terms(self) -> list[str]:
segs = [s for s in self.kv_path.strip("/").split("/")[1:]]
raw = [self.consumer, self.approle, self.k8s_role, *self.keywords, *segs]
seen: list[str] = []
for t in raw:
t = (t or "").strip().lower()
if len(t) >= 3 and t not in GENERIC and t not in seen:
seen.append(t)
return seen
def _fmt_ts(now: _dt.datetime) -> str:
return now.astimezone(_dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def survey(demand: Demand, bao: Bao, catalog_path: Path, now: _dt.datetime) -> tuple[str, str]:
terms = demand.terms()
names = {n for n in (demand.policy, demand.approle, demand.k8s_role) if n}
def relevant(name: str) -> list[str]:
low = name.lower()
return [t for t in terms if t in low] + (["proposed name"] if name.strip("/") in names else [])
# token identity — display_name and policies only, never id/accessor
tok = bao.call("token", "lookup")
if tok.status != "ok" or not isinstance(tok.data, dict) or not isinstance(tok.data.get("data"), dict):
raise SurveyError("grant token lookup failed (expired, revoked or unreachable); "
"re-run ./scripts/bao-session.sh grant <task>")
td = tok.data["data"]
policies = sorted(str(p) for p in td.get("policies") or [])
if GRANT_POLICY not in policies or set(policies) - {GRANT_POLICY, "default"}:
raise SurveyError(f"token is not an {GRANT_POLICY} grant (policies: {', '.join(policies) or 'none'}); "
"refusing to survey with a broader or different session")
display = str(td.get("display_name") or "-")
lines: list[str] = []
findings: dict[str, list[str]] = {"yes": [], "partial": [], "gaps": []}
L = lines.append
L("## 2. Existing-structure survey")
L("")
L(f"Generated by `scripts/plan-survey.py` at {_fmt_ts(now)} against "
f"`BAO_ADDR={bao.env.get('BAO_ADDR')}`, token display_name `{display}` "
f"(policies: {', '.join(policies)}). Metadata only: no `kv get` and no "
"`*/data/*` read was issued; policy bodies were read with `bao policy read` "
"for name-matched candidates only.")
L("")
L(f"Demand: KV path `{demand.kv_path}`, proposed policy `{demand.policy}`, "
f"AppRole `{demand.approle or '-'}`, Kubernetes role `{demand.k8s_role or '-'}`, "
f"consumer `{demand.consumer}`. Match terms: {', '.join(f'`{t}`' for t in terms) or '-'}.")
L("")
def block(title: str, body: list[str]) -> None:
L(f"### {title}")
L("")
L("```text")
lines.extend(body or ["(no relevant entries)"])
L("```")
L("")
# secrets list
mounts = bao.call("secrets", "list")
target_mount = None
if mounts.status == "ok" and isinstance(mounts.data, dict):
rows = []
for path, m in sorted(mounts.data.items()):
m = m if isinstance(m, dict) else {}
if demand.kv_path.strip("/").startswith(path.strip("/") + "/") and (
target_mount is None or len(path) > len(target_mount)):
target_mount = path.strip("/")
for path, m in sorted(mounts.data.items()):
m = m if isinstance(m, dict) else {}
hit = path.strip("/") == target_mount or relevant(path)
if hit:
ver = (m.get("options") or {}).get("version", "") if isinstance(m.get("options"), dict) else ""
rows.append(f"{path:<24} {m.get('type', '?'):<10} "
f"{'version=' + str(ver) if ver else '':<10} {m.get('description', '')}".rstrip())
block(f"`bao secrets list` ({len(mounts.data)} mounts, {len(rows)} relevant)", rows)
if target_mount is None:
findings["gaps"].append(f"no secrets engine is mounted at a prefix of `{demand.kv_path}`; "
"enabling one is a railiance-platform act")
else:
block("`bao secrets list`", [_status_line(mounts)])
findings["gaps"].append(f"`bao secrets list`: {_status_line(mounts)}")
# auth list
auth = bao.call("auth", "list")
role_mounts: list[tuple[str, str]] = []
if auth.status == "ok" and isinstance(auth.data, dict):
rows = []
for path, m in sorted(auth.data.items()):
m = m if isinstance(m, dict) else {}
typ = str(m.get("type", "?"))
if typ in ("approle", "kubernetes"):
role_mounts.append((path.strip("/"), typ))
if typ in ("approle", "kubernetes") or relevant(path):
rows.append(f"{path:<24} {typ:<12} {m.get('description', '')}".rstrip())
block(f"`bao auth list` ({len(auth.data)} auth mounts, {len(rows)} relevant)", rows)
else:
block("`bao auth list`", [_status_line(auth)])
findings["gaps"].append(f"`bao auth list`: {_status_line(auth)}")
role_mounts = [("approle", "approle")] + ([("kubernetes", "kubernetes")] if demand.k8s_role else [])
# policy list
plist = bao.call("policy", "list")
candidates: list[str] = []
if plist.status == "ok" and isinstance(plist.data, list):
rows = []
for name in sorted(str(p) for p in plist.data):
why = relevant(name)
if why:
candidates.append(name)
rows.append(f"{name:<56} matches: {', '.join(why)}")
block(f"`bao policy list` ({len(plist.data)} policies, {len(rows)} relevant)", rows)
if demand.policy in plist.data:
findings["partial"].append(f"proposed policy name `{demand.policy}` already exists")
else:
block("`bao policy list`", [_status_line(plist)])
findings["gaps"].append(f"`bao policy list`: {_status_line(plist)}")
# auth roles
bound: dict[str, list[str]] = {}
for mount, typ in role_mounts:
res = bao.call("list", f"auth/{mount}/role")
title = f"`bao list auth/{mount}/role` ({typ})"
if res.status == "missing":
block(f"{title} (0 roles)", [])
continue
if res.status != "ok" or not isinstance(res.data, list):
block(title, [_status_line(res)])
findings["gaps"].append(f"`bao list auth/{mount}/role`: {_status_line(res)}")
continue
rows = []
for role in sorted(str(r) for r in res.data):
why = relevant(role)
if not why:
continue
rows.append(f"{role:<48} matches: {', '.join(why)}")
if role in names:
findings["partial"].append(f"proposed role name `{role}` already exists on auth/{mount}")
detail = bao.call("read", f"auth/{mount}/role/{role}")
if detail.status == "ok" and isinstance(detail.data, dict) and isinstance(detail.data.get("data"), dict):
d = detail.data["data"]
for k in ROLE_FIELDS:
if d.get(k) not in (None, "", []):
rows.append(f" {k}={d[k]}")
pols = d.get("token_policies") or d.get("policies") or []
for p in pols if isinstance(pols, list) else []:
bound.setdefault(str(p), []).append(f"auth/{mount}/role/{role}")
else:
rows.append(f" (role read: {_status_line(detail)})")
block(f"{title} ({len(res.data)} roles, {sum(1 for r in rows if not r.startswith(' '))} relevant)", rows)
# policy bodies for name-matched candidates
data_path, meta_path = kv_api_paths(demand.kv_path, target_mount)
granting: list[str] = []
rows = []
for name in candidates[:MAX_POLICY_READS]:
res = bao.call("policy", "read", name, as_json=False)
if res.status != "ok":
rows.append(f"{name}: {_status_line(res)}")
if res.status == "denied":
findings["gaps"].append(f"`bao policy read {name}`: {NOT_PERMITTED}")
continue
hits = [(p, c) for p, c in policy_stanzas(res.text)
if path_matches(p, data_path) or path_matches(p, meta_path)]
if hits:
for p, c in hits:
rows.append(f'{name}: path "{p}" capabilities={c}')
if any(path_matches(p, data_path) and "read" in c and "deny" not in c for p, c in hits):
granting.append(name)
else:
rows.append(f"{name}: no stanza covers {data_path}")
if len(candidates) > MAX_POLICY_READS:
rows.append(f"({len(candidates) - MAX_POLICY_READS} further candidates not read; narrow --keyword)")
block(f"`bao policy read` of name-matched candidates (target `{data_path}`)", rows)
# KV metadata for the target and its siblings
rows = []
meta = bao.call("kv", "metadata", "get", demand.kv_path)
kv_exists = False
if meta.status == "ok" and isinstance(meta.data, dict) and isinstance(meta.data.get("data"), dict):
kv_exists = True
d = meta.data["data"]
rows.append(f"{demand.kv_path}: exists, current_version={d.get('current_version', '?')}, "
f"created={str(d.get('created_time', ''))[:10]}, updated={str(d.get('updated_time', ''))[:10]}")
cm = d.get("custom_metadata") or {}
for k in ("description", "owner", "used_by"):
if isinstance(cm, dict) and cm.get(k):
rows.append(f" {k}: {cm[k]}")
findings["partial"].append(f"KV path `{demand.kv_path}` already exists (v{d.get('current_version', '?')})")
elif meta.status == "missing":
rows.append(f"{demand.kv_path}: no metadata (path unused)")
else:
rows.append(f"{demand.kv_path}: {_status_line(meta)}")
findings["gaps"].append(f"KV metadata for `{demand.kv_path}`: {_status_line(meta)}")
parent = demand.kv_path.strip("/").rsplit("/", 1)[0]
sib = bao.call("kv", "list", parent)
if sib.status == "ok" and isinstance(sib.data, list):
rows.append(f"{parent}/: {', '.join(str(s) for s in sib.data) or '(empty)'}")
else:
rows.append(f"{parent}/: {_status_line(sib)}")
block("`bao kv metadata get` / `bao kv list` (metadata only)", rows)
# ops-warden catalog
try:
entries = load_catalog(catalog_path)
cat_terms = terms + [t for t in (demand.kv_path, demand.policy) if t not in terms]
matches = catalog_matches(entries, cat_terms)
rows = []
for e, hits in matches:
rows.append(f"{e.get('id')} owner={e.get('owner_repo')} status={e.get('status')} "
f"matches: {', '.join(hits)}")
for k in ("path_template", "policy_ref", "auth_method"):
if e.get(k):
rows.append(f" {k}: {e[k]}")
if demand.kv_path in str(e.get("path_template", "")) or demand.policy == e.get("policy_ref"):
findings["partial"].append(f"ops-warden catalog entry `{e.get('id')}` already routes this path/policy")
block(f"ops-warden routing catalog `{catalog_path}` ({len(entries)} entries, {len(matches)} relevant)", rows)
except SurveyError as exc:
block("ops-warden routing catalog", [str(exc)])
findings["gaps"].append(str(exc))
# verdict
satisfied = [p for p in granting if p in bound]
if satisfied:
verdict = "yes"
for p in satisfied:
findings["yes"].append(f"policy `{p}` grants read on `{data_path}` and is bound by "
f"{', '.join(bound[p])}: bind to the existing lane instead of creating one")
elif granting or findings["partial"]:
verdict = "partial"
for p in granting:
findings["partial"].insert(0, f"policy `{p}` already grants read on `{data_path}` "
"but no surveyed role binds it: reuse the policy, add only a role")
else:
verdict = "no"
L("### Does an existing lane already satisfy this demand?")
L("")
L(f"**Draft answer: {verdict}.** Review before committing; this is generated, not a decision.")
L("")
for r in findings["yes"] + findings["partial"]:
L(f"- {r}")
if verdict == "no":
L(f"- no surveyed policy grants `{data_path}`, the proposed names are free, the KV path is unused "
"and no catalog entry routes it")
L(f"- policy grants were checked only for name-matched candidates; a policy with an unrelated name "
f"that covers `{data_path}` is not detected")
for g in findings["gaps"]:
L(f"- incomplete: {g}")
L("")
return "\n".join(lines), verdict
def _status_line(res: BaoResult) -> str:
return {"denied": NOT_PERMITTED, "missing": "not found",
"error": "request failed (output suppressed)"}.get(res.status, res.status)
def parse_args(argv: list[str]) -> argparse.Namespace:
p = argparse.ArgumentParser(prog="plan-survey", description=__doc__.split("\n\n")[0])
p.add_argument("--kv-path", required=True, help="target KV path, e.g. platform/workloads/x/y")
p.add_argument("--policy", required=True, help="proposed policy name")
p.add_argument("--approle", help="proposed AppRole name")
p.add_argument("--k8s-role", help="proposed Kubernetes-auth role name")
p.add_argument("--consumer", required=True, help="consumer repo")
p.add_argument("--keyword", action="append", default=[], help="extra match term (repeatable)")
p.add_argument("--catalog", default=DEFAULT_CATALOG, help="ops-warden routing catalog path")
return p.parse_args(argv)
def main(argv: list[str] | None = None, runner: Runner = subprocess_runner,
environ: dict | None = None, now: _dt.datetime | None = None) -> int:
a = parse_args(sys.argv[1:] if argv is None else argv)
if "/data/" in a.kv_path or a.kv_path.startswith("/"):
raise SurveyError("--kv-path is the logical KV path (mount/path), not an API data path")
demand = Demand(a.kv_path.strip("/"), a.policy, a.consumer, a.approle, a.k8s_role, a.keyword)
bao = Bao(runner, grant_env(environ))
text, _ = survey(demand, bao, Path(os.path.expanduser(a.catalog)),
now or _dt.datetime.now(_dt.timezone.utc))
print(text)
return 0
if __name__ == "__main__":
try:
raise SystemExit(main())
except SurveyError as exc:
print(f"ERROR: {exc}", file=sys.stderr)
raise SystemExit(2)
Reference in a new issue View git blame Copy permalink