243 lines
8.3 KiB
Python
243 lines
8.3 KiB
Python
# modules/registrar.py (更新版)
|
||
"""OpenAI 注册流程控制器"""
|
||
|
||
import json
|
||
from typing import Dict, Optional
|
||
import secrets
|
||
|
||
from .fingerprint import BrowserFingerprint
|
||
from .sentinel_solver import SentinelSolver
|
||
from .http_client import HTTPClient
|
||
from config import AUTH_BASE_URL, DEBUG
|
||
|
||
|
||
class OpenAIRegistrar:
|
||
"""完整的 OpenAI 注册流程"""
|
||
|
||
def __init__(self, session_id: Optional[str] = None):
|
||
self.fingerprint = BrowserFingerprint(session_id)
|
||
self.solver = SentinelSolver(self.fingerprint)
|
||
self.http_client = HTTPClient(self.fingerprint)
|
||
|
||
def _step1_get_cookies_and_csrf(self):
|
||
"""访问注册页，获取必需的 session cookies"""
|
||
if DEBUG:
|
||
print("\n=== STEP 1: Getting session cookies ===")
|
||
|
||
# 访问注册页（会设置 login_session 等 cookies）
|
||
url = f"{AUTH_BASE_URL}/create-account/password"
|
||
|
||
headers = self.http_client.fingerprint.get_headers()
|
||
headers['Accept'] = 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8'
|
||
|
||
resp = self.http_client.get(url, headers=headers)
|
||
|
||
if DEBUG:
|
||
print(f"Cookies received: {list(self.http_client.cookies.keys())}")
|
||
|
||
# 检查关键 cookies
|
||
required_cookies = ['oai-did', 'login_session', 'oai-client-auth-session']
|
||
missing = [c for c in required_cookies if c not in self.http_client.cookies]
|
||
|
||
if missing:
|
||
print(f"⚠ Warning: Missing cookies: {missing}")
|
||
else:
|
||
print(f"✓ All required cookies present")
|
||
|
||
def _step2_init_sentinel(self):
|
||
"""初始化 Sentinel（生成 token）"""
|
||
if DEBUG:
|
||
print("\n=== STEP 2: Initializing Sentinel ===")
|
||
|
||
try:
|
||
token_data = self.solver.generate_requirements_token()
|
||
self.sentinel_token = json.dumps(token_data)
|
||
|
||
if DEBUG:
|
||
token_str = str(self.sentinel_token)
|
||
print(f"✓ Sentinel token: {token_str[:80]}...")
|
||
|
||
except Exception as e:
|
||
if DEBUG:
|
||
print(f"✗ Sentinel initialization failed: {e}")
|
||
raise
|
||
|
||
def _step3_attempt_register(self, email: str, password: str) -> Dict:
|
||
"""尝试注册（会触发 enforcement）"""
|
||
if DEBUG:
|
||
print("\n=== STEP 3: Attempting registration ===")
|
||
|
||
url = f"{AUTH_BASE_URL}/api/accounts/user/register"
|
||
|
||
# 正确的 payload 格式（根据真实抓包）
|
||
payload = {
|
||
'username': email,
|
||
'password': password
|
||
}
|
||
|
||
# 添加额外的 headers
|
||
headers = self.http_client.fingerprint.get_headers(with_sentinel=self.sentinel_token)
|
||
headers['Accept'] = 'application/json'
|
||
headers['Referer'] = f'{AUTH_BASE_URL}/create-account/password'
|
||
headers['Origin'] = AUTH_BASE_URL
|
||
|
||
# 添加 Datadog tracing headers（模拟真实浏览器）
|
||
headers.update({
|
||
'traceparent': f'00-0000000000000000{secrets.token_hex(8)}-{secrets.token_hex(8)}-01',
|
||
'tracestate': 'dd=s:1;o:rum',
|
||
'x-datadog-origin': 'rum',
|
||
'x-datadog-parent-id': str(secrets.randbits(63)),
|
||
'x-datadog-sampling-priority': '1',
|
||
'x-datadog-trace-id': str(secrets.randbits(63))
|
||
})
|
||
|
||
resp = self.http_client.session.post(
|
||
url,
|
||
json=payload,
|
||
headers=headers,
|
||
cookies=self.http_client.cookies,
|
||
)
|
||
|
||
if DEBUG:
|
||
print(f"Response status: {resp.status_code}")
|
||
|
||
# 更新 cookies
|
||
self.http_client.cookies.update(resp.cookies.get_dict())
|
||
|
||
# 403: Enforcement 挑战（预期）
|
||
if resp.status_code == 403:
|
||
try:
|
||
challenge = resp.json()
|
||
|
||
if DEBUG:
|
||
print(f"✓ Got enforcement challenge!")
|
||
print(f" Type: {challenge.get('type', 'unknown')}")
|
||
if 'proofofwork' in challenge:
|
||
pow_data = challenge['proofofwork']
|
||
print(f" PoW seed: {pow_data.get('seed', 'N/A')}")
|
||
print(f" Difficulty: {pow_data.get('difficulty', 'N/A')}")
|
||
|
||
return challenge
|
||
|
||
except Exception as e:
|
||
if DEBUG:
|
||
print(f"✗ Failed to parse challenge: {e}")
|
||
print(resp.text[:500])
|
||
raise
|
||
|
||
# 200: 成功
|
||
elif resp.status_code == 200:
|
||
if DEBUG:
|
||
print("✓ Registration succeeded!")
|
||
return {'success': True, 'data': resp.json()}
|
||
|
||
# 409: Invalid session
|
||
elif resp.status_code == 409:
|
||
error = resp.json()
|
||
if DEBUG:
|
||
print(f"✗ 409: {error}")
|
||
raise Exception(f"Invalid session: {error}")
|
||
|
||
# 400: 参数错误
|
||
elif resp.status_code == 400:
|
||
error = resp.json()
|
||
if DEBUG:
|
||
print(f"✗ 400: {error}")
|
||
raise Exception(f"Bad request: {error}")
|
||
|
||
# 其他错误
|
||
else:
|
||
if DEBUG:
|
||
print(f"✗ Unexpected status: {resp.status_code}")
|
||
print(resp.text[:500])
|
||
raise Exception(f"Unexpected status: {resp.status_code}\n{resp.text}")
|
||
|
||
|
||
|
||
|
||
def _step4_solve_challenge(self, challenge: Dict) -> str:
|
||
"""解决 enforcement 挑战"""
|
||
if DEBUG:
|
||
print("\n=== STEP 4: Solving enforcement challenge ===")
|
||
|
||
sentinel_token = self.solver.solve_enforcement(challenge)
|
||
|
||
return sentinel_token
|
||
|
||
def _step5_register_with_token(self, email: str, password: str,
|
||
sentinel_token: str) -> Dict:
|
||
"""带 Sentinel token 重新注册"""
|
||
if DEBUG:
|
||
print("\n=== STEP 5: Registering with Sentinel token ===")
|
||
|
||
url = f"{AUTH_BASE_URL}/api/accounts/user/register"
|
||
|
||
payload = {
|
||
'username': email,
|
||
'password': password,
|
||
'client_id': 'sentinel'
|
||
}
|
||
|
||
resp = self.http_client.post(
|
||
url,
|
||
json_data=payload,
|
||
sentinel_token=sentinel_token
|
||
)
|
||
|
||
if resp.status_code == 200:
|
||
if DEBUG:
|
||
print("✓ Registration successful!")
|
||
return {'success': True, 'data': resp.json()}
|
||
|
||
else:
|
||
if DEBUG:
|
||
print(f"✗ Registration failed: {resp.status_code}")
|
||
print(resp.text)
|
||
|
||
return {
|
||
'success': False,
|
||
'status_code': resp.status_code,
|
||
'error': resp.text
|
||
}
|
||
|
||
def register(self, email: str, password: str) -> Dict:
|
||
"""完整注册流程"""
|
||
if DEBUG:
|
||
print(f"\n{'='*60}")
|
||
print(f"Registering: {email}")
|
||
print(f"Session ID: {self.fingerprint.session_id}")
|
||
print(f"{'='*60}")
|
||
|
||
try:
|
||
# Step 1: 获取 cookies 和 CSRF token
|
||
self._step1_get_cookies_and_csrf()
|
||
|
||
# Step 2: 初始化 Sentinel（跳过）
|
||
self._step2_init_sentinel()
|
||
|
||
# Step 3: 尝试注册（获取挑战）
|
||
challenge = self._step3_attempt_register(email, password)
|
||
|
||
# 如果直接成功了
|
||
if challenge.get('success'):
|
||
return challenge
|
||
|
||
# Step 4: 解决挑战
|
||
sentinel_token = self._step4_solve_challenge(challenge)
|
||
|
||
# Step 5: 带 token 重新注册
|
||
result = self._step5_register_with_token(email, password, sentinel_token)
|
||
|
||
return result
|
||
|
||
except Exception as e:
|
||
if DEBUG:
|
||
import traceback
|
||
print(f"\n✗ Registration failed with exception:")
|
||
traceback.print_exc()
|
||
|
||
return {
|
||
'success': False,
|
||
'error': str(e)
|
||
}
|