a16db8e367
1. Sora session_key 按 accountID 隔离（消除跨账号指纹关联） 2. 有 per-account 代理的 Sora 账号跳过 sidecar（保持代理 IP） 3. 请求体用 base64 编码传输（防止二进制数据损坏） 4. Node.js 代理 Body 用 GetBody 安全复制（修复重试时 Body 枯竭）
104 lines
3.1 KiB
Python
104 lines
3.1 KiB
Python
"""
|
|
Sora curl_cffi sidecar — 用 Chrome 131 TLS 指纹绕过 Cloudflare
|
|
sub2api 通过 HTTP 调用此服务转发 Sora 请求
|
|
"""
|
|
from flask import Flask, request, Response
|
|
from curl_cffi import requests as cffi_requests
|
|
import json, os, time, threading, base64
|
|
|
|
app = Flask(__name__)
|
|
|
|
IMPERSONATE = os.environ.get("IMPERSONATE", "chrome131")
|
|
TIMEOUT = int(os.environ.get("TIMEOUT_SECONDS", "60"))
|
|
|
|
# 会话池：按 session_key 复用，避免每次请求都做 TLS 握手
|
|
sessions = {}
|
|
sessions_lock = threading.Lock()
|
|
SESSION_TTL = int(os.environ.get("SESSION_TTL_SECONDS", "3600"))
|
|
|
|
|
|
def get_session(session_key="default"):
|
|
with sessions_lock:
|
|
entry = sessions.get(session_key)
|
|
if entry and (time.time() - entry["created"]) < SESSION_TTL:
|
|
return entry["session"]
|
|
s = cffi_requests.Session(impersonate=IMPERSONATE)
|
|
sessions[session_key] = {"session": s, "created": time.time()}
|
|
return s
|
|
|
|
|
|
@app.route("/health", methods=["GET"])
|
|
def health():
|
|
return json.dumps({"status": "ok", "impersonate": IMPERSONATE}), 200
|
|
|
|
|
|
@app.route("/proxy", methods=["POST"])
|
|
def proxy():
|
|
"""
|
|
接收 sub2api 的代理请求，用 curl_cffi + Chrome 指纹转发到目标 URL
|
|
请求体 JSON:
|
|
{
|
|
"url": "https://sora.chatgpt.com/backend/me",
|
|
"method": "GET",
|
|
"headers": {"Authorization": "Bearer xxx", ...},
|
|
"body": "...",
|
|
"session_key": "account_123" // 可选
|
|
}
|
|
"""
|
|
try:
|
|
data = request.get_json(force=True)
|
|
except Exception:
|
|
return json.dumps({"error": "invalid json"}), 400
|
|
|
|
url = data.get("url", "")
|
|
method = data.get("method", "GET").upper()
|
|
headers = data.get("headers", {})
|
|
body = data.get("body")
|
|
is_base64 = data.get("is_base64", False)
|
|
session_key = data.get("session_key", "default")
|
|
|
|
if not url:
|
|
return json.dumps({"error": "url required"}), 400
|
|
|
|
# 解码 base64 编码的请求体（安全传输二进制数据）
|
|
req_data = None
|
|
if body:
|
|
if is_base64:
|
|
try:
|
|
req_data = base64.b64decode(body)
|
|
except Exception:
|
|
req_data = body.encode("utf-8") if isinstance(body, str) else body
|
|
else:
|
|
req_data = body.encode("utf-8") if isinstance(body, str) else body
|
|
|
|
try:
|
|
sess = get_session(session_key)
|
|
resp = sess.request(
|
|
method=method,
|
|
url=url,
|
|
headers=headers,
|
|
data=req_data,
|
|
timeout=TIMEOUT,
|
|
allow_redirects=True,
|
|
)
|
|
|
|
# 透传响应
|
|
excluded_headers = {"transfer-encoding", "content-encoding", "connection"}
|
|
resp_headers = {
|
|
k: v for k, v in resp.headers.items()
|
|
if k.lower() not in excluded_headers
|
|
}
|
|
|
|
return Response(
|
|
response=resp.content,
|
|
status=resp.status_code,
|
|
headers=resp_headers,
|
|
)
|
|
except Exception as e:
|
|
return json.dumps({"error": str(e)}), 502
|
|
|
|
|
|
if __name__ == "__main__":
|
|
port = int(os.environ.get("PORT", "8080"))
|
|
app.run(host="0.0.0.0", port=port)
|