Spaces:
Sleeping
Sleeping
File size: 5,103 Bytes
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 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 | #!/usr/bin/env python3
"""Fail-closed static audit for immutable-plan and approval mutation boundaries."""
from __future__ import annotations
import ast
import sys
from dataclasses import dataclass
from pathlib import Path
@dataclass(frozen=True)
class Finding:
path: str
line: int
message: str
CALL_ALLOWLIST = {
"create_approval": {
"hermes_overlay/trading/domain/approvals.py",
"hermes_overlay/tools/futures_dashboard_api.py",
},
"validate_approval": {
"hermes_overlay/trading/domain/approvals.py",
"hermes_overlay/trading/domain/execution_service.py",
"hermes_overlay/tools/futures_dashboard_api.py",
},
"consume_approval": {
"hermes_overlay/trading/domain/approvals.py",
"hermes_overlay/trading/domain/execution_service.py",
},
"execute_approved_plan": {
"hermes_overlay/trading/domain/execution_service.py",
"hermes_overlay/tools/futures_dashboard_api.py",
},
"mark_plan_executed": {
"hermes_overlay/trading/domain/plans_repo.py",
"hermes_overlay/trading/domain/execution_service.py",
},
}
MUTATION_NAMES = {
"execute_futures_position",
"close_futures_position",
"set_futures_leverage",
"execute_approved_plan",
"create_approval",
"consume_approval",
}
def _rel(path: Path, root: Path) -> str:
return path.relative_to(root).as_posix()
def _call_name(node: ast.Call) -> str | None:
func = node.func
if isinstance(func, ast.Name):
return func.id
if isinstance(func, ast.Attribute):
return func.attr
return None
def audit(root: Path) -> list[Finding]:
findings: list[Finding] = []
overlay = root / "hermes_overlay"
for path in sorted(overlay.rglob("*.py")):
rel = _rel(path, root)
if "/tests/" in f"/{rel}/" or "__pycache__" in rel:
continue
try:
tree = ast.parse(path.read_text(encoding="utf-8"), filename=rel)
except (OSError, SyntaxError) as exc:
findings.append(Finding(rel, getattr(exc, "lineno", 0) or 0, f"unreadable Python: {exc}"))
continue
for node in ast.walk(tree):
if not isinstance(node, ast.Call):
continue
name = _call_name(node)
if name in CALL_ALLOWLIST and rel not in CALL_ALLOWLIST[name]:
findings.append(Finding(rel, node.lineno, f"unauthorized {name} call"))
agent_tool = root / "hermes_overlay/tools/futures_trading_tool.py"
agent_text = agent_tool.read_text(encoding="utf-8")
try:
agent_tree = ast.parse(agent_text, filename=_rel(agent_tool, root))
except SyntaxError as exc: # pragma: no cover - handled above too
findings.append(Finding(_rel(agent_tool, root), exc.lineno or 0, "agent tool syntax invalid"))
else:
for node in ast.walk(agent_tree):
if isinstance(node, ast.Assign):
for target in node.targets:
if isinstance(target, ast.Name) and target.id in {"TOOLS", "TOOLSET", "TOOLS_LIST"}:
rendered = ast.unparse(node.value)
for name in MUTATION_NAMES:
if repr(name) in rendered or f'"{name}"' in rendered:
findings.append(Finding(_rel(agent_tool, root), node.lineno, f"agent toolset exposes {name}"))
telegram = root / "hermes_overlay/tools/telegram_bot.py"
telegram_text = telegram.read_text(encoding="utf-8")
for name in MUTATION_NAMES:
if name in telegram_text:
findings.append(Finding(_rel(telegram, root), 1, f"Telegram surface references mutation boundary {name}"))
state = root / "hermes_overlay/trading/state.py"
state_text = state.read_text(encoding="utf-8")
required_fragments = (
"verify_plan_hash(raw",
"verify_plan_hash(safe",
"CANONICAL_PLAN_FIELDS",
)
for fragment in required_fragments:
if fragment not in state_text:
findings.append(Finding(_rel(state, root), 1, f"bounded dashboard copy lacks {fragment}"))
dashboard = root / "hermes_overlay/tools/futures_dashboard_api.py"
dashboard_text = dashboard.read_text(encoding="utf-8")
required_dashboard = (
"expected_previous_version",
"expected_previous_version=payload.expected_previous_version",
"PlanVersionConflict",
"plan_id=payload.plan_id, approval_id=payload.approval_id",
)
for fragment in required_dashboard:
if fragment not in dashboard_text:
findings.append(Finding(_rel(dashboard, root), 1, f"dashboard boundary lacks {fragment}"))
return findings
def main() -> int:
root = Path(__file__).resolve().parents[1]
findings = audit(root)
for finding in findings:
print(f"{finding.path}:{finding.line}: {finding.message}")
print(f"findings={len(findings)}")
print("RESULT: PASS" if not findings else "RESULT: FAIL")
return 0 if not findings else 1
if __name__ == "__main__":
raise SystemExit(main())
|