#!/usr/bin/env python3 """Read-only deployed-runtime audit for Hermes Futures Desk. This utility intentionally never calls the Paper Execute endpoint. Credentials are read only from environment variables and are never written to the report. It can verify the served dashboard hash, authenticated Futures API response shapes, all supported chart intervals, and an optional analysis-only request. """ from __future__ import annotations import argparse import asyncio import hashlib import json import os import re import sys import time from pathlib import Path from typing import Any import httpx DEFAULT_BASE_URL = "https://really-amin-asset.hf.space" SUPPORTED_INTERVALS = ("1m", "5m", "15m", "1h") def _redact(value: Any) -> Any: if isinstance(value, dict): result = {} for key, item in value.items(): if re.search(r"(?i)(authorization|cookie|token|secret|password|api[_-]?key)", str(key)): result[str(key)] = "[redacted]" else: result[str(key)] = _redact(item) return result if isinstance(value, list): return [_redact(item) for item in value[:200]] if isinstance(value, str): text = re.sub( r"(?i)\b(authorization|cookie|token|secret|password|api[_-]?key)\b\s*[:=]\s*[^\s,;]+", r"\1=[redacted]", value, ) return text[:2000] return value def _response_summary(response: httpx.Response, payload: Any = None) -> dict[str, Any]: return { "statusCode": response.status_code, "contentType": response.headers.get("content-type"), "elapsedMs": round(response.elapsed.total_seconds() * 1000, 2), "payload": _redact(payload), } def _market_shape(payload: Any) -> dict[str, Any]: if not isinstance(payload, dict): return {"valid": False, "reason": "Payload is not an object"} candles = payload.get("candles") required_keys = { "state", "symbol", "interval", "candles", "currentPrice", "fundingRate", "openInterest", "source", "freshness", "tradingReadiness", } missing = sorted(required_keys - set(payload)) real_candles = isinstance(candles, list) and all( isinstance(item, dict) and all(key in item for key in ("timestamp", "open", "high", "low", "close", "volume")) for item in candles ) return { "valid": not missing and isinstance(candles, list) and real_candles, "missingKeys": missing, "candleCount": len(candles) if isinstance(candles, list) else None, "state": payload.get("state"), "freshness": payload.get("freshness"), "source": payload.get("source"), "verifiedFutures": payload.get("verifiedFutures"), "tradingReadiness": payload.get("tradingReadiness"), } def _client_auth() -> tuple[httpx.BasicAuth | None, dict[str, str]]: username = os.environ.get("HERMES_DASHBOARD_BASIC_AUTH_USERNAME", "admin") password = os.environ.get("HERMES_ADMIN_PASSWORD", "") cookie = os.environ.get("HERMES_DASHBOARD_COOKIE", "") auth = httpx.BasicAuth(username, password) if password else None headers = {"Accept": "application/json", "Cache-Control": "no-cache"} if cookie: headers["Cookie"] = cookie return auth, headers async def _json_request( client: httpx.AsyncClient, method: str, path: str, *, json_body: dict[str, Any] | None = None, ) -> tuple[httpx.Response, Any]: response = await client.request(method, path, json=json_body) try: payload = response.json() except ValueError: payload = {"error": "Response was not valid JSON", "bodyPreview": response.text[:500]} return response, payload async def audit(args: argparse.Namespace) -> dict[str, Any]: auth, headers = _client_auth() report: dict[str, Any] = { "generatedAt": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "baseUrl": args.base_url.rstrip("/"), "symbol": args.symbol.upper(), "analysisRequested": bool(args.analyze), "paperExecuteCalled": False, "checks": {}, } timeout = httpx.Timeout(args.timeout, connect=min(args.timeout, 10.0)) async with httpx.AsyncClient( base_url=args.base_url.rstrip("/"), auth=auth, headers=headers, follow_redirects=True, timeout=timeout, ) as client: dashboard = await client.get("/futures", headers={**headers, "Accept": "text/html"}) body_hash = hashlib.sha256(dashboard.content).hexdigest() header_hash = dashboard.headers.get("x-hermes-template-sha256") report["checks"]["dashboard"] = { "statusCode": dashboard.status_code, "bodySha256": body_hash, "headerTemplateSha256": header_hash, "headerRouterSha256": dashboard.headers.get("x-hermes-router-sha256"), "hashMatchesHeader": bool(header_hash and header_hash == body_hash), "cacheControl": dashboard.headers.get("cache-control"), "containsMarketEndpoint": "/api/futures/market" in dashboard.text, "containsPaperExecuteEndpoint": "/api/futures/paper/execute" in dashboard.text, } for name, path in ( ("status", "/api/futures/status"), ("symbols", "/api/futures/symbols"), ("positions", "/api/futures/positions"), ): response, payload = await _json_request(client, "GET", path) report["checks"][name] = _response_summary(response, payload) market_checks: dict[str, Any] = {} for interval in SUPPORTED_INTERVALS: path = ( f"/api/futures/market?symbol={args.symbol.upper()}" f"&interval={interval}&limit={args.limit}" ) response, payload = await _json_request(client, "GET", path) market_checks[interval] = { **_response_summary(response), "shape": _market_shape(payload), "payload": _redact(payload), } report["checks"]["marketIntervals"] = market_checks if args.analyze: response, payload = await _json_request( client, "POST", "/api/futures/analyze", json_body={ "symbol": args.symbol.upper(), "risk_profile": args.risk_profile, "include_external_context": False, }, ) report["checks"]["analysis"] = _response_summary(response, payload) statuses: list[bool] = [] dashboard_check = report["checks"]["dashboard"] statuses.append(dashboard_check["statusCode"] == 200) statuses.append(dashboard_check["hashMatchesHeader"] is True) for name in ("status", "symbols", "positions"): statuses.append(report["checks"][name]["statusCode"] == 200) for interval in SUPPORTED_INTERVALS: check = report["checks"]["marketIntervals"][interval] statuses.append(check["statusCode"] in {200, 503}) statuses.append(check["shape"]["valid"] is True) if args.analyze: statuses.append(report["checks"]["analysis"]["statusCode"] == 200) report["summary"] = { "passed": all(statuses), "checkCount": len(statuses), "failedCheckCount": sum(1 for value in statuses if not value), "credentialsProvided": bool( os.environ.get("HERMES_ADMIN_PASSWORD") or os.environ.get("HERMES_DASHBOARD_COOKIE") ), "note": "No Paper, Testnet, or Live execution endpoint was called.", } return report def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "--base-url", default=os.environ.get("HERMES_FUTURES_BASE_URL", DEFAULT_BASE_URL), ) parser.add_argument("--symbol", default="BTCUSDT") parser.add_argument("--limit", type=int, default=120, choices=range(20, 501), metavar="20..500") parser.add_argument( "--risk-profile", choices=("conservative", "moderate", "aggressive"), default="moderate", ) parser.add_argument( "--analyze", action="store_true", help="Also perform one analysis-only POST. Paper Execute is still never called.", ) parser.add_argument("--timeout", type=float, default=30.0) parser.add_argument("--report", type=Path, default=Path("futures_runtime_audit.json")) return parser.parse_args() def main() -> int: args = parse_args() try: report = asyncio.run(audit(args)) except Exception as exc: # Audit failure must still avoid leaking credentials. report = { "generatedAt": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "baseUrl": args.base_url, "paperExecuteCalled": False, "summary": {"passed": False, "error": type(exc).__name__}, } args.report.parent.mkdir(parents=True, exist_ok=True) args.report.write_text(json.dumps(_redact(report), indent=2, ensure_ascii=False), encoding="utf-8") print(f"Runtime audit report written to {args.report}") print("Paper Execute was not called.") return 0 if report.get("summary", {}).get("passed") else 1 if __name__ == "__main__": sys.exit(main())