frist
This commit is contained in:
dela
225
core/sentinel.py
Normal file
225
core/sentinel.py
Normal file
@@ -0,0 +1,225 @@
|
||||
"""
|
||||
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"]
|
||||
Reference in New Issue
Block a user