Spaces:
Sleeping
Sleeping
| #!/usr/bin/env python3 | |
| """Fail when an external Hermes client can bypass the execution service.""" | |
| from __future__ import annotations | |
| import ast | |
| from pathlib import Path | |
| ROOT = Path(__file__).resolve().parents[1] | |
| OVERLAY = ROOT / "hermes_overlay" | |
| SERVICE = OVERLAY / "trading" / "domain" / "execution_service.py" | |
| def _production_python_files() -> list[Path]: | |
| return sorted( | |
| path for path in OVERLAY.rglob("*.py") | |
| if "tests" not in path.parts and "__pycache__" not in path.parts | |
| ) | |
| def audit(root: Path = ROOT) -> list[str]: | |
| findings: list[str] = [] | |
| overlay = root / "hermes_overlay" | |
| service = overlay / "trading" / "domain" / "execution_service.py" | |
| for path in sorted( | |
| p for p in overlay.rglob("*.py") | |
| if "tests" not in p.parts and "__pycache__" not in p.parts | |
| ): | |
| text = path.read_text(encoding="utf-8") | |
| rel = path.relative_to(root).as_posix() | |
| if path != service: | |
| for token in ("._submit_entry(", "._submit_close("): | |
| if token in text: | |
| findings.append(f"private execution mutation referenced by {rel}: {token}") | |
| if path not in {service, overlay / "trading" / "adapters" / "paper_adapter.py"}: | |
| if ".create_order(" in text: | |
| findings.append(f"raw adapter order call outside service: {rel}") | |
| dashboard = (overlay / "tools" / "futures_dashboard_api.py").read_text(encoding="utf-8") | |
| if "ExecutionService().execute_approved_plan(ExecuteApprovedPlanCommand(" not in dashboard: | |
| findings.append("dashboard/API execute route does not use identifier-only service command") | |
| route_slice = dashboard[dashboard.find("async def futures_paper_execute"):] | |
| route_slice = route_slice[:route_slice.find("\n\n_") if "\n\n_" in route_slice else len(route_slice)] | |
| for token in ("._submit_entry(", "execute_futures_position(", ".create_order("): | |
| if token in route_slice: | |
| findings.append(f"dashboard/API execute route contains forbidden call: {token}") | |
| telegram = (overlay / "tools" / "telegram_bot.py").read_text(encoding="utf-8") | |
| if "execute=True" in telegram: | |
| findings.append("Telegram may request direct execution") | |
| for token in ("execute_approved_plan", "close_position(", "._submit_entry(", "._submit_close("): | |
| if token in telegram: | |
| findings.append(f"Telegram contains mutation path: {token}") | |
| agent_impl = (overlay / "plugins" / "futures_trading" / "tools.py").read_text(encoding="utf-8") | |
| plugin_init = (overlay / "plugins" / "futures_trading" / "__init__.py").read_text(encoding="utf-8") | |
| for name in ("execute_futures_position", "close_futures_position", "set_leverage_and_margin"): | |
| if f'name="{name}"' in plugin_init: | |
| findings.append(f"agent mutation tool registered: {name}") | |
| if "execute=False" not in agent_impl: | |
| findings.append("agent analysis workflow does not force execute=False") | |
| cycle_path = overlay / "trading" / "trade_cycle.py" | |
| cycle = cycle_path.read_text(encoding="utf-8") | |
| if "execute_futures_position" in cycle: | |
| findings.append("analysis cycle still imports or calls raw execution") | |
| if "Direct cycle execution is disabled" not in cycle: | |
| findings.append("analysis cycle does not explicitly reject execute=True") | |
| compat_path = overlay / "trading" / "futures_execution.py" | |
| compat = compat_path.read_text(encoding="utf-8") | |
| tree = ast.parse(compat) | |
| raw_entry = next( | |
| node for node in tree.body | |
| if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) | |
| and node.name == "execute_futures_position" | |
| ) | |
| raw_source = ast.get_source_segment(compat, raw_entry) or "" | |
| if "direct raw futures execution is disabled" not in raw_source: | |
| findings.append("legacy raw entry compatibility API is not hard-rejected") | |
| close_fn = next( | |
| node for node in tree.body | |
| if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) | |
| and node.name == "close_futures_position" | |
| ) | |
| close_source = ast.get_source_segment(compat, close_fn) or "" | |
| if "ClosePositionCommand" not in close_source or ".close_position(" not in close_source: | |
| findings.append("emergency close does not use typed service boundary") | |
| if "._submit_close(" in close_source: | |
| findings.append("emergency close calls private mutation method") | |
| return findings | |
| def main() -> int: | |
| findings = audit() | |
| for finding in findings: | |
| print(f"ERROR: {finding}") | |
| print(f"execution_client_path_findings={len(findings)}") | |
| return 1 if findings else 0 | |
| if __name__ == "__main__": | |
| raise SystemExit(main()) | |