前言 #
2023 年初,ChatGPT 火得一塌糊涂,但用起来有几个明显的痛点:Web 端体验不够灵活,没法接入自己的应用;官方 API1 按 Token 收费,对于重度使用者来说成本不低;而且 Web 端和 API 是两套完全独立的系统,ChatGPT Plus 订阅用户的 GPT-4 额度没法通过 API 消耗。
于是我萌生了一个想法:能不能把 ChatGPT Web 端的接口逆向出来,包装成标准 OpenAI API 格式来用?这样既能用上 Web 端的无限额度,又能接入自己的工具链。
这篇文章记录一下我从抓包分析到最终实现的完整过程。
先搞清楚 ChatGPT Web 端的技术架构 #
动手之前,先用浏览器 DevTools 的 Network 面板把 ChatGPT Web 端的请求链路摸清楚。
认证链路 #
ChatGPT 使用 Auth02 作为 OAuth2 认证提供商。正常的 Web 登录流程是:
浏览器 → chat.openai.com/auth/login
→ 跳转到 auth0.openai.com(Auth0 托管的登录页)
→ 用户输入邮箱密码
→ Auth0 回调返回 access_token
→ 前端把 token 存到 session 中但这个流程是给浏览器设计的,有多次 302 重定向、Cookie 传递、JavaScript 渲染。命令行工具没法直接走这个流程。
对话链路 #
登录后,前端发消息用的接口长这样:
POST https://chat.openai.com/backend-api/conversation
Authorization: Bearer <access_token>
Content-Type: application/json
Accept: text/event-stream请求体:
{
"action": "next",
"messages": [
{
"id": "uuid",
"role": "user",
"content": { "content_type": "text", "parts": ["你好"] }
}
],
"model": "text-davinci-002-render-sha",
"parent_message_id": "uuid"
}响应是 SSE(Server-Sent Events)3 格式,逐字输出:
data: {"message": {"content":{"parts":["你"]}, ...}}
data: {"message": {"content":{"parts":["你好"]}, ...}}
data: [DONE]这里有个关键差异:Web 端的对话是一棵消息树(每条消息有 parent_message_id,支持分支和重新生成),而 OpenAI API 是线性的 messages 数组。后面做协议转换的时候需要处理这个差异。
逆向 Auth0 认证流程 #
这是整个过程中最核心也最有趣的一步。
思路转变:从 Web 到 iOS #
一开始我尝试直接模拟浏览器的登录流程,但很快就发现行不通:Auth0 的登录页有大量反爬措施,JavaScript 校验、浏览器指纹检测、reCAPTCHA 都有。
换个思路:移动端的认证流程通常比 Web 端简单。于是抓包分析了 ChatGPT 的 iOS 客户端,发现它也用 Auth0,但走的是 OAuth2 + PKCE(Proof Key for Code Exchange)4 扩展,不需要浏览器环境。
PKCE 简述 #
PKCE 是 OAuth2 的安全扩展,原本是为无法安全存储 client_secret 的移动端和桌面端应用设计的。流程很简单:
客户端生成 code_verifier(随机字符串)
客户端计算 code_challenge = SHA256(code_verifier)
授权请求中带上 code_challenge
回调时带上原始的 code_verifier
服务端验证 SHA256(code_verifier) == code_challenge,发放 Token好处是:即使授权码被截获,没有 code_verifier 也无法换取 Token。
拆解出的认证流程 #
通过分析 iOS 客户端的网络请求,我把完整的认证流程拆解成了 7 步:
- 获取
preauth_cookie - 拼装 authorize URL,模拟 iOS 客户端参数
- 访问 authorize URL,提取 state 参数并保存 Cookie
- 提交邮箱
- 提交密码
- 处理回调或 MFA 验证
- 如果需要 MFA,提交验证码后回到第 6 步
- 最终用授权码换取 Access Token
有几个值得注意的细节:
code_verifier 为什么可以硬编码? 因为 iOS 客户端是可以反编译的,code_verifier 和 code_challenge 这对值写死在客户端里,所有 iOS 用户共用同一对。PKCE 在这种场景下保护的是传输链路(授权码泄露不会导致 Token 泄露),而不是客户端本身。
client_id 是哪来的? 同样来自 iOS 客户端反编译。这是 OpenAI 在 Auth0 注册的 iOS 应用 ID。
redirect_uri 为什么是 com.openai.chat://...? 这是 iOS 的 URL Scheme,用于 Auth0 授权完成后跳回 App。在我们的实现里不需要真正跳转,只需要从响应的 Location 头里提取 code 参数。
Python 实现大致长这样:
class Auth0:
def auth(self, login_local=False) -> str:
return self.__part_one() if login_local else self.get_access_token_proxy()
def __part_one(self): # Step 1: get preauth
def __part_two(self): # Step 2: build authorize URL
def __part_three(self): # Step 3: follow authorize
def __part_four(self): # Step 4: submit email
def __part_five(self): # Step 5: submit password
def __part_six(self): # Step 6: handle callback/MFA
def __part_seven(self): # Step 7: MFA OTP
def get_access_token(self): # Final: code → token实现 SSE 流式对话代理 #
拿到 Access Token 后,下一步是调 ChatGPT 的对话接口。
请求构造比较直观:每条消息需要一个 UUID 作为 id,parent_message_id 指向上一条消息形成对话链,首次对话不带 conversation_id,服务端会创建并返回。action 可以是 next(新消息)、variant(重新生成)、continue(继续输出)。
难点在于 SSE 响应的处理。Python 的 Flask 是同步框架,但 SSE 需要异步消费流式响应。我的解决方案是异步线程 + 阻塞队列 + Generator 桥接:
def _request_sse(self, url, headers, data):
queue, event = block_queue.Queue(), threading.Event()
t = threading.Thread(target=asyncio.run,
args=(self._do_request_sse(url, headers, data, queue, event),))
t.start()
return queue.get(), queue.get(), self.__generate_wrap(queue, t, event)为什么要绕这么一圈?因为 httpx5 的流式 API 是 async 的(async with client.stream('POST', url) 需要在 async 上下文中),但上层是同步代码(Flask 的路由处理函数、CLI 的 readline 循环都是同步的),又不想把全局架构从 Flask 改成 aiohttp/uvicorn,改动太大。
所以用一个线程跑异步事件循环,通过 queue.Queue 把异步世界的数据搬运到同步世界,对外暴露一个标准 Generator,上层代码完全无感。
还有一个细节:threading.Event 用于中断保护。如果客户端断开连接触发了 GeneratorExit,Event 被置位,异步线程检测到后主动关闭 httpx 连接,避免线程泄漏。
Web API 到 OpenAI API 的协议转换 #
这是把 ChatGPT Web 接口包装成标准 OpenAI API 的关键步骤。两种 API 格式差异很大:
| 维度 | ChatGPT Web API | OpenAI Public API |
|---|---|---|
| 认证 | Bearer access_token | Bearer sk-xxx(API Key) |
| 请求格式 | 消息树(parent_message_id) | messages 数组 |
| 响应格式 | SSE + 消息树节点 | SSE + choices 数组 |
| 会话管理 | 服务端存储 conversation_id | 无状态 |
请求转换 #
我的做法是维护一个本地的消息树,把 OpenAI 格式的 messages 数组转换成树形结构,支持多轮对话和 regenerate:
def talk(self, content, model, message_id, parent_message_id, ...):
if conversation_id:
parent = conversation.get_prompt(parent_message_id)
else:
parent = conversation.add_prompt(Prompt(parent_message_id))
parent = conversation.add_prompt(SystemPrompt(self.system_prompt, parent))
conversation.add_prompt(UserPrompt(message_id, content, parent))
user_prompt, gpt_prompt, messages = conversation.get_messages(message_id, model)响应转换 #
Web 端每次返回的是全量文本(parts[0] 越来越长),而 OpenAI API 返回的是增量文本。需要做差值计算:
# Web 端返回
{"message": {"content": {"parts": ["完整文本"]}, "author": {"role": "assistant"}}}
# 转换为 OpenAI 格式
data: {"choices": [{"delta": {"content": "增量文本"}, "finish_reason": null}]}
data: {"choices": [{"delta": {}, "finish_reason": "stop"}]}
data: [DONE]Token 超限裁剪 #
OpenAI API 有 Token 上限(gpt-3.5-turbo 是 4096,gpt-4 是 8192)。对话历史过长时需要本地裁剪:
def __reduce_messages(self, messages, model, token=None):
max_tokens = self.FAKE_TOKENS[model] if self.__is_fake_api(token) else self.MAX_TOKENS[model]
while gpt_num_tokens(messages) > max_tokens - 200:
if len(messages) < 2:
raise Exception('prompt too long')
messages.pop(1) # 从第 2 条开始删,保留 system prompt 和最新对话
return messages裁剪策略:保留 messages[0](system prompt)和最新的几轮对话,从最早的用户消息开始删。- 200 是给模型回复留的余量。
从技术验证到实际交付 #
接口跑通之后,接下来的问题是怎么让身边的同事和朋友也能用上。
批量注册 #
接口跑通之后,实际用起来发现 ChatGPT 有单账号频率限制,请求量一大就报错。要解决这个问题,最直接的办法就是多账号。于是我写了个注册机,使用自己的域名邮箱批量注册了 200 个 ChatGPT 账号。其中两个开了 Plus 订阅(只有 Plus 才能用 GPT-4),费用跟几个朋友均摊。剩下的账号都走免费的 GPT-3.5,日常使用完全够用。
Token 管理与持久化 #
Access Token 有效期 14 天,到期后需要重新认证刷新。我把所有账号的 Token 存到了 PostgreSQL 数据库中,写了个定时任务自动检测过期时间并批量刷新,保证 Token 池始终可用。
负载均衡 #
200 个账号不能只用一个。我在代理服务层加了一层简单的负载均衡:每次请求过来,从数据库中轮询选取一个可用的 Token 去调用 ChatGPT 接口。这样既避免了单账号频率限制,也分散了各账号的请求压力。
最终的效果是:对外暴露一个标准的 OpenAI API 地址,同事和朋友只需要在自己的应用里把 API Base URL 指向我的服务就行,完全感知不到背后是 200 个账号在轮转。GPT-4 的请求路由到 Plus 账号池,GPT-3.5 的请求路由到免费账号池。