#!/usr/bin/env python3 """Emit the versioned high-risk data-path artifact (WARDEN-WP-0033-T03). `railiance-platform` asked for a generated list of concrete high-risk KV data paths to consume, instead of hand-maintaining the deny set in `agent-high-risk-boundary.hcl`. Hand-maintaining it is what let the two lists drift for four lanes without anyone noticing (`RISK-F-0009`). **This artifact is an input, not a policy.** It states which paths ops-warden grades high. It does not say what to deny -- railiance-platform owns that, and `ADR-0002` keeps ops-warden a conduit rather than the author of another repo's control. A consumer is free to deny more, deny less, or disagree with a grade. Carries the catalog git revision so a consumer can tell exactly what it was derived from, and regenerate or diff against it. Read-only: it reads the catalog and `git`, never OpenBao and never a secret value. """ from __future__ import annotations import argparse import subprocess import sys from datetime import datetime, timezone from pathlib import Path REPO = Path(__file__).resolve().parent.parent CATALOG = REPO / "registry" / "routing" / "catalog.yaml" DEFAULT_OUT = REPO / "registry" / "generated" / "high-risk-data-paths.yaml" def catalog_revision() -> tuple[str, str]: """(commit, iso-date) of the last change to the catalog. Never guesses.""" try: out = subprocess.run( ["git", "log", "-1", "--format=%H %cI", "--", str(CATALOG)], cwd=REPO, capture_output=True, text=True, timeout=15, check=True, ).stdout.strip() commit, _, date = out.partition(" ") return commit or "unknown", date or "unknown" except (subprocess.SubprocessError, FileNotFoundError): return "unknown", "unknown" def dirty() -> bool: """True if the catalog has uncommitted edits -- the revision would be a lie.""" try: out = subprocess.run( ["git", "status", "--porcelain", "--", str(CATALOG)], cwd=REPO, capture_output=True, text=True, timeout=15, check=True, ).stdout.strip() return bool(out) except (subprocess.SubprocessError, FileNotFoundError): return False def build() -> tuple[str, int]: import yaml entries = yaml.safe_load(CATALOG.read_text())["entries"] commit, date = catalog_revision() rows, patternish = [], [] for entry in sorted(entries, key=lambda e: e["id"]): if entry.get("risk") != "high": continue template = entry.get("path_template") data_path = _to_data_path(template) if template else None if data_path is None: patternish.append(entry["id"]) continue rows.append({ "id": entry["id"], "data_path": data_path, "metadata_path": data_path.replace("/data/", "/metadata/", 1), "fields": entry.get("fields"), "owner_repo": entry.get("owner_repo"), }) lines = [ "# GENERATED by scripts/emit_high_risk_paths.py -- do not edit by hand.", "# Concrete KV data paths for lanes ops-warden grades `risk: high`.", "#", "# This is an INPUT, not a policy. ops-warden states which paths it grades", "# high; railiance-platform owns what agent-high-risk-boundary denies and may", "# deny more, deny less, or dispute a grade (ADR-0002, ADR-0008).", "#", "# Grades cover every field a read of the path discloses, not the field the", "# lane is named after (ADR-0008). `fields` is recorded where an owning CCR", "# declares it, and is null where the field set has not been established --", "# null means unknown, never 'one field'.", "", f"generated_at: \"{datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')}\"", "source: ops-warden/registry/routing/catalog.yaml", f"catalog_revision: \"{commit}\"", f"catalog_revision_date: \"{date}\"", f"catalog_dirty: {str(dirty()).lower()}", f"high_risk_lane_count: {len([e for e in entries if e.get('risk') == 'high'])}", f"concrete_path_count: {len(rows)}", "", "# Graded high but not a single KV address -- a routing pattern, a broker", "# grant, or a non-KV lane. Nothing here for a policy to deny.", "no_concrete_path:", ] lines += [f" - {i}" for i in sorted(patternish)] or [" []"] lines += ["", "paths:"] for row in rows: lines.append(f" - id: {row['id']}") lines.append(f" data_path: {row['data_path']}") lines.append(f" metadata_path: {row['metadata_path']}") lines.append(f" owner_repo: {row['owner_repo']}") if row["fields"]: lines.append(f" fields: [{', '.join(row['fields'])}]") else: lines.append(" fields: null # field set not established -- unknown, not one") return "\n".join(lines) + "\n", len(rows) def _to_data_path(template: str) -> str | None: import re if re.search(r"[<>{}*]", template) or " " in template or template.startswith("k8s:"): return None mount, _, rest = template.partition("/") return f"{mount}/data/{rest}" if rest else None def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--out", type=Path, default=DEFAULT_OUT) parser.add_argument("--check", action="store_true", help="exit 1 if the artifact on disk is stale (for CI)") args = parser.parse_args() content, count = build() if args.check: current = args.out.read_text() if args.out.exists() else "" # generated_at always differs; compare everything else. def strip_generated_at(text: str) -> str: return "\n".join( line for line in text.splitlines() if not line.startswith("generated_at:") ) if strip_generated_at(current) != strip_generated_at(content): print(f"STALE: {args.out} does not match the catalog. Re-run without --check.") return 1 print(f"fresh: {args.out} matches the catalog ({count} concrete paths)") return 0 args.out.parent.mkdir(parents=True, exist_ok=True) args.out.write_text(content) print(f"wrote {args.out} — {count} concrete high-risk data paths") if dirty(): print(" ! catalog has uncommitted changes; catalog_revision does not describe it") return 0 if __name__ == "__main__": sys.exit(main())