railiance-platform/scripts/agent_high_risk_boundary.py

#!/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())