diff --git a/scripts/check_agent_read_boundary.py b/scripts/check_agent_read_boundary.py new file mode 100755 index 0000000..97e3c40 --- /dev/null +++ b/scripts/check_agent_read_boundary.py @@ -0,0 +1,165 @@ +#!/usr/bin/env python3 +"""Verify the OpenBao half of the agent read-boundary (ADR-0004, WARDEN-WP-0032-T06). + +`warden access` exits 7 on every `risk: high` lane, but that only protects the +ops-warden path. The OpenBao policy `agent-high-risk-boundary` is what protects a +direct `bao kv get` -- the actual 2026-07-16 disclosure vector. This script +compares the high-risk lanes in the routing catalog against the paths that policy +actually denies. + +Read-only and capabilities-only by construction: it reads the *policy document* +and lane metadata. It never reads a secret value, and it never mints a token. +See `.claude/rules/credential-routing.md` -- verifying a lane with a read is the +mistake this whole control exists to prevent. + +Prefers the policy deployed on the server (`bao policy read`); falls back to the +file in railiance-platform and says loudly that it did, because deployment drift +is exactly what this check exists to catch. + +Exit codes: 0 every high-risk lane with a concrete path is denied; 1 at least one +is not; 2 the policy could not be obtained from either source. +""" + +from __future__ import annotations + +import argparse +import json +import os +import re +import subprocess +import sys +from pathlib import Path + +REPO = Path(__file__).resolve().parent.parent +CATALOG = REPO / "registry" / "routing" / "catalog.yaml" +POLICY_NAME = "agent-high-risk-boundary" +POLICY_FILE = Path.home() / "railiance-platform" / "openbao" / "policies" / f"{POLICY_NAME}.hcl" + +# A path_template with a placeholder is a pattern, not an address -- it names the +# shape of a lane rather than one secret, so there is nothing for a policy to deny. +PLACEHOLDER = re.compile(r"[<>{}]|\*") + + +def read_deployed_policy() -> tuple[str | None, str]: + """Return (policy_text, source). Server first, file second, neither third.""" + try: + proc = subprocess.run( + ["bao", "policy", "read", "-format=json", POLICY_NAME], + capture_output=True, text=True, timeout=20, + ) + except (FileNotFoundError, subprocess.TimeoutExpired) as exc: + server_err = str(exc) + else: + if proc.returncode == 0: + try: + return json.loads(proc.stdout)["policy"], "server" + except (json.JSONDecodeError, KeyError): + return proc.stdout, "server" + server_err = (proc.stderr or proc.stdout).strip().splitlines()[:1] + server_err = server_err[0] if server_err else f"exit {proc.returncode}" + + if POLICY_FILE.exists(): + print(f" ! could not read the deployed policy ({server_err})") + print(f" ! falling back to the FILE at {POLICY_FILE}") + print(" ! deployment drift cannot be detected in this mode\n") + return POLICY_FILE.read_text(), "file" + return None, f"unavailable ({server_err})" + + +def denied_data_paths(policy_text: str) -> set[str]: + """Paths the policy denies. Only a `deny` on a KV *data* path is a read-boundary.""" + denied: set[str] = set() + for match in re.finditer( + r'path\s+"([^"]+)"\s*\{[^}]*?capabilities\s*=\s*\[([^\]]*)\]', + policy_text, re.DOTALL, + ): + path, caps = match.group(1), match.group(2) + if "deny" in {c.strip().strip('"\'') for c in caps.split(",")}: + denied.add(path) + return denied + + +def to_data_path(path_template: str) -> str | None: + """Catalog path -> KV v2 data path. `/rest` -> `/data/rest`.""" + if PLACEHOLDER.search(path_template) or " " in path_template: + return None + mount, _, rest = path_template.partition("/") + return f"{mount}/data/{rest}" if rest else None + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--json", action="store_true", help="machine-readable output") + args = parser.parse_args() + + import yaml # local import so --help works without the dep + + entries = yaml.safe_load(CATALOG.read_text())["entries"] + high = [e for e in entries if e.get("risk") == "high"] + + policy_text, source = read_deployed_policy() + if policy_text is None: + print(f"FAIL: policy {POLICY_NAME} could not be obtained from server or file: {source}") + print(" Run `bao login -method=oidc`, or check out railiance-platform.") + return 2 + + denied = denied_data_paths(policy_text) + + covered, uncovered, no_address = [], [], [] + for entry in high: + template = entry.get("path_template") + data_path = to_data_path(template) if template else None + if data_path is None: + no_address.append(entry["id"]) + elif data_path in denied: + covered.append((entry["id"], data_path)) + else: + uncovered.append((entry["id"], data_path)) + + if args.json: + print(json.dumps({ + "policy": POLICY_NAME, + "policy_source": source, + "high_risk_lanes": len(high), + "denied_data_paths": sorted(denied), + "covered": [{"id": i, "path": p} for i, p in covered], + "uncovered": [{"id": i, "path": p} for i, p in uncovered], + "no_concrete_address": no_address, + "ok": not uncovered, + }, indent=2)) + return 1 if uncovered else 0 + + print(f"agent read-boundary — OpenBao half ({POLICY_NAME})\n") + print(f" policy source: {source}" + + (" <-- live" if source == "server" else " <-- NOT the deployed policy")) + print(f" high-risk lanes: {len(high)}") + print(f" denied data paths: {len(denied)}") + print(f" covered: {len(covered)}") + print(f" NOT covered: {len(uncovered)}") + print(f" no concrete address: {len(no_address)}\n") + + if covered: + print("COVERED — a direct `bao kv get` is denied for an agent token") + for lane_id, path in sorted(covered): + print(f" {lane_id:34} {path}") + print() + if uncovered: + print("NOT COVERED — graded high, but the policy does not deny the data path") + for lane_id, path in sorted(uncovered): + print(f" {lane_id:34} {path}") + print() + if no_address: + print("NO CONCRETE ADDRESS — a pattern or a non-KV lane, nothing to deny") + print(" " + ", ".join(sorted(no_address)) + "\n") + + if uncovered: + print(f"RESULT: FAIL — {len(uncovered)} high-risk lane(s) outside the OpenBao boundary.") + print(" The policy is railiance-platform's; ops-warden reports rather than amends" + " (RISK-F-0004).") + return 1 + print("RESULT: PASS — every high-risk lane with a concrete path is denied.") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_agent_read_boundary_check.py b/tests/test_agent_read_boundary_check.py new file mode 100644 index 0000000..db33c1f --- /dev/null +++ b/tests/test_agent_read_boundary_check.py @@ -0,0 +1,100 @@ +"""Tests for scripts/check_agent_read_boundary.py (WARDEN-WP-0032-T06). + +This script is a control, not a report: it is the invariant RISK-F-0009 asked for +("a check that fails when a high-risk lane has no corresponding deny"). So the +parsing has to be right about the two things that would make it lie -- treating a +non-deny grant as a deny, and treating a path pattern as a concrete address. +""" + +import importlib.util +from pathlib import Path + +REPO = Path(__file__).resolve().parent.parent +spec = importlib.util.spec_from_file_location( + "check_agent_read_boundary", REPO / "scripts" / "check_agent_read_boundary.py" +) +mod = importlib.util.module_from_spec(spec) +spec.loader.exec_module(mod) + + +class TestDeniedDataPaths: + def test_extracts_denied_paths(self): + policy = """ + path "platform/data/workloads/forgejo/forgejo-admin" { + capabilities = ["deny"] + } + """ + assert mod.denied_data_paths(policy) == {"platform/data/workloads/forgejo/forgejo-admin"} + + def test_metadata_read_is_not_a_deny(self): + """The policy permits metadata read alongside every data deny. + + Counting those as denies would double the apparent coverage. + """ + policy = """ + path "platform/metadata/workloads/forgejo/forgejo-admin" { + capabilities = ["read"] + } + """ + assert mod.denied_data_paths(policy) == set() + + def test_deny_is_matched_exactly_not_by_substring(self): + """A capability merely containing 'deny' must not register as a deny.""" + policy = """ + path "platform/data/workloads/x/y" { + capabilities = ["denylist-read"] + } + """ + assert mod.denied_data_paths(policy) == set() + + def test_multiple_blocks(self): + policy = """ + path "a/data/one" { capabilities = ["deny"] } + path "a/metadata/one" { capabilities = ["read"] } + path "b/data/two" { capabilities = ["deny"] } + """ + assert mod.denied_data_paths(policy) == {"a/data/one", "b/data/two"} + + +class TestToDataPath: + def test_inserts_kv_v2_data_segment(self): + assert ( + mod.to_data_path("platform/workloads/forgejo/forgejo-admin") + == "platform/data/workloads/forgejo/forgejo-admin" + ) + + def test_tenant_mount(self): + assert ( + mod.to_data_path("tenants/binky/company-email/imap") + == "tenants/data/binky/company-email/imap" + ) + + def test_placeholder_pattern_has_no_address(self): + """`openbao-api-key` is a routing pattern, not one secret. + + RISK-F-0009 counted it among the uncovered lanes; there is nothing for a + policy to deny, and reporting it as a gap overstates the exposure. + """ + assert mod.to_data_path("platform/workloads///") is None + + def test_non_kv_lane_has_no_address(self): + """`ops-warden-warden-sign-token` is a broker grant, not a KV path.""" + assert mod.to_data_path("credential-grants/catalog.yaml grant ops-warden/warden-sign") is None + + +class TestAgainstTheRealCatalog: + def test_every_high_risk_lane_resolves_or_is_explicitly_pattern(self): + """No high-risk lane may fall through the classifier silently. + + Each is either a concrete data path the policy can deny, or a pattern -- + never an unhandled third case, which is the ADR-0007 failure mode. + """ + import yaml + + entries = yaml.safe_load((REPO / "registry" / "routing" / "catalog.yaml").read_text())["entries"] + for entry in (e for e in entries if e.get("risk") == "high"): + template = entry.get("path_template") + if not template: + continue + resolved = mod.to_data_path(template) + assert resolved is None or resolved.count("/data/") == 1, entry["id"]