scripts/verify_futures_runtime.py

#!/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())