"""Real JWT generation provider for Free Fire guest accounts. This module intentionally contains the provider-specific HTTP logic so the bot can swap providers later without touching command handlers or storage code. """ from __future__ import annotations import asyncio import os from typing import Any import httpx JWT_PROVIDER_URL = os.getenv( "JWT_PROVIDER_URL", "https://ff-jwt-acc-api.vercel.app/token", ) JWT_API_TIMEOUT = float(os.getenv("JWT_API_TIMEOUT", os.getenv("API_TIMEOUT", "30"))) JWT_API_RETRIES = int(os.getenv("JWT_API_RETRIES", "3")) JWT_API_RETRY_DELAY = float(os.getenv("JWT_API_RETRY_DELAY", "1")) JWT_API_CONCURRENCY = int(os.getenv("JWT_API_CONCURRENCY", "10")) SENSITIVE_RESPONSE_KEYS = {"access_token", "open_id", "password"} def _clean_reason(message: str, max_length: int = 180) -> str: """Keep failure reasons short and safe for logs or Telegram messages.""" text = " ".join(str(message).split()) return text[:max_length] if text else "unknown error" def _debug_payload(payload: dict[str, Any]) -> dict[str, Any]: """Return a sanitized subset for explicit debug mode.""" return { key: value for key, value in payload.items() if key not in SENSITIVE_RESPONSE_KEYS and key != "token" } async def generate_jwt_for_account( uid: str, password: str, server: str, *, client: httpx.AsyncClient | None = None, timeout: float | None = None, retries: int | None = None, retry_delay: float | None = None, debug: bool = False, ) -> dict[str, Any]: """Generate one JWT using the configured HTTP provider. Success: {"uid": uid, "server": server, "jwt": token, "status": "success"} Failure: {"uid": uid, "server": server, "status": "failed", "reason": "..."} The password is only used as a provider query parameter and is never returned from this function. """ safe_uid = str(uid).strip() safe_server = str(server).strip().lower() password_value = str(password) if not safe_uid: return { "uid": safe_uid, "server": safe_server, "status": "failed", "reason": "missing uid", } if not password_value: return { "uid": safe_uid, "server": safe_server, "status": "failed", "reason": "missing password", } request_timeout = JWT_API_TIMEOUT if timeout is None else timeout max_retries = max(0, JWT_API_RETRIES if retries is None else retries) delay = JWT_API_RETRY_DELAY if retry_delay is None else retry_delay owns_client = client is None http_client = client or httpx.AsyncClient(timeout=request_timeout) try: last_reason = "request failed" for attempt in range(max_retries + 1): try: response = await http_client.get( JWT_PROVIDER_URL, params={"uid": safe_uid, "password": password_value}, timeout=request_timeout, ) if response.status_code >= 500 or response.status_code == 429: last_reason = f"provider HTTP {response.status_code}" raise httpx.HTTPStatusError( last_reason, request=response.request, response=response, ) if response.status_code != 200: return { "uid": safe_uid, "server": safe_server, "status": "failed", "reason": f"provider HTTP {response.status_code}", } try: payload = response.json() except ValueError: return { "uid": safe_uid, "server": safe_server, "status": "failed", "reason": "provider returned invalid JSON", } if not isinstance(payload, dict): return { "uid": safe_uid, "server": safe_server, "status": "failed", "reason": "provider returned non-object JSON", } provider_status = payload.get("status") token = payload.get("token") if provider_status != "success" or not isinstance(token, str) or not token.strip(): reason = payload.get("reason") or payload.get("message") result: dict[str, Any] = { "uid": safe_uid, "server": safe_server, "status": "failed", "reason": _clean_reason(reason or "provider did not return a token"), } if debug: result["debug"] = _debug_payload(payload) return result result = { "uid": safe_uid, "server": safe_server, "jwt": token.strip(), "status": "success", } if debug: result["debug"] = _debug_payload(payload) return result except httpx.TimeoutException: last_reason = "provider timeout" if attempt >= max_retries: break await asyncio.sleep(delay * (attempt + 1)) except httpx.NetworkError: last_reason = "provider network error" if attempt >= max_retries: break await asyncio.sleep(delay * (attempt + 1)) except httpx.HTTPStatusError as exc: last_reason = f"provider HTTP {exc.response.status_code}" if attempt >= max_retries: break await asyncio.sleep(delay * (attempt + 1)) except httpx.HTTPError: return { "uid": safe_uid, "server": safe_server, "status": "failed", "reason": "provider request error", } return { "uid": safe_uid, "server": safe_server, "status": "failed", "reason": _clean_reason(last_reason), } finally: if owns_client: await http_client.aclose() def jwt_output_entry(result: dict[str, Any]) -> dict[str, str]: """Convert a successful result to the generated JWT JSON output format.""" return { "token": str(result["jwt"]), } async def generate_jwts_for_accounts( accounts: list[dict[str, Any]], server: str, *, concurrency: int | None = None, debug: bool = False, ) -> tuple[list[dict[str, str]], list[dict[str, Any]]]: """Generate JWTs for many accounts. Returns: (successful_output_entries, failed_results) The successful entries match the generated output JSON format and contain only the token value. """ limit = max(1, concurrency or JWT_API_CONCURRENCY) semaphore = asyncio.Semaphore(limit) async with httpx.AsyncClient(timeout=JWT_API_TIMEOUT) as client: async def generate(account: dict[str, Any]) -> dict[str, Any]: async with semaphore: return await generate_jwt_for_account( uid=str(account.get("uid", "")), password=str(account.get("password", "")), server=server, client=client, debug=debug, ) results = await asyncio.gather(*(generate(account) for account in accounts)) successes = [jwt_output_entry(item) for item in results if item.get("status") == "success"] failures = [ { "uid": item.get("uid", ""), "server": item.get("server", server), "status": "failed", "reason": item.get("reason", "unknown error"), } for item in results if item.get("status") != "success" ] return successes, failures