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