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