""" modules/bot_api_server.py — direct HTTP API the AWS bot exposes (v18.0). Replaces the old push-heartbeat/poll-relay pair (bot.py's _status_heartbeat_loop/_command_poll_loop/_handle_query_command, all deleted — see bot.py's v18.0 changelog) that used to bounce every dashboard interaction through a Cloudflare Worker's /status, /command and /command-result routes. That Worker is retired; this bot now runs standalone on AWS (Frankfurt) and serves its own status/control API directly, which the Hugging Face dashboard (app.py, DASHBOARD_ONLY mode) calls over the internet using BOT_API_URL + BOT_API_TOKEN instead of ORACLE_URL + RELAY_AUTH_TOKEN. Endpoints: GET /status — live-computed JSON status snapshot (no auth — same shape the old /status push payload had, plus a new jupiter_tier field). Computed fresh on every request, never cached. POST /command — Bearer-token gated (BOT_API_TOKEN). Body: {"action": str, "params": dict|null}. Routes `action` to the same handler functions the old poll loop called, but SYNCHRONOUSLY — one HTTP request/response, no more request_id + second poll for the result. Returns {"ok": true, "result": "..."} or {"ok": false, "error": "..."}. Run as an asyncio background task from bot.py (self._spawn_background), started unconditionally — with no BOT_API_TOKEN configured, /command always 503s (logged once) but /status still works, since it carries no secrets. """ from __future__ import annotations import asyncio import hmac import logging import os import time from typing import Any logger = logging.getLogger(__name__) _DEFAULT_HOST = "0.0.0.0" _DEFAULT_PORT = 8081 # Fire-and-forget actions — reuse the exact same CommandHandlers methods # Telegram's /ghost_mode, /mint_mode and /hunt already call, and report to # the same Telegram chat, so a dashboard-triggered action leaves the same # audit trail a Telegram command would. Each takes (chat_id) and sends its # own Telegram message; the HTTP response below just confirms it ran. _FIRE_AND_FORGET_ACTIONS = ("toggle_ghost", "toggle_mint", "hunt") # Data-query actions — rendered via the shared modules/dashboard_panels.py # renderers (same ones app.py's non-DASHBOARD_ONLY path and the /speedtest # Telegram command use) and returned directly in the HTTP response body. _QUERY_ACTIONS = ( "wallet_status", "gas_health", "price_check", "scanner_tokens", "solana_prices", "solana_routes", "arbitrage_detail", "audit_log", "speedtest", ) def _bot_api_token() -> str: return os.getenv("BOT_API_TOKEN", "").strip() def _note_auth(request: Any, ok: bool) -> None: """Report an authentication outcome to the HTTP guard (v18.2). Best-effort by design: if the guard is unavailable or raises, the auth decision itself has already been made and must stand. This can only ever add a future reason to refuse, never grant access. """ try: from modules.http_guard import client_key, get_guard guard = get_guard() key = request.get("client_key") or client_key(request) if ok: guard.note_auth_success(key) else: guard.note_auth_failure(key) except Exception as exc: # noqa: BLE001 logger.debug("[BotAPI] could not record auth outcome: %s", exc) def _bot_api_host() -> str: return os.getenv("BOT_API_HOST", _DEFAULT_HOST).strip() or _DEFAULT_HOST def _bot_api_port() -> int: try: return int(os.getenv("BOT_API_PORT", str(_DEFAULT_PORT)).strip()) except ValueError: return _DEFAULT_PORT def _auto_hunt_enabled() -> bool: """Same semantics as bot.py's own _auto_hunt_enabled() — duplicated as a tiny env-var read rather than imported, to avoid a circular import (bot.py imports this module at startup, before its own module-level code has finished running).""" return os.environ.get("AUTO_HUNT_ENABLED", "false").strip().lower() in ( "1", "true", "yes", "on", ) async def _build_status_snapshot(bot: Any) -> dict[str, Any]: """Same fields the old _status_heartbeat_loop's push payload had, plus jupiter_tier (new). Computed fresh every call — no caching.""" from constants import jupiter_tier_status h = getattr(bot, "_handlers", None) payout = getattr(bot, "_payout", None) ledger = await payout.get_status() if payout is not None else None solana = getattr(bot, "_solana", None) _is_paid, tier_name = jupiter_tier_status() return { "uptime_s": int(time.time() - getattr(bot, "start_time", time.time())), "telegram_connected": bool(getattr(getattr(bot, "_tg", None), "reachable", False)), "ghost_mode": bool(getattr(h, "_ghost_mode", False)) if h else False, "mint_mode": bool(getattr(h, "_mint_mode", False)) if h else False, "auto_hunt": _auto_hunt_enabled(), "min_profit_arb": h.effective_min_profit if h else getattr(bot.config, "min_profit_usd", None), # FIX — this used to independently re-read SOLANA_MIN_PROFIT_USD # with its own hardcoded default (2.0, different from # solana_arb.py's own default), the same class of banner-vs- # enforced-value drift already fixed once for mint mode (see # modules/command_handlers.py's v17.20 FIX). Now reads the LIVE # SolanaArbEngine instance's own .min_profit attribute so the # displayed value can never drift from what's actually enforced. "min_profit_sol": solana.min_profit if solana is not None else None, "total_profit_usd": (ledger or {}).get("net_profit", 0.0), "scans_total": (ledger or {}).get("scan_count", 0), "scans_buy": (ledger or {}).get("buy_count", 0), "dry_run": getattr(bot.config, "dry_run", None), "jupiter_tier": tier_name, # v18.1 — operational fields for the public page. Audited as # non-secret before being added here, because this endpoint is # unauthenticated and fronted by bot.elghaly.dev: scan age is a # timestamp delta, route counts are integers, and the governor's # rate is a number about our own pacing. No URL, no address, no # key, no config value that would help an attacker. The RPC URL in # particular stays OUT — SOLANA_RPC_PRIMARY embeds a provider API # key in its path on this deployment. "solana": ({ "last_scan_age_secs": ( round(time.time() - solana.last_scan_ts, 1) if solana.last_scan_ts else None ), "scan_count": solana.scan_count, "signal_count": solana.signal_count, "routes": len(solana.routes), "paused": bool(getattr(solana, "paused", False)), } if solana is not None else None), } async def _dispatch_action(bot: Any, action: str, params: dict) -> str: """Compute one action's result. Fire-and-forget actions return a short confirmation string (the real audit trail is the Telegram message they send); query actions return the rendered markdown directly. Any failure is rendered as an error string rather than raised, so a bad request can't 500 the whole endpoint for an unrelated reason.""" from modules import dashboard_panels as panels handlers = getattr(bot, "_handlers", None) chat_id = getattr(bot.config, "telegram_chat_id", None) if action in _FIRE_AND_FORGET_ACTIONS: if handlers is None: return "❌ Bot not ready yet — handlers not constructed." if action == "toggle_ghost": await handlers._cmd_ghost_mode(chat_id) elif action == "toggle_mint": await handlers._cmd_mint_mode(chat_id) elif action == "hunt": await handlers._cmd_hunt(chat_id) return f"✅ {action} executed — see Telegram for confirmation." if action in _QUERY_ACTIONS: if action == "wallet_status": return await panels.render_wallet_status(bot) if action == "gas_health": return await panels.render_gas_health(bot) if action == "price_check": return await panels.render_price_check(bot, params.get("assets", "")) if action == "scanner_tokens": return await panels.render_scanner_tokens(bot, _auto_hunt_enabled()) if action == "solana_prices": return await panels.render_solana_prices(bot) if action == "solana_routes": return await panels.render_solana_routes(bot) if action == "arbitrage_detail": return await panels.render_arbitrage_detail(bot) if action == "audit_log": return await panels.render_audit_log(bot) if action == "speedtest": return await panels.render_speedtest(bot) return f"❌ Unknown action: {action}" def _read_recent_log_lines(limit: int = 200) -> str: """v18.1 — tail the bot's own rotating log file. Operator need: the Hugging Face dashboard runs in DASHBOARD_ONLY mode and therefore has no bot of its own, so the only logs it could ever show were its own (empty) process — i.e. never the AWS bot actually trading. This serves the real thing. Reads modules/logger.py's LOG_FILE rather than shelling out to journalctl: no subprocess, no sudo, and it works regardless of which user the service runs as or whether journald is even in use. Bounded to the tail so a multi-megabyte log can't be pulled into memory or pushed down the wire in one response.""" path = os.getenv("LOG_FILE", "").strip() if not path: return ("LOG_FILE is not set, so the bot is not writing a log file — " "only stdout/journald. Set LOG_FILE=bot.log in .env and " "restart to enable this view.") try: with open(path, "r", encoding="utf-8", errors="replace") as fh: # Seek to a bounded window from EOF instead of reading the whole # file; 200 lines is far under 256 KiB in this log's format. try: fh.seek(0, os.SEEK_END) fh.seek(max(0, fh.tell() - 256_000)) fh.readline() # discard the partial line the seek landed in except OSError: pass lines = fh.readlines() tail = [ln.rstrip("\n") for ln in lines[-limit:]] return "\n".join(tail) if tail else "(log file is empty)" except FileNotFoundError: return f"Log file not found at {path} — the bot may not have written to it yet." except Exception as exc: return f"Could not read log file: {exc}" async def run_bot_api_server(bot: Any) -> None: """Serve GET /status and POST /command until the bot stops. Spawned as a background task from bot.py's start() — see this module's own docstring for the full endpoint contract.""" try: from aiohttp import web except ImportError: logger.error( "[BotAPI] aiohttp not installed — the bot's HTTP status/" "command API will not run. pip install aiohttp." ) return token = _bot_api_token() if not token: logger.warning( "[BotAPI] BOT_API_TOKEN is not set — /status will still work " "(no secrets in it), but /command will reject every request. " "Set BOT_API_TOKEN to let the Hugging Face dashboard control " "this bot remotely." ) async def handle_status(request: "web.Request") -> "web.Response": try: snapshot = await _build_status_snapshot(bot) return web.json_response(snapshot) except Exception as exc: logger.warning("[BotAPI] /status failed: %s", exc) return web.json_response({"error": str(exc)}, status=500) async def handle_command(request: "web.Request") -> "web.Response": configured_token = _bot_api_token() if not configured_token: return web.json_response( {"ok": False, "error": "BOT_API_TOKEN not configured on this bot"}, status=503, ) auth = request.headers.get("Authorization", "") if not hmac.compare_digest(auth, f"Bearer {configured_token}"): # v18.2 — feed the failure to modules/http_guard.py, which locks # the client out after a handful of attempts. Without this, the # only thing bounding a brute-force run against a fund-moving # endpoint is how fast the attacker's network is. _note_auth(request, ok=False) return web.json_response({"ok": False, "error": "unauthorized"}, status=401) _note_auth(request, ok=True) try: body = await request.json() except Exception: return web.json_response({"ok": False, "error": "invalid JSON body"}, status=400) action = (body or {}).get("action") params = (body or {}).get("params") or {} if not action: return web.json_response({"ok": False, "error": "missing action"}, status=400) try: logger.info("[BotAPI] executing remote command: %s", action) result = await _dispatch_action(bot, action, params) return web.json_response({"ok": True, "result": result}) except Exception as exc: logger.warning("[BotAPI] action %s failed: %s", action, exc) return web.json_response({"ok": False, "error": str(exc)}, status=500) async def handle_logs(request: "web.Request") -> "web.Response": """Bearer-gated like /command, not open like /status: log lines are operational detail (wallet addresses, tx signatures, route sizes), which /status deliberately does not expose.""" configured_token = _bot_api_token() if not configured_token: return web.json_response( {"ok": False, "error": "BOT_API_TOKEN not configured on this bot"}, status=503, ) if not hmac.compare_digest( request.headers.get("Authorization", ""), f"Bearer {configured_token}" ): _note_auth(request, ok=False) return web.json_response({"ok": False, "error": "unauthorized"}, status=401) _note_auth(request, ok=True) try: limit = min(int(request.query.get("limit", 200)), 1000) except ValueError: limit = 200 text = await asyncio.to_thread(_read_recent_log_lines, limit) return web.json_response({"ok": True, "logs": text}) # ── Public read-only surface for bot.elghaly.dev (v18.1, 2026-07-29) ── # # Operator: "a public read-only stats page vs an authenticated control # page — you already have this instinct (the /logs 401 earlier was # correct behavior, keep that pattern everywhere)." # # So these four routes are UNAUTHENTICATED BY DESIGN and carry nothing # that needs protecting: the page itself is a static string, and both # JSON endpoints read only the trade journal (route names, sizes, # spreads, on-chain signatures — all of it either already public on # Solana or meaningless to an attacker). No wallet address, no RPC URL, # no key material, no config, and above all no way to ACT. /command # keeps its Bearer token; /logs keeps its 401. # # That line matters more than it looks: this page is meant to be public # at bot.elghaly.dev, and a public page that can move funds is a public # page that will move funds. async def handle_index(request: "web.Request") -> "web.Response": from modules.web_dashboard import INDEX_HTML return web.Response(text=INDEX_HTML, content_type="text/html") async def handle_api_pnl(request: "web.Request") -> "web.Response": from modules.ledger_report import pnl # to_thread: these read (tailed) CSVs off disk, and the event loop # this shares with the scan loop must never block on I/O. return web.json_response(await asyncio.to_thread(pnl)) async def handle_api_receipts(request: "web.Request") -> "web.Response": from modules.ledger_report import receipts try: # max(1, ...) as well as min(..., 500): Python slices backwards # on a negative bound, so `?limit=-1` would have returned # nearly every row and defeated the cap entirely. This endpoint # is deliberately unauthenticated, so its parameters have to be # robust against someone who is not being polite. limit = max(1, min(int(request.query.get("limit", 100)), 500)) except (TypeError, ValueError): limit = 100 rows = await asyncio.to_thread(receipts, limit) return web.json_response({ "receipts": rows, "verifiable": True, "note": ( "Each signature is independently checkable on Solana. This " "endpoint returns ONLY trades that produced a real on-chain " "signature — never estimates, simulations or projections." ), }) async def handle_receipts_csv(request: "web.Request") -> "web.Response": from modules.ledger_report import receipts_csv text = await asyncio.to_thread(receipts_csv) return web.Response( text=text, content_type="text/csv", headers={"Content-Disposition": 'attachment; filename="receipts.csv"'}, ) async def handle_api_routes(request: "web.Request") -> "web.Response": """v18.2 — route-pruning transparency, public. Operator: "expose why a route got pruned (e.g. '0 signals in 500 evals over 6h') so you can tell a dead route from a temporarily quiet one." Serves route_scorer.public_table(), which is the internal table with every dollar figure removed — see that method for the audit. Route names and basis points are safe to publish; loan sizes are position information and are not. """ from modules.route_scorer import get_scorer scorer = get_scorer() try: table = await asyncio.to_thread(scorer.public_table) status = scorer.status() except Exception as exc: # noqa: BLE001 logger.debug("[BotAPI] /api/routes failed: %s", exc) return web.json_response({"routes": [], "error": "unavailable"}) return web.json_response({ "routes": table, "pruning_enabled": status.get("enabled"), "routes_pruned": status.get("routes_pruned"), "min_active_routes": status.get("min_active_routes"), "dead_bps": status.get("dead_bps"), "floor_bound": status.get("floor_bound"), "probe_every_cycles": status.get("probe_every_cycles"), "note": ( "A pruned route is asleep on a timer, not deleted — it is " "re-probed on a fixed cycle and reinstated the moment it " "produces. `explain` states the evidence and the observation " "window for every route, pruned or not." ), }) async def handle_api_engine(request: "web.Request") -> "web.Response": """v18.3 — the engine's own instruments, in one payload. Operator: "add some more data" — and, in the same breath, "make secure just me". Those two travel together: this endpoint carries edge-decay measurements, Jito tip state, the trading circuit breaker and the drawdown watcher, which is meaningfully more than the earlier public payloads exposed. Each block is still built by hand rather than dumping the modules' full status() dicts, because those are written for /doctor — an authenticated, operator-only view — and would carry the signing wallet, file paths and threshold config out onto a web page. The rule from the public endpoints has not been relaxed just because there is a password in front of it now: a password is one credential, and a page that leaks a wallet address to whoever holds it is still a page that leaks a wallet address. """ out: dict[str, Any] = {} try: from modules.decay_tracker import get_tracker st = get_tracker().status() out["decay"] = { "recording": st.get("recording"), "samples": st.get("samples"), "min_samples": st.get("min_samples"), "usd_per_sec": st.get("decay_usd_per_sec"), "pipeline_median_secs": st.get("pipeline_median_secs"), "pipeline_p90_secs": st.get("pipeline_p90_secs"), "implied_adder_usd": st.get("implied_adder_usd"), "gate_enabled": st.get("gate_enabled"), } except Exception as exc: # noqa: BLE001 logger.debug("[BotAPI] decay block failed: %s", exc) try: from modules.trading_guard import get_guard st = get_guard().status() out["guard"] = { "state": st.get("state"), "trips": st.get("trips"), "attempts": st.get("attempts"), "successes": st.get("successes"), "blocked": st.get("blocked"), "reason": (st.get("trip_reason") or "")[:200], } except Exception as exc: # noqa: BLE001 logger.debug("[BotAPI] guard block failed: %s", exc) try: from modules.jito_tip_engine import get_tip_engine st = get_tip_engine().status() out["jito"] = { "gamma": st.get("gamma"), "bundles_sent": st.get("bundles_submitted"), "bundles_landed": st.get("bundles_landed"), "landing_rate": st.get("landing_rate"), "priced_out": st.get("priced_out"), "tips_paid_sol": st.get("tip_sol_paid"), # COUNT, not the list. The block-engine URLs are not secret, # but they are infrastructure detail with no reason to be on # a web page — and "4 engines" is the whole of what a reader # needs from that field. "engines": len(st.get("block_engines") or []), } except Exception as exc: # noqa: BLE001 logger.debug("[BotAPI] jito block failed: %s", exc) try: from modules.drawdown_alert import get_watcher st = get_watcher().status() out["drawdown"] = { "enabled": st.get("enabled"), "halt_on_breach": st.get("halt_on_breach"), "thresholds": st.get("thresholds"), "latches": st.get("latches"), } except Exception as exc: # noqa: BLE001 logger.debug("[BotAPI] drawdown block failed: %s", exc) try: from modules.http_guard import get_guard as get_http_guard st = get_http_guard().status() out["http"] = { "served": st.get("served"), "throttled_429": st.get("throttled_429"), "shed_503": st.get("shed_503"), "auth_lockouts": st.get("auth_lockouts"), "locked_clients": st.get("locked_clients"), } except Exception as exc: # noqa: BLE001 logger.debug("[BotAPI] http block failed: %s", exc) return web.json_response(out) async def handle_api_nearmiss(request: "web.Request") -> "web.Response": """v18.2 — failure-mode logging, public. Operator: "logging near-misses — signals that crossed your detection threshold but reverted or got beaten to the block — since that tells you if you're losing races, not just whether you're finding opportunities." Public because it is the honest half of the story. A page showing only executed trades reads as either "no losses" or "no activity" with no way to tell which; publishing the near-misses alongside is what makes the receipts section a claim about completeness rather than a selection. """ from modules.near_miss import get_log near = get_log() try: limit = max(1, min(int(request.query.get("limit", 25)), 200)) except (TypeError, ValueError): limit = 25 try: summary = await asyncio.to_thread(near.summary) recent = await asyncio.to_thread(near.recent, limit) except Exception as exc: # noqa: BLE001 logger.debug("[BotAPI] /api/nearmiss failed: %s", exc) return web.json_response({"summary": {}, "recent": [], "error": "unavailable"}) # log_path is a server filesystem path — useful in /doctor, not on # a public endpoint, so it is dropped here rather than never # collected. summary.pop("log_path", None) return web.json_response({"summary": summary, "recent": recent}) async def handle_api_observatory(request: "web.Request") -> "web.Response": """v18.5 — the recorded series, not another snapshot. Operator's request, in their own framing: a public view of Solana tip/MEV conditions that runs continuously, because other bot builders want that data and producing it risks no capital. Everything here is either published by Jito already or measured from public quotes, so there is nothing to withhold — but the endpoint-name labels from rpc_latency are deliberately env-var names ("SOLANA_RPC_PRIMARY"), never URLs, because the primary RPC embeds a provider API key in its path on this deployment. See scripts/observe.py, which records them that way for this reason. """ from modules.observatory import viability try: data = await asyncio.to_thread(viability) except Exception as exc: # noqa: BLE001 logger.debug("[BotAPI] /api/observatory failed: %s", exc) return web.json_response({"rows": 0, "error": "unavailable"}) return web.json_response(data) async def handle_api_lighthouse(request: "web.Request") -> "web.Response": """v18.4 — the verdict that ranks the other panels, public. Operator: "improve his web its so ugly." The page opened with a table, which meant a reader — including the operator on a phone — had to assemble the answer from five sections before knowing whether anything was wrong. This serves the same five-gate judgement /lighthouse gives in Telegram, so the page can lead with the conclusion and keep the evidence underneath it. `public=True` drops the `action` field. The findings carry shell commands naming scripts and env vars, and operator instructions on a web page are exactly what this file's payload rule exists to prevent. The diagnosis stays — knowing that the pipeline is slow is not a credential — only the runbook line is withheld. """ from modules.lighthouse_report import lighthouse try: data = await asyncio.to_thread(lighthouse, bot, True) except Exception as exc: # noqa: BLE001 logger.debug("[BotAPI] /api/lighthouse failed: %s", exc) return web.json_response({"stages": [], "verdict": None, "error": "unavailable"}) return web.json_response(data) # v18.2 — modules/http_guard.py. The server now shares an event loop # with the scan loop AND faces the public internet, which makes an # enthusiastic scraper indistinguishable from an attack in its effect: # scan cadence stretches and the bot stops trading. The middleware caps # concurrency, buckets per client, and locks out repeated failed # authentication against /command. It fails OPEN on internal error — # a broken limiter must not take down the page the operator diagnoses # everything else from. middlewares = [] try: from modules.http_guard import make_middleware middlewares.append(make_middleware()) except Exception as exc: # noqa: BLE001 logger.warning("[BotAPI] HTTP guard unavailable, serving unprotected: %s", exc) app = web.Application(middlewares=middlewares) app.router.add_get("/", handle_index) app.router.add_get("/api/pnl", handle_api_pnl) app.router.add_get("/api/receipts", handle_api_receipts) app.router.add_get("/api/routes", handle_api_routes) app.router.add_get("/api/nearmiss", handle_api_nearmiss) app.router.add_get("/api/engine", handle_api_engine) app.router.add_get("/api/lighthouse", handle_api_lighthouse) app.router.add_get("/api/observatory", handle_api_observatory) app.router.add_get("/receipts.csv", handle_receipts_csv) app.router.add_get("/status", handle_status) app.router.add_post("/command", handle_command) app.router.add_get("/logs", handle_logs) host, port = _bot_api_host(), _bot_api_port() runner = web.AppRunner(app) await runner.setup() site = web.TCPSite(runner, host, port) await site.start() logger.info("🌐 Bot API listening on %s:%d (/status, /command)", host, port) try: while getattr(bot, "_running", True): await asyncio.sleep(1) finally: await runner.cleanup()