"""RunPod access. Two transports, deliberately unequal: * **API key (default).** One header against the documented GraphQL endpoint. Nothing to expire, nothing to refresh, works from a container. * **Console login (fallback).** The old flow: sign in to Clerk with an email and password, then reuse the session JWT. Kept because it is the only path that works if the API key is revoked, but it cannot run unattended. The "No active session found" failure in the old script was not a UI change: the account now has two-factor authentication on. Clerk verifies the password and then answers ``status: needs_second_factor`` with an emailed code, so no session is ever created. That cannot be automated without reading the mailbox. If the account is switched from email codes to an authenticator app, set CX_RUNPOD_TOTP_SECRET and this fallback can complete on its own; otherwise CX_RUNPOD_2FA_CODE accepts a code for a single manual run. Set CX_RUNPOD_API_KEY and this module never touches the console at all. """ from __future__ import annotations import os import threading import time from typing import Any, Optional import httpx GRAPHQL_URL = os.environ.get("CX_RUNPOD_GRAPHQL_URL", "https://api.runpod.io/graphql") CLERK_BASE = os.environ.get("CX_RUNPOD_CLERK_BASE", "https://clerk.runpod.io") CONSOLE_URL = "https://console.runpod.io" CLERK_QS = "__clerk_api_version=2025-11-10&_clerk_js_version=5.125.12" class RunPodError(RuntimeError): pass # --- queries ---------------------------------------------------------------- MACHINES_QUERY = """ query getMachinesForHostDashboard { myself { machineQuota machines { id name listed registered verified gpuTypeId gpuReserved gpuTotal dataCenterId machineType hostPricePerGpu margin uptimePercentListedOneWeek uptimePercentListedFourWeek maintenanceStart maintenanceEnd gpuType { displayName manufacturer } machineSystem { os cudaVersion kernelVersion } } } } """ SUMMARY_QUERY = """ query getMyMachines { myself { machinesSummary { id displayName listed machineType gpuTypeId gpuRented gpuTotal podProfitPerHr diskProfitPerHr onDemandPods spotPods } } } """ LIST_MUTATION = """ mutation listMachineBulk($input: MachineListBulkInput) { machineListBulk(input: $input) } """ UNLIST_MUTATION = """ mutation unlistMachineBulk($input: MachineUnlistBulkInput) { machineUnlistBulk(input: $input) } """ MAINTENANCE_MUTATION = """ mutation machineScheduleMaintenance($input: MachineScheduleMaintenanceInput) { machineScheduleMaintenance(input: $input) } """ class RunPodClient: def __init__(self, api_key: str = "", email: str = "", password: str = "", team_id: str = "", totp_secret: str = "", otp_code: str = "", timeout: int = 30): self.api_key = (api_key or "").strip() self.email = (email or "").strip() self.password = password or "" self.team_id = (team_id or "").strip() self.totp_secret = (totp_secret or "").strip() self.otp_code = (otp_code or "").strip() self.timeout = timeout self._jwt: Optional[str] = None self._jwt_at = 0.0 self._lock = threading.Lock() # --- auth --------------------------------------------------------------- @property def mode(self) -> str: if self.api_key: return "api_key" if self.email and self.password: return "console_login" return "unconfigured" def _auth_header(self) -> dict[str, str]: if self.api_key: return {"Authorization": f"Bearer {self.api_key}"} jwt = self._console_jwt() header = {"Authorization": f"Bearer {jwt}"} if self.team_id: header["x-team-id"] = self.team_id return header def _console_jwt(self) -> str: """Sign in to the console and return a session JWT. Only used when no API key is set. Cached for 50 minutes; Clerk tokens last an hour. """ with self._lock: if self._jwt and time.time() - self._jwt_at < 3000: return self._jwt if not (self.email and self.password): raise RunPodError( "RunPod is not configured. Set CX_RUNPOD_API_KEY, or " "CX_RUNPOD_EMAIL and CX_RUNPOD_PASSWORD for the console fallback." ) with httpx.Client(timeout=self.timeout, follow_redirects=True, headers={ "User-Agent": "Mozilla/5.0", "Origin": CONSOLE_URL, "Referer": f"{CONSOLE_URL}/", }) as client: client.get(CONSOLE_URL) signin = client.post( f"{CLERK_BASE}/v1/client/sign_ins?{CLERK_QS}", data={"identifier": self.email, "password": self.password, "strategy": "password"}, ) if signin.status_code >= 400: raise RunPodError(f"Console sign-in rejected ({signin.status_code}): {signin.text[:200]}") body = signin.json().get("response") or {} session_id = body.get("created_session_id") if body.get("status") == "needs_second_factor": session_id = self._second_factor(client, body) # The old script read the session out of this response and gave # up when it was absent. Ask for the client record instead. jwt, found_id = self._session_from_client(client, session_id) if not jwt and found_id: jwt = self._mint_token(client, found_id) if not jwt: raise RunPodError( "Console sign-in completed but no session token came back. RunPod's console auth has " "changed again - use CX_RUNPOD_API_KEY instead." ) self._jwt, self._jwt_at = jwt, time.time() return jwt def _second_factor(self, client: httpx.Client, body: dict[str, Any]) -> Optional[str]: """Satisfy Clerk's second factor, if we have been given the means to. The account currently uses emailed codes, which no unattended process can read - that is the real reason the old login stopped working. """ sign_in_id = body.get("id") offered = [f.get("strategy") for f in (body.get("supported_second_factors") or [])] code = "" if "totp" in offered and self.totp_secret: code, strategy = self._totp_code(), "totp" elif self.otp_code: code, strategy = self.otp_code, ("email_code" if "email_code" in offered else offered[0] if offered else "") if "email_code" in offered: client.post(f"{CLERK_BASE}/v1/client/sign_ins/{sign_in_id}/prepare_second_factor?{CLERK_QS}", data={"strategy": "email_code"}) if not code: raise RunPodError( "RunPod console login needs a second factor " f"({', '.join(offered) or 'unknown strategy'}) and no code is available. This is why the old " "script failed - the password is accepted, but the account has 2FA on. Use CX_RUNPOD_API_KEY for " "unattended runs, or set CX_RUNPOD_TOTP_SECRET if you move the account to an authenticator app." ) attempt = client.post( f"{CLERK_BASE}/v1/client/sign_ins/{sign_in_id}/attempt_second_factor?{CLERK_QS}", data={"strategy": strategy, "code": code}, ) if attempt.status_code >= 400: raise RunPodError(f"Second factor rejected ({attempt.status_code}): {attempt.text[:200]}") return ((attempt.json().get("response") or {}).get("created_session_id")) def _totp_code(self) -> str: """RFC 6238 code from a base32 secret - no third-party dependency.""" import base64 import hashlib import hmac import struct secret = self.totp_secret.replace(" ", "").upper() secret += "=" * (-len(secret) % 8) key = base64.b32decode(secret, casefold=True) counter = struct.pack(">Q", int(time.time()) // 30) digest = hmac.new(key, counter, hashlib.sha1).digest() offset = digest[-1] & 0x0F value = struct.unpack(">I", digest[offset:offset + 4])[0] & 0x7FFFFFFF return f"{value % 1_000_000:06d}" def _session_from_client(self, client: httpx.Client, prefer_id: Optional[str]) -> tuple[str, str]: resp = client.get(f"{CLERK_BASE}/v1/client?{CLERK_QS}") if resp.status_code >= 400: return "", "" sessions = ((resp.json().get("response") or {}).get("sessions") or []) if not sessions: return "", "" chosen = next((s for s in sessions if s.get("id") == prefer_id), sessions[0]) token = ((chosen.get("last_active_token") or {}).get("jwt")) or "" return token, str(chosen.get("id") or "") def _mint_token(self, client: httpx.Client, session_id: str) -> str: resp = client.post(f"{CLERK_BASE}/v1/client/sessions/{session_id}/tokens?{CLERK_QS}") if resp.status_code >= 400: return "" return resp.json().get("jwt", "") # --- transport ---------------------------------------------------------- def execute(self, query: str, variables: Optional[dict[str, Any]] = None) -> dict[str, Any]: headers = {"Content-Type": "application/json", **self._auth_header()} payload: dict[str, Any] = {"query": query} if variables is not None: payload["variables"] = variables with httpx.Client(timeout=self.timeout) as client: resp = client.post(GRAPHQL_URL, json=payload, headers=headers) if resp.status_code == 401: raise RunPodError("RunPod rejected the credentials (401). Check CX_RUNPOD_API_KEY.") if resp.status_code >= 400: raise RunPodError(f"RunPod returned {resp.status_code}: {resp.text[:300]}") body = resp.json() if body.get("errors"): raise RunPodError("; ".join(e.get("message", "?") for e in body["errors"])[:400]) data = body.get("data") if data is None: raise RunPodError(f"RunPod returned no data: {str(body)[:200]}") return data # --- operations --------------------------------------------------------- def machines(self) -> list[dict[str, Any]]: data = self.execute(MACHINES_QUERY) return list(((data.get("myself") or {}).get("machines")) or []) def summary(self) -> list[dict[str, Any]]: data = self.execute(SUMMARY_QUERY) return list(((data.get("myself") or {}).get("machinesSummary")) or []) def list_machines(self, machine_ids: list[str]) -> dict[str, Any]: return self.execute(LIST_MUTATION, {"input": {"machineIds": machine_ids}}) def unlist_machines(self, machine_ids: list[str]) -> dict[str, Any]: return self.execute(UNLIST_MUTATION, {"input": {"machineIds": machine_ids}}) def schedule_maintenance(self, machine_ids: list[str], start_utc: str, minutes: int, reason: str = "EMERGENCY", destructive: bool = False) -> dict[str, Any]: if reason not in ("UPGRADE", "ROUTINE", "EMERGENCY", "REMOVE"): raise RunPodError(f"Unknown maintenance reason '{reason}'.") return self.execute(MAINTENANCE_MUTATION, {"input": { "machineIds": machine_ids, "maintenanceStartUtc": start_utc, "maintenanceMinutes": minutes, "maintenanceReason": reason, "destructive": destructive, }}) def check(self) -> dict[str, Any]: """Cheap connectivity probe for the health endpoint.""" try: machines = self.machines() return {"ok": True, "mode": self.mode, "machines": len(machines)} except RunPodError as exc: return {"ok": False, "mode": self.mode, "error": str(exc)}