From a15046a2ff3227cfb8615baa182747a2aab24e3a Mon Sep 17 00:00:00 2001 From: genz27 Date: Sat, 10 Jan 2026 16:06:48 +0800 Subject: [PATCH] feat: implement openai-sentinel-token PoW verification --- src/services/sora_client.py | 192 ++++++++++++++++++++++++++++++++++-- 1 file changed, 182 insertions(+), 10 deletions(-) diff --git a/src/services/sora_client.py b/src/services/sora_client.py index 8122e60..1fb0d05 100644 --- a/src/services/sora_client.py +++ b/src/services/sora_client.py @@ -1,35 +1,207 @@ """Sora API client module""" import base64 +import hashlib +import json import io import time import random import string import re +from datetime import datetime, timedelta, timezone from typing import Optional, Dict, Any, Tuple +from uuid import uuid4 from curl_cffi.requests import AsyncSession from curl_cffi import CurlMime from .proxy_manager import ProxyManager from ..core.config import config from ..core.logger import debug_logger +# PoW related constants +POW_MAX_ITERATION = 500000 +POW_CORES = [8, 16, 24, 32] +POW_SCRIPTS = [ + "https://cdn.oaistatic.com/_next/static/cXh69klOLzS0Gy2joLDRS/_ssgManifest.js?dpl=453ebaec0d44c2decab71692e1bfe39be35a24b3" +] +POW_DPL = ["prod-f501fe933b3edf57aea882da888e1a544df99840"] +POW_NAVIGATOR_KEYS = [ + "registerProtocolHandler−function registerProtocolHandler() { [native code] }", + "storage−[object StorageManager]", + "locks−[object LockManager]", + "appCodeName−Mozilla", + "permissions−[object Permissions]", + "webdriver−false", + "vendor−Google Inc.", + "mediaDevices−[object MediaDevices]", + "cookieEnabled−true", + "product−Gecko", + "productSub−20030107", + "hardwareConcurrency−32", + "onLine−true", +] +POW_DOCUMENT_KEYS = ["_reactListeningo743lnnpvdg", "location"] +POW_WINDOW_KEYS = [ + "0", "window", "self", "document", "name", "location", + "navigator", "screen", "innerWidth", "innerHeight", + "localStorage", "sessionStorage", "crypto", "performance", + "fetch", "setTimeout", "setInterval", "console", +] + class SoraClient: """Sora API client with proxy support""" + CHATGPT_BASE_URL = "https://chatgpt.com" + SENTINEL_FLOW = "sora_2_create_task" + def __init__(self, proxy_manager: ProxyManager): self.proxy_manager = proxy_manager self.base_url = config.sora_base_url self.timeout = config.sora_timeout @staticmethod - def _generate_sentinel_token() -> str: - """ - 生成 openai-sentinel-token - 根据测试文件的逻辑，传入任意随机字符即可 - 生成10-20个字符的随机字符串（字母+数字） - """ - length = random.randint(10, 20) - random_str = ''.join(random.choices(string.ascii_letters + string.digits, k=length)) - return random_str + def _get_pow_parse_time() -> str: + """Generate time string for PoW (EST timezone)""" + now = datetime.now(timezone(timedelta(hours=-5))) + return now.strftime("%a %b %d %Y %H:%M:%S") + " GMT-0500 (Eastern Standard Time)" + + @staticmethod + def _get_pow_config(user_agent: str) -> list: + """Generate PoW config array with browser fingerprint""" + return [ + random.choice([1920 + 1080, 2560 + 1440, 1920 + 1200, 2560 + 1600]), + SoraClient._get_pow_parse_time(), + 4294705152, + 0, # [3] dynamic + user_agent, + random.choice(POW_SCRIPTS) if POW_SCRIPTS else "", + random.choice(POW_DPL) if POW_DPL else None, + "en-US", + "en-US,es-US,en,es", + 0, # [9] dynamic + random.choice(POW_NAVIGATOR_KEYS), + random.choice(POW_DOCUMENT_KEYS), + random.choice(POW_WINDOW_KEYS), + time.perf_counter() * 1000, + str(uuid4()), + "", + random.choice(POW_CORES), + time.time() * 1000 - (time.perf_counter() * 1000), + ] + + @staticmethod + def _solve_pow(seed: str, difficulty: str, config_list: list) -> Tuple[str, bool]: + """Execute PoW calculation using SHA3-512 hash collision""" + diff_len = len(difficulty) // 2 + seed_encoded = seed.encode() + target_diff = bytes.fromhex(difficulty) + + static_part1 = (json.dumps(config_list[:3], separators=(',', ':'), ensure_ascii=False)[:-1] + ',').encode() + static_part2 = (',' + json.dumps(config_list[4:9], separators=(',', ':'), ensure_ascii=False)[1:-1] + ',').encode() + static_part3 = (',' + json.dumps(config_list[10:], separators=(',', ':'), ensure_ascii=False)[1:]).encode() + + for i in range(POW_MAX_ITERATION): + dynamic_i = str(i).encode() + dynamic_j = str(i >> 1).encode() + + final_json = static_part1 + dynamic_i + static_part2 + dynamic_j + static_part3 + b64_encoded = base64.b64encode(final_json) + + hash_value = hashlib.sha3_512(seed_encoded + b64_encoded).digest() + + if hash_value[:diff_len] <= target_diff: + return b64_encoded.decode(), True + + error_token = "wQ8Lk5FbGpA2NcR9dShT6gYjU7VxZ4D" + base64.b64encode(f'"{seed}"'.encode()).decode() + return error_token, False + + @staticmethod + def _get_pow_token(user_agent: str) -> str: + """Generate initial PoW token""" + config_list = SoraClient._get_pow_config(user_agent) + seed = format(random.random()) + difficulty = "0fffff" + solution, _ = SoraClient._solve_pow(seed, difficulty, config_list) + return "gAAAAAC" + solution + + @staticmethod + def _build_sentinel_token( + flow: str, + req_id: str, + pow_token: str, + resp: Dict[str, Any], + user_agent: str, + ) -> str: + """Build openai-sentinel-token from PoW response""" + final_pow_token = pow_token + + # Check if PoW is required + proofofwork = resp.get("proofofwork", {}) + if proofofwork.get("required"): + seed = proofofwork.get("seed", "") + difficulty = proofofwork.get("difficulty", "") + if seed and difficulty: + config_list = SoraClient._get_pow_config(user_agent) + solution, success = SoraClient._solve_pow(seed, difficulty, config_list) + final_pow_token = "gAAAAAB" + solution + if not success: + debug_logger.log_warning("PoW calculation failed, using error token") + + token_payload = { + "p": final_pow_token, + "t": resp.get("turnstile", {}).get("dx", ""), + "c": resp.get("token", ""), + "id": req_id, + "flow": flow, + } + return json.dumps(token_payload, ensure_ascii=False, separators=(",", ":")) + + async def _generate_sentinel_token(self, token: Optional[str] = None) -> str: + """Generate openai-sentinel-token by calling /backend-api/sentinel/req and solving PoW""" + req_id = str(uuid4()) + user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36" + pow_token = self._get_pow_token(user_agent) + + proxy_url = await self.proxy_manager.get_proxy_url() + + # Request sentinel/req endpoint + url = f"{self.CHATGPT_BASE_URL}/backend-api/sentinel/req" + payload = {"p": pow_token, "flow": self.SENTINEL_FLOW, "id": req_id} + headers = { + "Accept": "application/json, text/plain, */*", + "Content-Type": "application/json", + "Origin": "https://sora.chatgpt.com", + "Referer": "https://sora.chatgpt.com/", + "User-Agent": user_agent, + } + if token: + headers["Authorization"] = f"Bearer {token}" + + async with AsyncSession() as session: + kwargs = { + "headers": headers, + "json": payload, + "timeout": 10, + "impersonate": "chrome" + } + if proxy_url: + kwargs["proxy"] = proxy_url + + response = await session.post(url, **kwargs) + + if response.status_code not in [200, 201]: + debug_logger.log_error( + error_message=f"Sentinel request failed: {response.status_code}", + status_code=response.status_code, + response_text=response.text + ) + raise Exception(f"Sentinel request failed: {response.status_code}") + + resp = response.json() + + # Build final sentinel token + sentinel_token = self._build_sentinel_token( + self.SENTINEL_FLOW, req_id, pow_token, resp, user_agent + ) + return sentinel_token @staticmethod def is_storyboard_prompt(prompt: str) -> bool: @@ -117,7 +289,7 @@ class SoraClient: # 只在生成请求时添加 sentinel token if add_sentinel_token: - headers["openai-sentinel-token"] = self._generate_sentinel_token() + headers["openai-sentinel-token"] = await self._generate_sentinel_token(token) if not multipart: headers["Content-Type"] = "application/json"