Compare commits
3 commits
1cd71e74d7
...
0dafb53e84
| Author | SHA1 | Date | |
|---|---|---|---|
| 0dafb53e84 | |||
| edb587851c | |||
| 801bbe7b35 |
4 changed files with 302 additions and 7 deletions
|
|
@ -115,7 +115,7 @@
|
|||
| task | WARDEN-WP-0031-T03 | done | — | workplans/WARDEN-WP-0031-policy-caller-identity.md |
|
||||
| task | WARDEN-WP-0031-T04 | done | — | workplans/WARDEN-WP-0031-policy-caller-identity.md |
|
||||
| task | WARDEN-WP-0031-T05 | cancel | — | workplans/WARDEN-WP-0031-policy-caller-identity.md |
|
||||
| task | WARDEN-WP-0032-T01 | todo | — | workplans/WARDEN-WP-0032-security-zones.md |
|
||||
| task | WARDEN-WP-0032-T01 | done | — | workplans/WARDEN-WP-0032-security-zones.md |
|
||||
| task | WARDEN-WP-0032-T02 | wait | — | workplans/WARDEN-WP-0032-security-zones.md |
|
||||
| task | WARDEN-WP-0032-T03 | wait | — | workplans/WARDEN-WP-0032-security-zones.md |
|
||||
| task | WARDEN-WP-0032-T04 | wait | — | workplans/WARDEN-WP-0032-security-zones.md |
|
||||
|
|
|
|||
165
scripts/check_agent_read_boundary.py
Executable file
165
scripts/check_agent_read_boundary.py
Executable 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())
|
||||
100
tests/test_agent_read_boundary_check.py
Normal file
100
tests/test_agent_read_boundary_check.py
Normal 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"]
|
||||
|
|
@ -293,12 +293,42 @@ direct `bao kv get`, which is the actual 2026-07-16 vector.
|
|||
Routed to `risk-nexus` as **`RISK-F-0004`**, `fix_owner: railiance-platform` —
|
||||
the policy is theirs, and ops-warden does not amend another repo's control.
|
||||
|
||||
**Live confirmation still outstanding.** ops-warden's OpenBao token is expired
|
||||
(`bao token lookup` → 403), so `bao policy read` and `bao token capabilities`
|
||||
could not be run and the deployed policy may differ from the file. A
|
||||
capabilities-only verification script is ready and needs an operator
|
||||
`bao login -method=oidc` to run; it is deliberately capabilities-only, never a
|
||||
read.
|
||||
**Live confirmation done 2026-08-21 — and the blocker was not real.** The
|
||||
token was not expired: `bao token lookup` returned a valid `platform-admin`
|
||||
token from an OIDC login, and `bao policy read agent-high-risk-boundary`
|
||||
succeeded. No `bao login` was needed. Worth recording as a small instance of the
|
||||
lesson `.claude/rules/finding-routing.md` already states — *re-read a blocker
|
||||
before trusting it*. This one was a stale claim about the world, carried for a
|
||||
day in both this workplan and `RISK-F-0009`.
|
||||
|
||||
The verification script also did not exist. It was described here as "ready",
|
||||
and nothing was committed. `scripts/check_agent_read_boundary.py` now exists,
|
||||
with tests, and is the invariant `RISK-F-0009` asked for rather than a one-off
|
||||
audit: exit 1 when any high-risk lane has no corresponding deny. It prefers the
|
||||
deployed policy and falls back to the file only with a loud warning, because
|
||||
deployment drift is the thing it exists to catch. Capabilities-only by
|
||||
construction — it reads the policy document and lane metadata, never a secret
|
||||
value, and never mints a token.
|
||||
|
||||
Three results, one against ourselves:
|
||||
|
||||
1. **Coverage confirmed at 6 of 17** against the live policy — the headline was
|
||||
right and is no longer inferred from a checkout.
|
||||
2. **Six lanes uncovered, not eight.** `RISK-F-0009` counted `openbao-api-key`
|
||||
(a `<domain>/<workload>/<bundle>` 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. Neither can be expressed as a deny.
|
||||
Corrected: 6 covered, 6 uncovered, 5 with no address.
|
||||
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 — it matters as evidence, since
|
||||
`RISK-F-0009` named exactly this divergence as its unconfirmed risk.
|
||||
|
||||
Still not established, and not ops-warden's to establish: whether any agent
|
||||
token carries `agent-high-risk-boundary`, and whether any carries it together
|
||||
with a `workload-kv-read-*` policy. Confirming that means minting or inspecting
|
||||
tokens — a write against railiance-platform's OpenBao. Exposure stays theoretical
|
||||
to the same degree as before.
|
||||
|
||||
**Still open:** whether the maturity-derived default replaces the `ungraded`
|
||||
sentinel entirely (waits on `ZONE-WP-0001-T03`).
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue