""" QuantForge Telegram Notifier ──────────────────────────── Sends auth-expiry alerts, listens for cookie replies, and fires submission notifications. Pure httpx — no extra dependencies. """ import asyncio import httpx import logging import time from typing import Optional logger = logging.getLogger("TelegramBot") _COOKIE_HINTS = ("t=", "session=", "csrftoken=", "sessionid=") def _looks_like_cookie(text: str) -> bool: """Heuristic: does this text look like a WQ session cookie string?""" if not text or len(text) < 15: return False # Named cookie (t=xxx; session=xxx …) if any(hint in text for hint in _COOKIE_HINTS): return True # Raw long token — no spaces, mostly alnum + - _ . = stripped = text.replace("-", "").replace("_", "").replace(".", "").replace("=", "").replace(";", "") if len(text) > 30 and " " not in text and stripped.isalnum(): return True return False class TelegramNotifier: """ Async Telegram bot client using HTTPX. """ def __init__(self, token: str, chat_id: str, proxy_url: Optional[str] = None) -> None: self.token = token.strip() self.chat_id = str(chat_id).strip() self.proxy_url = proxy_url.strip() if (proxy_url and proxy_url.strip()) else None base_domain = self.proxy_url if self.proxy_url else "https://api.telegram.org" if base_domain.endswith("/"): base_domain = base_domain[:-1] self._base = f"{base_domain}/bot{self.token}" self._client: Optional[httpx.AsyncClient] = None self._last_update_id: int = 0 # Biometric state variables self._auth_active = False self._biometric_url = None self._biometric_client = None self._biometric_cancelled = False self._successful_cookie = None self._polling_failed = False async def _get_client(self) -> httpx.AsyncClient: if self._client is None or self._client.is_closed: # Bypass SSL certificate checks for api.telegram.org in container sandbox self._client = httpx.AsyncClient( verify=False, timeout=httpx.Timeout(35.0) ) return self._client async def close(self) -> None: if self._client and not self._client.is_closed: await self._client.aclose() self._cleanup_biometric() async def send(self, text: str, parse_mode: str = "HTML", reply_markup: Optional[dict] = None) -> bool: """Send a message to the configured chat. Returns True on success.""" try: client = await self._get_client() payload = {"chat_id": self.chat_id, "text": text, "parse_mode": parse_mode} if reply_markup is not None: payload["reply_markup"] = reply_markup resp = await client.post(f"{self._base}/sendMessage", json=payload) if resp.status_code == 200: return True body = resp.text logger.error(f"Telegram sendMessage failed {resp.status_code}: {body[:200]}") return False except Exception as exc: import traceback logger.error(f"Telegram send error: {exc}\n{traceback.format_exc()}") return False async def _answer_callback_query(self, callback_query_id: str) -> None: try: client = await self._get_client() payload = {"callback_query_id": callback_query_id} await client.post(f"{self._base}/answerCallbackQuery", json=payload) except Exception as e: logger.warning(f"Failed to answer callback query: {e}") async def send_auth_expired_alert(self) -> None: msg = ( "🔴 QuantForge — Session Expired\n\n" "The WorldQuant BRAIN session is paused. Choose an option to authenticate:\n\n" "1️⃣ Biometric Flow: Click Authenticate Now below (or reply OK) to get a face scan link.\n" "2️⃣ One-Tap Sync: Use your configured bookmarklet on mobile or Chrome extension on desktop.\n" "3️⃣ Manual Pasting: Reply directly with your fresh cookie string (starting with t=)." ) reply_markup = { "inline_keyboard": [ [ {"text": "🔐 Authenticate Now", "callback_data": "auth_now"} ] ] } await self.send(msg, reply_markup=reply_markup) logger.info("Telegram: Auth-expired alert sent.") async def send_session_restored(self) -> None: await self.send( "✅ QuantForge — Session Restored\n\n" "New cookie accepted. Miner is resuming operations. 🚀" ) logger.info("Telegram: Session-restored confirmation sent.") async def send_auth_failed_permanently(self) -> None: await self.send( "⚠️ QuantForge — Auth Recovery Timed Out\n\n" "No cookie was received within 30 minutes.\n" "The miner will keep retrying every 10 minutes.\n" "Please send your new cookie whenever you're ready." ) async def send_submission_alert( self, alpha_id: str, sharpe: float, fitness: float, region: str = "USA" ) -> None: msg = ( f"🎯 Alpha Submitted to BRAIN!\n\n" f"• Region: {region}\n" f"• Sharpe: {sharpe:.4f}\n" f"• Fitness: {fitness:.4f}\n\n" "Check your WorldQuant BRAIN dashboard 💰" ) await self.send(msg) async def send_startup(self) -> None: await self.send( "🟢 QuantForge Miner — Started\n\n" "System is live and authenticated. Mining alphas... 🔬" ) async def _drain_pending_updates(self) -> None: """Advance offset past all currently queued messages so we only listen for messages sent *after* this call.""" try: client = await self._get_client() resp = await client.get( f"{self._base}/getUpdates", params={"offset": -1, "limit": 1, "timeout": 1}, timeout=5.0 ) if resp.status_code == 200: data = resp.json() updates = data.get("result", []) if updates: self._last_update_id = updates[-1]["update_id"] except Exception: pass def _cleanup_biometric(self) -> None: self._auth_active = False self._biometric_url = None self._biometric_cancelled = False self._successful_cookie = None self._polling_failed = False if self._biometric_client: client = self._biometric_client self._biometric_client = None try: asyncio.create_task(client.aclose()) except RuntimeError: pass def _cancel_biometric(self) -> None: self._biometric_cancelled = True if hasattr(self, "_polling_task") and self._polling_task and not self._polling_task.done(): self._polling_task.cancel() self._cleanup_biometric() async def _run_biometric_polling(self) -> None: import os import base64 from config import WQ_MFA_POLL_INTERVAL, WQ_MFA_TIMEOUT poll_interval = WQ_MFA_POLL_INTERVAL timeout = WQ_MFA_TIMEOUT username = os.environ.get("WQ_USERNAME", "").strip() password = os.environ.get("WQ_PASSWORD", "").strip() credentials = f"{username}:{password}" encoded_credentials = base64.b64encode(credentials.encode()).decode() auth_headers = { "Authorization": f"Basic {encoded_credentials}", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" } # The inquiry URL is the ORIGINAL /authentication/persona?inquiry=inq_xxx # returned in the Location header of the initial 401. Polling this specific # URL (rather than /authentication) checks the status of the original scan. poll_url = getattr(self, "_biometric_url", None) logger.info( f"Telegram Bot: Starting biometric polling loop " f"(interval={poll_interval}s, timeout={timeout}s, " f"username={'SET' if username else 'MISSING'}, " f"poll_url={poll_url})" ) if not poll_url: logger.error("Telegram Bot: No biometric URL stored — cannot poll. Aborting.") self._polling_failed = True await self.send("⚠️ Internal error: biometric URL not set. Please try again.") return # Log initial cookie jar state initial_cookies = dict(self._biometric_client.cookies) logger.info(f"Telegram Bot: Initial cookie jar keys: {list(initial_cookies.keys())}") deadline = time.monotonic() + timeout poll_count = 0 try: while time.monotonic() < deadline: if self._biometric_cancelled: logger.info("Telegram Bot: Biometric polling cancelled by user request.") return await asyncio.sleep(poll_interval) if self._biometric_cancelled: return poll_count += 1 try: # Log exactly what we're sending current_cookies = dict(self._biometric_client.cookies) logger.info( f"Telegram Bot: Poll #{poll_count} — POST {poll_url} " f"| Cookie jar keys: {list(current_cookies.keys())} " f"| Auth header present: {'Authorization' in auth_headers}" ) # POST to the original inquiry URL — WQ returns 201 once the Persona scan is complete resp = await self._biometric_client.post( poll_url, headers=auth_headers ) # Log the complete response for diagnosis resp_body = "" try: resp_body = resp.text[:500] except Exception: resp_body = "" resp_headers_log = dict(resp.headers) if hasattr(resp, "headers") else {} logger.info( f"Telegram Bot: Poll #{poll_count} response — " f"Status: {resp.status_code} | " f"Body: {resp_body!r} | " f"Location: {resp_headers_log.get('location', resp_headers_log.get('Location', 'none'))} | " f"WWW-Auth: {resp_headers_log.get('www-authenticate', resp_headers_log.get('WWW-Authenticate', 'none'))}" ) # Log updated cookie jar after each poll after_cookies = dict(self._biometric_client.cookies) logger.info(f"Telegram Bot: Poll #{poll_count} cookie jar after request: {list(after_cookies.keys())}") if resp.status_code == 201: logger.info("Telegram Bot: Biometric authentication successful! Got 201.") cookies_dict = {} for k, v in self._biometric_client.cookies.items(): cookies_dict[k] = v logger.info(f"Telegram Bot: Captured {len(cookies_dict)} cookies on success: {list(cookies_dict.keys())}") cookie_parts = [f"{k}={v}" for k, v in cookies_dict.items()] cookie_string = "; ".join(cookie_parts) if cookie_string: self._successful_cookie = cookie_string await self.send("✅ Biometric verification successful! Session restored. Resuming miner loop.") return else: logger.error("Telegram Bot: Got 201 but no session cookies found in jar.") self._polling_failed = True await self.send("⚠️ Biometric succeeded but no session cookies were returned.") return # Check if the scan is pending: # 202 = accepted/pending, 401 = still awaiting scan, # 403 with INQUIRY_INCOMPLETE body = user has not finished scan yet. is_pending = False if resp.status_code in (202, 401): is_pending = True elif resp.status_code == 403 and "INQUIRY_INCOMPLETE" in resp_body: is_pending = True if is_pending: pending_location = resp_headers_log.get("location", resp_headers_log.get("Location", "none")) logger.info( f"Telegram Bot: Poll #{poll_count} — Scan still pending ({resp.status_code}). " f"Location: {pending_location}. Waiting..." ) continue elif resp.status_code == 429: # Throttled mid-polling — wait for Retry-After, then continue retry_after = int(resp_headers_log.get("retry-after", resp_headers_log.get("Retry-After", 30))) logger.warning( f"Telegram Bot: Poll #{poll_count} — Throttled (429). " f"Waiting {retry_after}s before continuing..." ) await asyncio.sleep(retry_after) continue else: logger.error( f"Telegram Bot: Poll #{poll_count} — Unexpected status {resp.status_code}. " f"Full body: {resp_body!r}" ) self._polling_failed = True await self.send(f"❌ Biometric verification failed. (Status {resp.status_code})\n
{resp_body[:200]}
") return except Exception as poll_err: logger.warning(f"Telegram Bot: Poll #{poll_count} exception: {type(poll_err).__name__}: {poll_err}", exc_info=True) logger.warning(f"Telegram Bot: Biometric polling timed out after {poll_count} polls.") self._polling_failed = True await self.send("⏰ Biometric authentication timed out. Please try again.") except Exception as e: logger.error(f"Telegram Bot: Exception in polling task: {type(e).__name__}: {e}", exc_info=True) self._polling_failed = True async def _initiate_biometric_auth(self) -> None: import os import base64 import urllib.parse username = os.environ.get("WQ_USERNAME", "").strip() password = os.environ.get("WQ_PASSWORD", "").strip() if not username or not password: await self.send("⚠️ Credentials Missing: WQ_USERNAME and WQ_PASSWORD must be configured in your environment to authenticate.") return await self.send("⏳ Initiating login to generate verification link...") self._cleanup_biometric() credentials = f"{username}:{password}" encoded_credentials = base64.b64encode(credentials.encode()).decode() default_headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" } self._biometric_client = httpx.AsyncClient( verify=False, timeout=httpx.Timeout(10.0), headers=default_headers ) auth_headers = { "Authorization": f"Basic {encoded_credentials}" } try: resp = await self._biometric_client.post( "https://api.worldquantbrain.com/authentication", headers=auth_headers ) # Handle rate limiting — WQ throttles biometric attempts across sessions if resp.status_code == 429: retry_after = int(resp.headers.get("Retry-After", 60)) body_text = "" try: body_text = resp.text[:200] except Exception: pass logger.warning(f"Telegram Bot: WQ auth throttled (429). Retry-After: {retry_after}s. Body: {body_text!r}") await self.send( f"⏱ WorldQuant is throttling auth requests.\n" f"Too many attempts were made recently. Waiting {retry_after}s before retrying automatically…" ) await asyncio.sleep(retry_after) logger.info("Telegram Bot: Retrying authentication after throttle backoff...") resp = await self._biometric_client.post( "https://api.worldquantbrain.com/authentication", headers=auth_headers ) logger.info(f"Telegram Bot: Retry response status: {resp.status_code}") if resp.status_code == 429: logger.error("Telegram Bot: Still throttled after backoff. Aborting.") await self.send("❌ Still throttled by WorldQuant. Please wait a few minutes and try again.") return if resp.status_code == 201: logger.info("Telegram Bot: Basic Auth succeeded directly without biometrics.") cookie_parts = [] for k, v in self._biometric_client.cookies.items(): cookie_parts.append(f"{k}={v}") cookie_string = "; ".join(cookie_parts) if cookie_string: self._successful_cookie = cookie_string await self.send("✅ Login succeeded immediately! No biometric scan was required.") else: await self.send("⚠️ Login succeeded but no cookies were returned.") return elif resp.status_code == 401: www_auth = resp.headers.get("WWW-Authenticate") location = resp.headers.get("Location") if www_auth == "persona" and location: logger.info("Telegram Bot: Biometric challenge detected. Location: %s", location) biometric_url = urllib.parse.urljoin("https://api.worldquantbrain.com/authentication", location) self._biometric_url = biometric_url # Capture and parse challenge cookies self._biometric_cookies = "" set_cookies = [] if hasattr(resp.headers, "getlist"): set_cookies = resp.headers.getlist("set-cookie") elif hasattr(resp.headers, "get"): cookie_val = resp.headers.get("set-cookie") if cookie_val: set_cookies = [cookie_val] if isinstance(cookie_val, str) else cookie_val cookies_dict = {} for cookie_str in set_cookies: parts = cookie_str.split(";")[0].split("=") if len(parts) == 2: cookies_dict[parts[0].strip()] = parts[1].strip() # Also include client jar cookies for k, v in self._biometric_client.cookies.items(): cookies_dict[k] = v if cookies_dict: self._biometric_cookies = "; ".join([f"{k}={v}" for k, v in cookies_dict.items()]) logger.info(f"Telegram Bot: Captured challenge cookies: {self._biometric_cookies}") self._auth_active = True self._biometric_cancelled = False self._polling_failed = False self._successful_cookie = None self._polling_task = asyncio.create_task(self._run_biometric_polling()) reply_markup = { "inline_keyboard": [ [ {"text": "❌ Cancel", "callback_data": "cancel_auth"} ] ] } await self.send( f"🔗 Biometric Verification Required:\n\n" f"Click here to start your Face Scan\n\n" f"⏳ Once completed, the session will restore automatically.\n" f"(You can click Cancel to abort)", reply_markup=reply_markup ) else: logger.warning("Telegram Bot: 401 response without persona challenge: %s", resp.text) await self.send("❌ Authentication failed: Incorrect WorldQuant username or password.") else: resp_body = "" try: resp_body = resp.text[:200] except Exception: pass logger.error(f"Telegram Bot: WQ auth returned status {resp.status_code}. Body: {resp_body!r}") await self.send(f"❌ WorldQuant Auth Error: Received status code {resp.status_code}\n
{resp_body}
") except Exception as e: logger.exception(f"Telegram Bot: Exception during authentication initiation: {e}") await self.send(f"❌ Connection Error: Failed to contact WorldQuant Brain: {e}") async def poll_for_cookie(self, timeout_seconds: int = 1800, wake_event: asyncio.Event = None) -> Optional[str]: """ Long-polls Telegram for a message that looks like a WQ session cookie. Also wakes up immediately if wake_event is set (by /login-cookie HTTP endpoint). Returns the cookie string on success, None on timeout. Default timeout: 30 minutes. """ import os logger.info("Telegram: Listening for cookie reply (30-minute window)…") await self._drain_pending_updates() client = await self._get_client() deadline = time.monotonic() + timeout_seconds warned = False while time.monotonic() < deadline: remaining = deadline - time.monotonic() if remaining <= 0: break # --- Check if /login-cookie HTTP endpoint already delivered a cookie --- if wake_event and wake_event.is_set(): wake_event.clear() fresh_cookie = os.environ.get("WQ_SESSION_COOKIE", "").strip() if fresh_cookie: logger.info("Telegram: Wake event fired by /login-cookie endpoint. Using HTTP-delivered cookie.") self._cancel_biometric() return fresh_cookie # --- Check if biometric polling task completed successfully --- if self._successful_cookie: cookie = self._successful_cookie self._cleanup_biometric() return cookie # Warn at 10-minute mark if remaining < 600 and not warned: warned = True await self.send( "⏰ Reminder: 10 minutes remaining to send your new cookie." ) try: poll_timeout_param = min(20, int(remaining)) if self._auth_active: poll_timeout_param = min(3, poll_timeout_param) import json params = { "offset": self._last_update_id + 1, "timeout": poll_timeout_param, "allowed_updates": json.dumps(["message", "callback_query"]), } poll_timeout = float(poll_timeout_param + 5) resp = await client.get( f"{self._base}/getUpdates", params=params, timeout=poll_timeout ) if resp.status_code != 200: await asyncio.sleep(5) continue data = resp.json() for update in data.get("result", []): self._last_update_id = max( self._last_update_id, update["update_id"] ) # 1. Handle callback query (button clicks) callback_query = update.get("callback_query", {}) if callback_query: chat_id = str(callback_query.get("message", {}).get("chat", {}).get("id", "")) if chat_id != self.chat_id: continue query_id = callback_query.get("id", "") await self._answer_callback_query(query_id) data_payload = callback_query.get("data", "") if data_payload == "auth_now": if self._auth_active: await self.send("⚠️ Auth in progress: An active biometric login session is already running.") else: asyncio.create_task(self._initiate_biometric_auth()) elif data_payload == "cancel_auth": if self._auth_active: self._cancel_biometric() await self.send("❌ Authentication cancelled.") else: await self.send("⚠️ No active authentication session to cancel.") continue # 2. Handle message msg = update.get("message", {}) if str(msg.get("chat", {}).get("id", "")) != self.chat_id: continue text = (msg.get("text") or "").strip() if not text: continue # Check if user typed "/cancel" if text == "/cancel": if self._auth_active: self._cancel_biometric() await self.send("❌ Authentication cancelled.") else: await self.send("⚠️ No active authentication session to cancel.") continue # Check if user typed "/force-cookie" if text == "/force-cookie": self._cancel_biometric() await self.send( "❌ Biometric auth aborted.\n\n" "Please paste your fresh cookie starting with t= directly here." ) continue # Check if user replied with "OK" or "/auth" if text.upper() in ("OK", "/AUTH"): if self._auth_active: await self.send("⚠️ Auth in progress: An active biometric login session is already running.") else: asyncio.create_task(self._initiate_biometric_auth()) continue # Check if text is a pasted session cookie if _looks_like_cookie(text): logger.info("Telegram: Valid cookie reply received. Cancelling any active biometric poll.") self._cancel_biometric() return text # User sent something else — tell them what we need await self.send( "🤔 That doesn't look like a session cookie.\n" "Please paste the t= value, click Authenticate Now, or reply OK." ) except httpx.TimeoutException: continue except Exception as exc: logger.warning(f"Telegram poll error: {exc}") await asyncio.sleep(10) # Timed out self._cancel_biometric() await self.send_auth_failed_permanently() logger.warning("Telegram: Cookie poll timed out.") return None