#!/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-simplechatbot.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], bool]: username = os.environ.get("HERMES_DASHBOARD_BASIC_AUTH_USERNAME", "admin") password = os.environ.get("HERMES_ADMIN_PASSWORD", "") cookie = os.environ.get("HERMES_DASHBOARD_COOKIE", "") hf_token = os.environ.get("HF_TOKEN", "").strip() headers = {"Accept": "application/json", "Cache-Control": "no-cache"} if hf_token: # Private Spaces require a Hugging Face bearer token at the proxy. The # dashboard's Basic auth cannot share the Authorization header, so use # the password-login route below to establish the application cookie. headers["Authorization"] = f"Bearer {hf_token}" if cookie: headers["Cookie"] = cookie auth = httpx.BasicAuth(username, password) if password and not hf_token and not cookie else None needs_password_login = bool(hf_token and password and not cookie) return auth, headers, needs_password_login async def _establish_dashboard_session( client: httpx.AsyncClient, needs_password_login: bool, ) -> dict[str, Any]: if not needs_password_login: return {"attempted": False} username = os.environ.get("HERMES_DASHBOARD_BASIC_AUTH_USERNAME", "admin") response = await client.post( "/auth/password-login", json={ "provider": "basic", "username": username, "password": os.environ.get("HERMES_ADMIN_PASSWORD", ""), "next": "", }, ) return { "attempted": True, "statusCode": response.status_code, "cookieEstablished": bool(client.cookies), } 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, needs_password_login = _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: report["checks"]["authentication"] = await _establish_dashboard_session( client, needs_password_login, ) 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())