scripts/audit_phase4_adjacent_paths.py

#!/usr/bin/env python3
"""Independent adjacent-gap audit for the Phase 4 execution boundary."""
from __future__ import annotations

import ast
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
OVERLAY = ROOT / "hermes_overlay"
SERVICE = OVERLAY / "trading" / "domain" / "execution_service.py"
PAPER_ADAPTER = OVERLAY / "trading" / "adapters" / "paper_adapter.py"

sys.path.insert(0, str(ROOT / "scripts"))
from audit_execution_client_paths import audit as audit_client_paths  # noqa: E402

_ADAPTER_MUTATIONS = {
    "create_order",
    "cancel_order",
    "configure_account",
    "set_leverage",
    "set_margin_mode",
}
_RAW_HTTP_MUTATIONS = {"post", "put", "patch", "delete"}
_ALLOWED_MUTATION_ROUTES = {
    ("hermes_overlay/tools/futures_dashboard_api.py", "/api/futures/analyze"),
    ("hermes_overlay/tools/futures_dashboard_api.py", "/api/futures/paper/approve"),
    ("hermes_overlay/tools/futures_dashboard_api.py", "/api/futures/paper/execute"),
    ("hermes_overlay/tools/futures_dashboard_api.py", "/api/futures/positions/close"),
    ("hermes_overlay/tools/telegram_bot.py", "/api/telegram/webhook"),
}
_PRIVATE_EXCHANGE_MARKERS = {
    "/fapi/v1/order",
    "/api/v1/order",
    "/api/v2/order",
    "/api/v3/order",
    "X-MBX-APIKEY",
    "KC-API-KEY",
    "KC-API-SIGN",
}


def _production_files(root: Path) -> list[Path]:
    overlay = root / "hermes_overlay"
    return sorted(
        path for path in overlay.rglob("*.py")
        if "tests" not in path.parts and "__pycache__" not in path.parts
    )


def _route_path(decorator: ast.AST) -> tuple[str | None, str | None]:
    if not isinstance(decorator, ast.Call) or not isinstance(decorator.func, ast.Attribute):
        return None, None
    verb = decorator.func.attr.lower()
    if verb not in {"post", "put", "patch", "delete"}:
        return None, None
    if not decorator.args or not isinstance(decorator.args[0], ast.Constant):
        return verb, None
    value = decorator.args[0].value
    return verb, value if isinstance(value, str) else None


def _function_source(text: str, node: ast.AST) -> str:
    return ast.get_source_segment(text, node) or ""


def audit(root: Path = ROOT) -> list[str]:
    findings = list(audit_client_paths(root))
    overlay = root / "hermes_overlay"
    service = overlay / "trading" / "domain" / "execution_service.py"
    paper_adapter = overlay / "trading" / "adapters" / "paper_adapter.py"

    for path in _production_files(root):
        text = path.read_text(encoding="utf-8")
        rel = path.relative_to(root).as_posix()
        tree = ast.parse(text, filename=rel)

        for node in ast.walk(tree):
            if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute):
                attr = node.func.attr
                if attr in _ADAPTER_MUTATIONS and path not in {service, paper_adapter}:
                    findings.append(f"direct adapter mutation call outside service/adapter: {rel}:{node.lineno} {attr}")
                if attr in _RAW_HTTP_MUTATIONS and path.is_relative_to(overlay / "trading"):
                    findings.append(f"raw HTTP mutation verb inside trading package: {rel}:{node.lineno} {attr}")
            elif isinstance(node, ast.Call) and isinstance(node.func, ast.Name):
                if node.func.id == "create_order" and path != service:
                    findings.append(f"direct durable order creation outside service: {rel}:{node.lineno}")

        for node in tree.body:
            if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                continue
            for decorator in node.decorator_list:
                verb, route = _route_path(decorator)
                if verb is None:
                    continue
                if (rel, route or "") not in _ALLOWED_MUTATION_ROUTES:
                    findings.append(f"unreviewed mutation-capable route: {rel}:{node.lineno} {verb.upper()} {route}")
                source = _function_source(text, node)
                for token in ("._submit_entry(", "._submit_close(", ".create_order(", "execute_futures_position("):
                    if token in source:
                        findings.append(f"route contains forbidden execution bypass: {rel}:{node.lineno} {token}")

        if path != service:
            for marker in _PRIVATE_EXCHANGE_MARKERS:
                if marker in text:
                    findings.append(f"private exchange mutation marker outside service: {rel} {marker}")

    # Background/scheduled and debug-adjacent surfaces must remain analysis-only.
    for rel in (
        "hermes_overlay/trading/trade_cycle.py",
        "hermes_overlay/trading/alerts.py",
        "hermes_overlay/trading/provider_reliability.py",
        "hermes_overlay/tools/telegram_bot.py",
    ):
        text = (root / rel).read_text(encoding="utf-8")
        for token in (
            "execute_approved_plan(",
            ".close_position(",
            "._submit_entry(",
            "._submit_close(",
            ".create_order(",
        ):
            if token in text:
                findings.append(f"background/notification surface contains mutation call: {rel} {token}")
        if rel.endswith("trade_cycle.py") and "Direct cycle execution is disabled" not in text:
            findings.append("trade cycle does not fail closed when execute=True")

    service_text = service.read_text(encoding="utf-8")
    for fn_name in ("_submit_entry", "_submit_close"):
        tree = ast.parse(service_text)
        fn = next(
            node for node in ast.walk(tree)
            if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == fn_name
        )
        source = _function_source(service_text, fn)
        required = ("claim_idempotency(", "begin_execution(", "acquire_lock_lease(", "self._adapter.create_order(")
        missing = [token for token in required if token not in source]
        if missing:
            findings.append(f"{fn_name} missing lifecycle controls: {', '.join(missing)}")
        elif not (
            source.index("claim_idempotency(")
            < source.index("begin_execution(")
            < source.index("acquire_lock_lease(")
            < source.index("self._adapter.create_order(")
        ):
            findings.append(f"{fn_name} lifecycle controls are ordered incorrectly")
        if "release_lock(" not in source or "if lease is not None" not in source:
            findings.append(f"{fn_name} does not conditionally release an acquired lease")

    compat = (overlay / "trading" / "futures_execution.py").read_text(encoding="utf-8")
    leverage_start = compat.index("async def set_leverage_and_margin")
    leverage_end = compat.index("\ndef _estimate_slippage_pct", leverage_start)
    leverage_source = compat[leverage_start:leverage_end]
    if '"simulated": True' not in leverage_source or "non-Paper leverage mutation is disabled" not in leverage_source:
        findings.append("legacy leverage compatibility boundary is not Paper-simulated and fail-closed")
    for token in (".set_leverage(", ".set_margin_mode(", ".configure_account("):
        if token in leverage_source:
            findings.append(f"legacy leverage compatibility boundary calls adapter directly: {token}")

    return sorted(set(findings))


def main() -> int:
    findings = audit()
    for finding in findings:
        print(f"ERROR: {finding}")
    print(f"phase4_adjacent_findings={len(findings)}")
    return 1 if findings else 0


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