scripts/audit_plan_approval_boundaries.py

#!/usr/bin/env python3
"""Fail-closed static audit for immutable-plan and approval mutation boundaries."""
from __future__ import annotations

import ast
import sys
from dataclasses import dataclass
from pathlib import Path


@dataclass(frozen=True)
class Finding:
    path: str
    line: int
    message: str


CALL_ALLOWLIST = {
    "create_approval": {
        "hermes_overlay/trading/domain/approvals.py",
        "hermes_overlay/tools/futures_dashboard_api.py",
    },
    "validate_approval": {
        "hermes_overlay/trading/domain/approvals.py",
        "hermes_overlay/trading/domain/execution_service.py",
        "hermes_overlay/tools/futures_dashboard_api.py",
    },
    "consume_approval": {
        "hermes_overlay/trading/domain/approvals.py",
        "hermes_overlay/trading/domain/execution_service.py",
    },
    "execute_approved_plan": {
        "hermes_overlay/trading/domain/execution_service.py",
        "hermes_overlay/tools/futures_dashboard_api.py",
    },
    "mark_plan_executed": {
        "hermes_overlay/trading/domain/plans_repo.py",
        "hermes_overlay/trading/domain/execution_service.py",
    },
}

MUTATION_NAMES = {
    "execute_futures_position",
    "close_futures_position",
    "set_futures_leverage",
    "execute_approved_plan",
    "create_approval",
    "consume_approval",
}


def _rel(path: Path, root: Path) -> str:
    return path.relative_to(root).as_posix()


def _call_name(node: ast.Call) -> str | None:
    func = node.func
    if isinstance(func, ast.Name):
        return func.id
    if isinstance(func, ast.Attribute):
        return func.attr
    return None


def audit(root: Path) -> list[Finding]:
    findings: list[Finding] = []
    overlay = root / "hermes_overlay"
    for path in sorted(overlay.rglob("*.py")):
        rel = _rel(path, root)
        if "/tests/" in f"/{rel}/" or "__pycache__" in rel:
            continue
        try:
            tree = ast.parse(path.read_text(encoding="utf-8"), filename=rel)
        except (OSError, SyntaxError) as exc:
            findings.append(Finding(rel, getattr(exc, "lineno", 0) or 0, f"unreadable Python: {exc}"))
            continue
        for node in ast.walk(tree):
            if not isinstance(node, ast.Call):
                continue
            name = _call_name(node)
            if name in CALL_ALLOWLIST and rel not in CALL_ALLOWLIST[name]:
                findings.append(Finding(rel, node.lineno, f"unauthorized {name} call"))

    agent_tool = root / "hermes_overlay/tools/futures_trading_tool.py"
    agent_text = agent_tool.read_text(encoding="utf-8")
    try:
        agent_tree = ast.parse(agent_text, filename=_rel(agent_tool, root))
    except SyntaxError as exc:  # pragma: no cover - handled above too
        findings.append(Finding(_rel(agent_tool, root), exc.lineno or 0, "agent tool syntax invalid"))
    else:
        for node in ast.walk(agent_tree):
            if isinstance(node, ast.Assign):
                for target in node.targets:
                    if isinstance(target, ast.Name) and target.id in {"TOOLS", "TOOLSET", "TOOLS_LIST"}:
                        rendered = ast.unparse(node.value)
                        for name in MUTATION_NAMES:
                            if repr(name) in rendered or f'"{name}"' in rendered:
                                findings.append(Finding(_rel(agent_tool, root), node.lineno, f"agent toolset exposes {name}"))

    telegram = root / "hermes_overlay/tools/telegram_bot.py"
    telegram_text = telegram.read_text(encoding="utf-8")
    for name in MUTATION_NAMES:
        if name in telegram_text:
            findings.append(Finding(_rel(telegram, root), 1, f"Telegram surface references mutation boundary {name}"))

    state = root / "hermes_overlay/trading/state.py"
    state_text = state.read_text(encoding="utf-8")
    required_fragments = (
        "verify_plan_hash(raw",
        "verify_plan_hash(safe",
        "CANONICAL_PLAN_FIELDS",
    )
    for fragment in required_fragments:
        if fragment not in state_text:
            findings.append(Finding(_rel(state, root), 1, f"bounded dashboard copy lacks {fragment}"))

    dashboard = root / "hermes_overlay/tools/futures_dashboard_api.py"
    dashboard_text = dashboard.read_text(encoding="utf-8")
    required_dashboard = (
        "expected_previous_version",
        "expected_previous_version=payload.expected_previous_version",
        "PlanVersionConflict",
        "plan_id=payload.plan_id, approval_id=payload.approval_id",
    )
    for fragment in required_dashboard:
        if fragment not in dashboard_text:
            findings.append(Finding(_rel(dashboard, root), 1, f"dashboard boundary lacks {fragment}"))

    return findings


def main() -> int:
    root = Path(__file__).resolve().parents[1]
    findings = audit(root)
    for finding in findings:
        print(f"{finding.path}:{finding.line}: {finding.message}")
    print(f"findings={len(findings)}")
    print("RESULT: PASS" if not findings else "RESULT: FAIL")
    return 0 if not findings else 1


if __name__ == "__main__":
    raise SystemExit(main())