diff --git a/AGENTS.md b/AGENTS.md index e7031b3..6d35f6a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -250,6 +250,11 @@ and every action lands in the audit log as you. ./scripts/bao-session.sh revoke # end it immediately ``` +With a grant in place, generate a construction plan's section 2 with +`./scripts/plan-survey.py --kv-path --policy --approle +--consumer [--keyword ]` (metadata only; see +`docs/construction-plan-format.md` § 2). + `grant` mints a **separate** token into `~/.claude-bao-token` — your own `~/.vault-token` is untouched, and the two revoke independently. The agent uses it as `BAO_TOKEN=$(cat ~/.claude-bao-token) bao `. diff --git a/docs/construction-plan-format.md b/docs/construction-plan-format.md index b1077f3..1fe1dc5 100644 --- a/docs/construction-plan-format.md +++ b/docs/construction-plan-format.md @@ -60,6 +60,37 @@ anything new: minting a new one, because the scoping already covered the new consumer.) +Generate the section rather than gathering it by hand (MASON-WP-0008). The +founder mints the scoped grant; the agent then runs the survey with it: + +```bash +./scripts/bao-session.sh grant # founder: ops-mason-build, 45m, claude- +./scripts/plan-survey.py \ + --kv-path platform/workloads// \ + --policy workload-kv-read-- \ + --approle [--k8s-role ] \ + --consumer [--keyword ...] \ + [--catalog ~/ops-warden/registry/routing/catalog.yaml] +``` + +It prints a Markdown block headed `## 2. Existing-structure survey`: the +relevant lines and counts of `bao secrets list`, `bao auth list`, +`bao policy list` and `bao list auth//role` (approle and kubernetes +mounts), `bao policy read` stanzas covering the target for name-matched +policies only, KV metadata existence for the target and its siblings, +matching ops-warden catalog entries, a UTC timestamp, `BAO_ADDR`, the token +`display_name`, and a draft yes/no/partial answer with reasons. Edit the +draft answer into a decision; it is generated evidence, not a review. + +The survey refuses to run without a grant, refuses a token whose policies +are anything but `ops-mason-build` (plus `default`), and never prints the +token. Every bao call passes an allowlist that rejects `kv get` and any +`*/data/*` path before a process starts. A query the grant does not permit +is reported as `not permitted by ops-mason-build` and listed under +"incomplete" rather than read as an empty answer. Policies with names that +do not match the demand are not read, so a broadly named policy covering +the path is not detected; the output says so. + ### 3. Proposed changes A list, each item tagged `create` / `modify` / `tear-down`, each with a diff --git a/scripts/plan-survey.py b/scripts/plan-survey.py new file mode 100755 index 0000000..5b38843 --- /dev/null +++ b/scripts/plan-survey.py @@ -0,0 +1,503 @@ +#!/usr/bin/env python3 +"""plan-survey — section 2 of a construction plan, generated from live metadata. + + ./scripts/bao-session.sh grant + ./scripts/plan-survey.py --kv-path platform/workloads// \\ + --policy [--approle ] [--k8s-role ] \\ + --consumer [--keyword ...] [--catalog ] + +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-build 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-build" +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-build" +# 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-build grant (~/.claude-bao-token absent and BAO_TOKEN unset); " + "ask the founder to run ./scripts/bao-session.sh grant ") + 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 ") + 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) diff --git a/tests/test_plan_survey.py b/tests/test_plan_survey.py new file mode 100644 index 0000000..f1d168f --- /dev/null +++ b/tests/test_plan_survey.py @@ -0,0 +1,216 @@ +"""plan-survey reads metadata only, never prints a token, and fails loudly.""" +import datetime as dt +import importlib.util +import json +import sys +from pathlib import Path + +import pytest + +spec = importlib.util.spec_from_file_location("plan_survey", Path(__file__).parents[1] / "scripts/plan-survey.py") +survey = importlib.util.module_from_spec(spec) +sys.modules["plan_survey"] = survey +spec.loader.exec_module(survey) + +TOKEN = "hvs.SYNTHETICgrantTOKENvalue0123456789" +NOW = dt.datetime(2026, 10, 5, 12, 0, tzinfo=dt.timezone.utc) +ARGS = ["--kv-path", "platform/workloads/inbound-library/imap", + "--policy", "workload-kv-read-inbound-library-imap", + "--approle", "inbound-library", "--consumer", "inbound-library", "--keyword", "imap"] + +CATALOG = """ +version: 1 +entries: + - id: inbound-library-imap + title: inbound-library IMAP mailbox + need_keywords: [inbound-library, imap] + owner_repo: ops-mason + path_template: platform/workloads/inbound-library/imap + status: draft + - id: npm-publish-lane + title: something else + need_keywords: [npm] + owner_repo: railiance-platform + status: active +""" + +POLICY_GRANTING = '''path "platform/data/workloads/inbound-library/imap" { + capabilities = ["read"] +} +path "platform/metadata/workloads/inbound-library/imap" { + capabilities = ["read"] +} +''' + + +def lookup(policies=("default", "ops-mason-build")): + return {"data": {"id": TOKEN, "accessor": "acc-SECRET", "display_name": "token-claude-mason-wp-0008", + "policies": list(policies), "ttl": 2000}} + + +class FakeBao: + """Records every argv; answers from a table keyed by positional words.""" + + def __init__(self, table=None, denied=(), missing=()): + self.calls = [] + self.table = { + ("token", "lookup"): lookup(), + ("secrets", "list"): {"platform/": {"type": "kv", "options": {"version": "2"}, "description": "workloads"}, + "sys/": {"type": "system"}}, + ("auth", "list"): {"approle/": {"type": "approle"}, "kubernetes/": {"type": "kubernetes"}, + "token/": {"type": "token"}}, + ("policy", "list"): ["default", "root", "ops-mason-build", "workload-kv-read-inbound-library-imap", + "workload-kv-read-other"], + ("list", "auth/approle/role"): ["inbound-library", "rein-openweights"], + ("list", "auth/kubernetes/role"): ["external-secrets-activity-core"], + ("read", "auth/approle/role/inbound-library"): {"data": { + "token_policies": ["workload-kv-read-inbound-library-imap"], "token_ttl": 900}}, + ("policy", "read", "workload-kv-read-inbound-library-imap"): POLICY_GRANTING, + ("kv", "metadata", "get", "platform/workloads/inbound-library/imap"): {"data": { + "current_version": 1, "created_time": "2026-10-01T00:00:00Z", + "custom_metadata": {"description": "IMAP login for inbound-library"}}}, + ("kv", "list", "platform/workloads/inbound-library"): ["imap"], + } + self.table.update(table or {}) + self.denied, self.missing = set(denied), set(missing) + + def __call__(self, argv, env): + assert argv[0] == "bao" + assert env["BAO_TOKEN"] == TOKEN + self.calls.append(argv) + key = tuple(a for a in argv[1:] if not a.startswith("-")) + if key in self.denied: + return 2, "", "Error: permission denied" + if key in self.missing or key not in self.table: + return 2, "", "No value found" + value = self.table[key] + return 0, value if isinstance(value, str) else json.dumps(value), "" + + +@pytest.fixture +def catalog(tmp_path): + p = tmp_path / "catalog.yaml" + p.write_text(CATALOG) + return p + + +def run(fake, catalog, capsys, args=ARGS): + rc = survey.main([*args, "--catalog", str(catalog)], runner=fake, + environ={"BAO_TOKEN": TOKEN, "BAO_ADDR": "http://127.0.0.1:18200"}, now=NOW) + return rc, capsys.readouterr().out + + +def assert_metadata_only(calls): + for argv in calls: + pos = [a for a in argv[1:] if not a.startswith("-")] + assert not (pos[:1] == ["kv"] and "get" in pos and "metadata" not in pos), argv + assert pos[:2] != ["kv", "get"], argv + assert not any("/data/" in a or a.startswith("data/") for a in pos), argv + assert not any(seg in ("role-id", "secret-id") for a in pos for seg in a.split("/")), argv + + +def test_existing_lane_found_and_only_metadata_read(catalog, capsys): + fake = FakeBao() + rc, out = run(fake, catalog, capsys) + assert rc == 0 + assert_metadata_only(fake.calls) + assert "## 2. Existing-structure survey" in out + assert "**Draft answer: yes.**" in out + assert "token-claude-mason-wp-0008" in out and "BAO_ADDR=http://127.0.0.1:18200" in out + assert "2026-10-05T12:00:00Z" in out + assert "(5 policies, 1 relevant)" in out + assert "inbound-library-imap owner=ops-mason" in out and "npm-publish-lane" not in out + assert "rein-openweights" not in out + assert TOKEN not in out and "acc-SECRET" not in out + + +def test_no_existing_lane(catalog, capsys, tmp_path): + empty = tmp_path / "empty.yaml" + empty.write_text("version: 1\nentries: []\n") + fake = FakeBao(table={("policy", "list"): ["default"], ("list", "auth/approle/role"): ["other"]}, + missing={("kv", "metadata", "get", "platform/workloads/inbound-library/imap"), + ("kv", "list", "platform/workloads/inbound-library")}) + rc, out = run(fake, empty, capsys) + assert rc == 0 + assert "**Draft answer: no.**" in out + assert "no metadata (path unused)" in out + assert_metadata_only(fake.calls) + + +def test_policy_without_bound_role_is_partial(catalog, capsys): + fake = FakeBao(table={("list", "auth/approle/role"): ["other"]}) + rc, out = run(fake, catalog, capsys) + assert "**Draft answer: partial.**" in out + assert "no surveyed role binds it" in out + + +def test_missing_capability_degrades_and_is_reported(catalog, capsys): + fake = FakeBao(denied={("list", "auth/approle/role"), ("list", "auth/kubernetes/role"), + ("policy", "read", "workload-kv-read-inbound-library-imap")}) + rc, out = run(fake, catalog, capsys) + assert rc == 0 + assert "not permitted by ops-mason-build" in out + assert "incomplete:" in out + + +def test_no_grant_fails_loudly(monkeypatch, tmp_path): + monkeypatch.setattr(survey.os.path, "expanduser", lambda p: str(tmp_path / "absent")) + with pytest.raises(survey.SurveyError, match="no ops-mason-build grant"): + survey.main(ARGS, runner=FakeBao(), environ={}) + + +@pytest.mark.parametrize("policies", [("root",), ("default", "ops-mason-build", "platform-admin"), ("default",)]) +def test_non_grant_token_is_refused(catalog, policies): + fake = FakeBao(table={("token", "lookup"): lookup(policies)}) + with pytest.raises(survey.SurveyError, match="not an ops-mason-build grant") as exc: + survey.main([*ARGS, "--catalog", str(catalog)], runner=fake, environ={"BAO_TOKEN": TOKEN}) + assert TOKEN not in str(exc.value) + assert len(fake.calls) == 1 + + +def test_expired_grant_fails_loudly(catalog): + fake = FakeBao(denied={("token", "lookup")}) + with pytest.raises(survey.SurveyError, match="lookup failed"): + survey.main([*ARGS, "--catalog", str(catalog)], runner=fake, environ={"BAO_TOKEN": TOKEN}) + + +def test_script_exits_nonzero_without_grant(tmp_path): + import os + import subprocess + env = {k: v for k, v in os.environ.items() if k not in ("BAO_TOKEN", "VAULT_TOKEN")} + env["HOME"] = str(tmp_path) + p = subprocess.run([sys.executable, str(Path(__file__).parents[1] / "scripts/plan-survey.py"), *ARGS], + capture_output=True, text=True, env=env) + assert p.returncode == 2 + assert "ERROR: no ops-mason-build grant" in p.stderr + + +@pytest.mark.parametrize("args", [ + ["kv", "get", "platform/workloads/x"], + ["kv", "get", "-field=password", "platform/workloads/x"], + ["read", "platform/data/workloads/x"], + ["list", "platform/data/workloads"], + ["kv", "metadata", "get", "platform/data/x"], + ["read", "auth/approle/role/x/role-id"], + ["write", "auth/approle/role/x/secret-id"], + ["read", "sys/raw/x"], + ["policy", "write", "x", "-"], + ["token", "create"], +]) +def test_guard_refuses_value_reads_and_writes(args): + assert not survey.allowed_bao_call(args) + bao = survey.Bao(FakeBao(), {"BAO_TOKEN": TOKEN}) + with pytest.raises(survey.SurveyError, match="refused"): + bao.call(*args) + + +def test_kv_path_must_not_be_an_api_data_path(catalog): + with pytest.raises(survey.SurveyError, match="logical KV path"): + survey.main(["--kv-path", "platform/data/workloads/x", "--policy", "p", "--consumer", "c"], + runner=FakeBao(), environ={"BAO_TOKEN": TOKEN}) + + +def test_policy_glob_matching(): + assert survey.path_matches("platform/data/workloads/*", "platform/data/workloads/a/b") + assert survey.path_matches("platform/data/+/a/b", "platform/data/workloads/a/b") + assert not survey.path_matches("platform/data/workloads/a", "platform/data/workloads/a/b") diff --git a/workplans/MASON-WP-0008-plan-survey.md b/workplans/MASON-WP-0008-plan-survey.md new file mode 100644 index 0000000..531c1a0 --- /dev/null +++ b/workplans/MASON-WP-0008-plan-survey.md @@ -0,0 +1,112 @@ +--- +id: MASON-WP-0008 +type: workplan +title: "Generate the construction-plan existing-structure survey from live metadata" +domain: infotech +repo: ops-mason +status: active +flavor: implementation +owner: claude +topic_slug: netkingdom +created: "2026-10-05" +updated: "2026-10-05" +--- + +# MASON-WP-0008 — Generate the existing-structure survey + +## Goal + +Section 2 of a construction plan (`docs/construction-plan-format.md`) needs +actual OpenBao output: relevant policy, auth and secrets-engine entries, the +matching ops-warden routing catalog entries, and an explicit answer to "does an +existing lane already satisfy this demand?". Today the drafting agent gathers +this by hand. Make it reproducible with one command that runs under the +`ops-mason-build` grant. + +Founder decision: ops-warden routes, ops-mason surveys. ops-warden gets a +pointer-only catalog entry (WARDEN-WP-0042), no survey code and no token. + +## Boundaries + +- Metadata only. No `kv get`, no `*/data/*` read, no role-id or secret-id read; + enforced by an allowlist in the script and asserted by tests. +- `policies/ops-mason-build.hcl` is not changed. A query the grant does not + permit is reported as `not permitted by ops-mason-build`, not widened. +- No live OpenBao call in this workplan's implementation; tests use a fake bao. + +## Tasks + +## Implement scripts/plan-survey.py + +```task +id: MASON-WP-0008-T01 +status: done +priority: high +``` + +Script in the `custody-inventory.py` idiom (stdlib plus PyYAML, grant token +from `~/.claude-bao-token`, `BAO_ADDR` default `http://127.0.0.1:18200`, loud +failure). Takes KV path, proposed policy, AppRole and/or Kubernetes role, +consumer and keywords. Runs `bao token lookup` (display_name and policies only; +refuses any token other than `ops-mason-build` plus `default`), +`bao secrets list`, `bao auth list`, `bao policy list`, +`bao list auth//role` for approle and kubernetes mounts, `bao read` of +name-matched role configs, `bao policy read` for name-matched policies, +`bao kv metadata get` on the target and `bao kv list` on its parent. Greps the +ops-warden catalog (`--catalog`, default +`~/ops-warden/registry/routing/catalog.yaml`). Emits a pasteable section 2 with +counts, timestamp, `BAO_ADDR`, display_name and a draft yes/no/partial answer. + +## Unit tests with a fake bao runner + +```task +id: MASON-WP-0008-T02 +status: done +priority: high +``` + +`tests/test_plan_survey.py`: yes/no/partial verdicts, denied capabilities +degrade and are listed as incomplete, no grant and non-grant tokens fail +loudly, the token never appears in output, every recorded call is +metadata-only, and the guard refuses `kv get`, `*/data/*`, role-id/secret-id +and write calls. + +## Document the command + +```task +id: MASON-WP-0008-T03 +status: done +priority: medium +``` + +`docs/construction-plan-format.md` § 2 (grant, then survey) and the OpenBao +Access section of `AGENTS.md`. + +## Route to it from ops-warden + +```task +id: MASON-WP-0008-T04 +status: done +priority: medium +``` + +Pointer-only catalog entry `openbao-structure-survey` in ops-warden +(WARDEN-WP-0042): owner ops-mason, `warden_executes: false`, points to the +grant, this script and the format doc. + +## First live survey: inbound-library IMAP lane + +```task +id: MASON-WP-0008-T05 +status: wait +priority: medium +blocking_reason: "Needs a founder-minted ops-mason-build grant (./scripts/bao-session.sh grant MASON-WP-0008-T05); agents do not log in or grant." +``` + +Run against the real store for `platform/workloads/inbound-library/imap`, +policy `workload-kv-read-inbound-library-imap`, AppRole `inbound-library`, +consumer `inbound-library`, and check that `bao list auth//role`, +role reads and `bao policy read` behave as the policy file implies +(`auth/approle/role/*`, `auth/kubernetes/role/*`, `sys/policies/acl/*`). +Record any `not permitted by ops-mason-build` line as a founder decision, +not a policy edit.