226 lines
7.2 KiB
Python
226 lines
7.2 KiB
Python
"""
|
||
Sentinel 反爬机制处理器
|
||
|
||
Sentinel 是 OpenAI 用于防止自动化注册的安全机制，包括:
|
||
- Proof of Work (PoW) 挑战
|
||
- 设备指纹验证
|
||
- 行为分析
|
||
|
||
✅ 已集成完整的 Sentinel 解决方案（使用 reference/ 下的代码）
|
||
"""
|
||
|
||
from typing import Optional, Dict, Any
|
||
from utils.logger import logger
|
||
from utils.fingerprint import BrowserFingerprint
|
||
|
||
|
||
class SentinelHandler:
|
||
"""
|
||
Sentinel 反爬机制处理器
|
||
|
||
✅ 已集成用户的 Sentinel 解决方案
|
||
使用 reference/ 目录下的完整实现
|
||
"""
|
||
|
||
# Sentinel API 端点（从开发文档提取）
|
||
SENTINEL_API = "https://chatgpt.com/_next/static/chunks/sentinel.js"
|
||
SENTINEL_TOKEN_ENDPOINT = "https://api.openai.com/sentinel/token"
|
||
|
||
def __init__(self, session):
|
||
"""
|
||
初始化 Sentinel 处理器
|
||
|
||
参数:
|
||
session: OAISession 实例（需要使用其 Cookie 和代理）
|
||
"""
|
||
from core.session import OAISession
|
||
self.session: OAISession = session
|
||
self.oai_did = session.oai_did
|
||
|
||
# 初始化浏览器指纹
|
||
self.fingerprint = BrowserFingerprint(session_id=self.oai_did)
|
||
|
||
# 延迟导入 SentinelSolver（避免循环导入）
|
||
self._solver = None
|
||
|
||
logger.info("SentinelHandler initialized")
|
||
|
||
def _get_solver(self):
|
||
"""延迟初始化 Sentinel 求解器"""
|
||
if self._solver is None:
|
||
try:
|
||
from reference.sentinel_solver import SentinelSolver
|
||
self._solver = SentinelSolver(self.fingerprint)
|
||
logger.debug("SentinelSolver initialized successfully")
|
||
except ImportError as e:
|
||
logger.error(f"Failed to import SentinelSolver: {e}")
|
||
raise ImportError(
|
||
"Sentinel solver not found. Please check reference/ directory."
|
||
) from e
|
||
return self._solver
|
||
|
||
async def get_token(
|
||
self,
|
||
flow: str = "username_password_create",
|
||
**kwargs
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
获取 Sentinel Token
|
||
|
||
✅ 已实现 - 使用 reference/ 下的完整解决方案
|
||
|
||
参数:
|
||
flow: 注册流程类型，常见值:
|
||
- "username_password_create" (注册流程)
|
||
- "username_password_create__auto" (自动注册)
|
||
**kwargs: 其他可能需要的参数
|
||
|
||
返回:
|
||
Sentinel Token 字典 {"p": "...", "t": "...", "c": "...", "id": "...", "flow": "..."}
|
||
|
||
实现说明:
|
||
使用 reference/sentinel_solver.py 生成 requirements token
|
||
返回完整的 JSON 对象（用于 Header）
|
||
"""
|
||
logger.info(f"Generating Sentinel token for flow='{flow}'")
|
||
|
||
try:
|
||
# 获取求解器
|
||
solver = self._get_solver()
|
||
|
||
# 生成 requirements token
|
||
token_dict = solver.generate_requirements_token()
|
||
|
||
# 构建完整 token
|
||
# 格式: {"p": "gAAAAAC...", "t": "...", "c": "...", "id": "uuid", "flow": "..."}
|
||
token_dict["flow"] = flow
|
||
token_dict["id"] = self.oai_did
|
||
|
||
# 验证必需字段
|
||
if "p" not in token_dict or not token_dict["p"]:
|
||
raise ValueError("Generated token missing 'p' field")
|
||
|
||
logger.success(f"Sentinel token generated: {str(token_dict)[:50]}...")
|
||
return token_dict
|
||
|
||
except ImportError as e:
|
||
logger.error(f"Sentinel solver not available: {e}")
|
||
raise NotImplementedError(
|
||
"❌ Sentinel solver import failed.\n\n"
|
||
"Please ensure:\n"
|
||
"1. reference/ directory exists with sentinel_solver.py\n"
|
||
"2. sdk/sdk.js file exists\n"
|
||
"3. Node.js is installed and available in PATH\n\n"
|
||
f"Error: {e}"
|
||
) from e
|
||
|
||
except Exception as e:
|
||
logger.exception(f"Failed to generate Sentinel token: {e}")
|
||
raise RuntimeError(f"Sentinel token generation failed: {e}") from e
|
||
|
||
async def solve_enforcement(
|
||
self,
|
||
enforcement_config: Dict[str, Any]
|
||
) -> str:
|
||
"""
|
||
解决完整的 enforcement 挑战（PoW + Turnstile）
|
||
|
||
✅ 已实现 - 使用 reference/sentinel_solver.py
|
||
|
||
参数:
|
||
enforcement_config: 服务器返回的挑战配置
|
||
{
|
||
'proofofwork': {
|
||
'seed': '...',
|
||
'difficulty': '0003a',
|
||
'token': '...', # cached token
|
||
'turnstile': {
|
||
'dx': '...' # VM bytecode
|
||
}
|
||
}
|
||
}
|
||
|
||
返回:
|
||
完整的 Sentinel token (JSON string)
|
||
"""
|
||
logger.info("Solving enforcement challenge...")
|
||
|
||
try:
|
||
solver = self._get_solver()
|
||
token_json = solver.solve_enforcement(enforcement_config)
|
||
|
||
logger.success(f"Enforcement solved: {token_json[:50]}...")
|
||
return token_json
|
||
|
||
except Exception as e:
|
||
logger.exception(f"Failed to solve enforcement: {e}")
|
||
raise RuntimeError(f"Enforcement solving failed: {e}") from e
|
||
|
||
def _build_payload(self, flow: str) -> Dict[str, Any]:
|
||
"""
|
||
构建 Sentinel 请求 Payload
|
||
|
||
参考开发文档中的请求格式:
|
||
{
|
||
"p": "gAAAAAB...", # Proof of Work 答案
|
||
"id": "a1b2c3d4-...", # oai-did
|
||
"flow": "username_password_create__auto"
|
||
}
|
||
|
||
参数:
|
||
flow: 注册流程类型
|
||
|
||
返回:
|
||
Payload 字典
|
||
"""
|
||
return {
|
||
"p": "gAAAAAB_PLACEHOLDER_POW_ANSWER",
|
||
"id": self.oai_did,
|
||
"flow": flow
|
||
}
|
||
|
||
async def verify_token(self, token: str) -> bool:
|
||
"""
|
||
验证 Sentinel Token 是否有效（可选功能）
|
||
|
||
参数:
|
||
token: 待验证的 Sentinel Token
|
||
|
||
返回:
|
||
True 如果有效，否则 False
|
||
"""
|
||
if not token or token == "placeholder_token":
|
||
logger.warning("Received placeholder token, validation skipped")
|
||
return False
|
||
|
||
# Token 基本格式检查
|
||
if not token.startswith("gAAAAA"):
|
||
logger.warning(f"Invalid token format: {token[:20]}...")
|
||
return False
|
||
|
||
logger.info(f"Token validation: {token[:20]}... (length={len(token)})")
|
||
return True
|
||
|
||
def get_sentinel_script(self) -> Optional[str]:
|
||
"""
|
||
获取 Sentinel JavaScript 代码（可选，用于分析）
|
||
|
||
返回:
|
||
Sentinel JS 代码内容，失败则返回 None
|
||
"""
|
||
try:
|
||
response = self.session.get(self.SENTINEL_API)
|
||
if response.status_code == 200:
|
||
logger.info(f"Sentinel script fetched: {len(response.text)} bytes")
|
||
return response.text
|
||
else:
|
||
logger.error(f"Failed to fetch Sentinel script: {response.status_code}")
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f"Error fetching Sentinel script: {e}")
|
||
return None
|
||
|
||
|
||
# 导出主要接口
|
||
__all__ = ["SentinelHandler"]
|