Spaces:
Sleeping
Sleeping
File size: 4,707 Bytes
2e658e7 1f8cf56 2e658e7 1f8cf56 2e658e7 1f8cf56 2e658e7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 | #!/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())
|