Verify the OpenBao read-boundary live; ship the invariant

The operator token was not expired after all -- `bao policy read` succeeded, so
the deployed policy is now compared directly instead of the file. Three
corrections to RISK-F-0009, which was filed static:

1. Six high-risk lanes are uncovered, not eight. The finding counted
   openbao-api-key (a path pattern) and ops-warden-warden-sign-token (a broker
   grant, not KV) among the concrete uncovered paths, while its own prose said
   the first was a pattern. Five lanes have no address for a policy to deny.
2. Coverage holds at 6 of 17 against the live policy.
3. The deployed policy has drifted from the file: the file denies
   platform/workloads/core-hub/runtime, the server does not. No ops-warden lane
   maps there so our numbers are unchanged, but it proves the file was never a
   safe proxy for the server -- which is what the finding flagged as unconfirmed.

scripts/check_agent_read_boundary.py is the invariant RISK-F-0009 asked for
rather than a one-off audit: it fails when a high-risk lane has no corresponding
deny. Capabilities-only by construction -- it reads the policy document, never a
secret value, and never mints a token.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
tegwick 2026-08-21 00:50:00 +02:00
parent 801bbe7b35
commit edb587851c
2 changed files with 265 additions and 0 deletions

View file

@ -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. `<mount>/rest` -> `<mount>/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())

View file

@ -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/<domain>/<workload>/<bundle>") 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"]