Finish coding-agent high-risk boundary coverage

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a02669-87ee-7a31-b111-edc95a16e0fa

scripts/agent_high_risk_boundary.py

@@ -0,0 +1,146 @@
+#!/usr/bin/env python3
+"""Check the platform-owned agent deny policy against ops-warden's risk input."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import sys
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+
+REPO_DIR = Path(__file__).resolve().parents[1]
+DEFAULT_ARTIFACT = (
+    REPO_DIR / "openbao/policies/inputs/ops-warden-high-risk-data-paths.yaml"
+)
+DEFAULT_POLICY = REPO_DIR / "openbao/policies/agent-high-risk-boundary.hcl"
+
+
+def load_artifact(path: Path) -> dict[str, Any]:
+    data = yaml.safe_load(path.read_text(encoding="utf-8"))
+    if not isinstance(data, dict):
+        raise ValueError(f"artifact must be a mapping: {path}")
+    return data
+
+
+def parse_policy(text: str) -> dict[str, set[str]]:
+    paths: dict[str, set[str]] = {}
+    for match in re.finditer(
+        r'path\s+"([^"]+)"\s*\{[^}]*?capabilities\s*=\s*\[([^\]]*)\]',
+        text,
+        re.DOTALL,
+    ):
+        paths[match.group(1)] = {
+            value.strip().strip('"\'')
+            for value in match.group(2).split(",")
+            if value.strip()
+        }
+    return paths
+
+
+def check_boundary(
+    artifact: dict[str, Any], policy: dict[str, set[str]]
+) -> dict[str, Any]:
+    errors: list[str] = []
+    rows = artifact.get("paths")
+    no_concrete = artifact.get("no_concrete_path")
+    if not isinstance(rows, list):
+        rows = []
+        errors.append("artifact paths must be a list")
+    if not isinstance(no_concrete, list):
+        no_concrete = []
+        errors.append("artifact no_concrete_path must be a list")
+    if artifact.get("catalog_dirty") is not False:
+        errors.append("artifact catalog_dirty must be false")
+    if artifact.get("concrete_path_count") != len(rows):
+        errors.append("artifact concrete_path_count does not match paths")
+    if artifact.get("high_risk_lane_count") != len(rows) + len(no_concrete):
+        errors.append("artifact high_risk_lane_count does not match its entries")
+
+    seen_ids: set[str] = set()
+    unique_paths: set[str] = set()
+    for row in rows:
+        if not isinstance(row, dict):
+            errors.append("artifact path entry must be a mapping")
+            continue
+        lane_id = row.get("id")
+        data_path = row.get("data_path")
+        metadata_path = row.get("metadata_path")
+        if not isinstance(lane_id, str) or not lane_id:
+            errors.append("artifact path entry has no id")
+            continue
+        if lane_id in seen_ids:
+            errors.append(f"duplicate lane id: {lane_id}")
+        seen_ids.add(lane_id)
+        if not isinstance(data_path, str) or "/data/" not in data_path:
+            errors.append(f"{lane_id}: invalid data_path")
+            continue
+        expected_metadata = data_path.replace("/data/", "/metadata/", 1)
+        if metadata_path != expected_metadata:
+            errors.append(f"{lane_id}: metadata_path does not match data_path")
+            continue
+        unique_paths.add(data_path)
+        if "deny" not in policy.get(data_path, set()):
+            errors.append(f"{lane_id}: data path is not denied: {data_path}")
+        if "read" not in policy.get(metadata_path, set()):
+            errors.append(f"{lane_id}: metadata path is not readable: {metadata_path}")
+
+    return {
+        "catalog_revision": artifact.get("catalog_revision"),
+        "high_risk_lanes": artifact.get("high_risk_lane_count"),
+        "concrete_entries": len(rows),
+        "unique_concrete_paths": len(unique_paths),
+        "no_concrete_paths": len(no_concrete),
+        "errors": errors,
+        "ok": not errors,
+    }
+
+
+def comparable_artifact(data: dict[str, Any]) -> dict[str, Any]:
+    return {key: value for key, value in data.items() if key != "generated_at"}
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--artifact", type=Path, default=DEFAULT_ARTIFACT)
+    parser.add_argument("--policy", type=Path, default=DEFAULT_POLICY)
+    parser.add_argument(
+        "--upstream",
+        type=Path,
+        help="optionally require the vendored artifact to match this upstream copy",
+    )
+    parser.add_argument("--json", action="store_true")
+    args = parser.parse_args()
+
+    try:
+        artifact = load_artifact(args.artifact)
+        policy = parse_policy(args.policy.read_text(encoding="utf-8"))
+        report = check_boundary(artifact, policy)
+        if args.upstream:
+            upstream = load_artifact(args.upstream)
+            if comparable_artifact(artifact) != comparable_artifact(upstream):
+                report["errors"].append("vendored artifact differs from upstream")
+                report["ok"] = False
+    except (OSError, ValueError, yaml.YAMLError) as exc:
+        report = {"ok": False, "errors": [str(exc)]}
+
+    if args.json:
+        print(json.dumps(report, indent=2, sort_keys=True))
+    elif report["ok"]:
+        print(
+            "PASS: agent boundary covers "
+            f"{report['concrete_entries']} high-risk catalog entries at "
+            f"{report['catalog_revision']}"
+        )
+    else:
+        for error in report["errors"]:
+            print(f"FAIL: {error}", file=sys.stderr)
+    return 0 if report["ok"] else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())