From 1f7e91b645a28a5cdb7e673ecd78bd00455160bd Mon Sep 17 00:00:00 2001 From: paysen Date: Tue, 20 Oct 2026 11:25:52 +0800 Subject: [PATCH] fix(llm): serialize ChatGPT OAuth token refresh (#7092) --- app/agent/llm/auth.py | 63 ++++++++---- app/agent/llm/provider.py | 1 + app/agent/llm/runtime.py | 6 +- tests/test_llm_chatgpt_refresh.py | 166 ++++++++++++++++++++++++++++++ 4 files changed, 213 insertions(+), 23 deletions(-) create mode 100644 tests/test_llm_chatgpt_refresh.py diff --git a/app/agent/llm/auth.py b/app/agent/llm/auth.py index 9e71e8c2c..3f928917a 100644 --- a/app/agent/llm/auth.py +++ b/app/agent/llm/auth.py @@ -4,6 +4,7 @@ LLM Provider 持久鉴权和外部授权协议的唯一 owner。 from __future__ import annotations +import asyncio import base64 import copy import hashlib @@ -102,6 +103,7 @@ class _ProviderAuth: """ _lock: Any + _chatgpt_refresh_lock: asyncio.Lock _pending_sessions: dict[str, PendingAuthSession] _oauth_state_index: dict[str, str] @@ -485,6 +487,23 @@ class _ProviderAuth: }, raise_exception=True, ) + if response.status_code >= 400: + # 只展示已知错误码的本地说明，不把响应正文或令牌带入日志和管理接口。 + try: + error = response.json().get("error") + except (ValueError, AttributeError): + error = None + code = error.get("code") if isinstance(error, dict) else error + messages = { + "refresh_token_reused": "ChatGPT 刷新令牌已被使用，请重新授权；请勿在多个客户端复制同一授权", + "refresh_token_expired": "ChatGPT 刷新令牌已过期，请重新授权", + "refresh_token_invalidated": "ChatGPT 授权已被撤销，请重新授权", + "invalid_grant": "ChatGPT 授权已失效，请重新授权", + } + message = messages.get(code) if isinstance(code, str) else None + raise LLMProviderAuthError( + message or f"ChatGPT 令牌刷新失败（HTTP {response.status_code}），请稍后重试" + ) response.raise_for_status() payload = response.json() if not isinstance(payload, dict): @@ -578,25 +597,27 @@ class _ProviderAuth: """ 解析并返回 ChatGPT OAuth 鉴权，支持自动刷新 Token。 """ - auth = self.get_saved_auth("chatgpt") - if not auth or auth.get("type") != "oauth": - raise LLMProviderAuthError("尚未完成 ChatGPT Plus/Pro 授权") + # 读取、刷新和持久化必须串行；等待者重新读取轮换后的令牌，不能重用旧令牌。 + async with self._chatgpt_refresh_lock: + auth = self.get_saved_auth("chatgpt") + if not auth or auth.get("type") != "oauth": + raise LLMProviderAuthError("尚未完成 ChatGPT Plus/Pro 授权") - expires_at = auth.get("expires_at") - refresh_token = auth.get("refresh_token") - # 预留 60 秒刷新缓冲，避免刚发起请求就遇到过期。 - if expires_at and refresh_token and int(expires_at) <= int(time.time()) + 60: - payload = await self._refresh_chatgpt_tokens(refresh_token) - claims = self._jwt_claims(payload.get("id_token") or payload["access_token"]) - auth.update( - { - "access_token": payload["access_token"], - "refresh_token": payload.get("refresh_token") or refresh_token, - "expires_at": int(time.time() + int(payload.get("expires_in") or 3600)), - "account_id": auth.get("account_id") or self._extract_chatgpt_account_id(claims), - "email": auth.get("email") or claims.get("email"), - "label": auth.get("label") or claims.get("email") or auth.get("account_id") or "ChatGPT Plus/Pro", - } - ) - await self.save_auth("chatgpt", auth) - return auth + expires_at = auth.get("expires_at") + refresh_token = auth.get("refresh_token") + # 预留 60 秒刷新缓冲，避免刚发起请求就遇到过期。 + if expires_at and refresh_token and int(expires_at) <= int(time.time()) + 60: + payload = await self._refresh_chatgpt_tokens(refresh_token) + claims = self._jwt_claims(payload.get("id_token") or payload["access_token"]) + auth.update( + { + "access_token": payload["access_token"], + "refresh_token": payload.get("refresh_token") or refresh_token, + "expires_at": int(time.time() + int(payload.get("expires_in") or 3600)), + "account_id": auth.get("account_id") or self._extract_chatgpt_account_id(claims), + "email": auth.get("email") or claims.get("email"), + "label": auth.get("label") or claims.get("email") or auth.get("account_id") or "ChatGPT Plus/Pro", + } + ) + await self.save_auth("chatgpt", auth) + return auth diff --git a/app/agent/llm/provider.py b/app/agent/llm/provider.py index 96503dff5..084cfef6c 100644 --- a/app/agent/llm/provider.py +++ b/app/agent/llm/provider.py @@ -58,6 +58,7 @@ class LLMProviderManager( """ self._lock = threading.RLock() self._models_dev_lock = asyncio.Lock() + self._chatgpt_refresh_lock = asyncio.Lock() self._pending_sessions: dict[str, PendingAuthSession] = {} self._oauth_state_index: dict[str, str] = {} self._models_dev_data: dict[str, Any] | None = None diff --git a/app/agent/llm/runtime.py b/app/agent/llm/runtime.py index 785bac47d..108e32a28 100644 --- a/app/agent/llm/runtime.py +++ b/app/agent/llm/runtime.py @@ -99,10 +99,12 @@ class _ProviderRuntime: if normalized_provider_id == "chatgpt": auth = None + auth_error: Exception = LLMProviderAuthError("请提供 API Key 或完成 ChatGPT 授权") try: auth = await self._resolve_chatgpt_oauth() except Exception as err: - logger.debug(f"解析 ChatGPT OAuth 鉴权失败，回退 API Key 模式: {err}") + auth_error = err + logger.debug(f"解析 ChatGPT OAuth 鉴权失败: {err}") if auth: headers = {"originator": "moviepilot"} @@ -138,7 +140,7 @@ class _ProviderRuntime: ) return result - raise LLMProviderAuthError("请提供 API Key 或完成 ChatGPT 授权") + raise auth_error if normalized_provider_id == "github-copilot": auth = self.get_saved_auth("github-copilot") diff --git a/tests/test_llm_chatgpt_refresh.py b/tests/test_llm_chatgpt_refresh.py new file mode 100644 index 000000000..ae85670aa --- /dev/null +++ b/tests/test_llm_chatgpt_refresh.py @@ -0,0 +1,166 @@ +""" +验证 ChatGPT 轮换令牌的并发续期、持久化和错误归因。 +""" + +import asyncio +import copy +import time +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import httpx2 +import pytest + +from app.agent.llm.provider import LLMProviderAuthError, LLMProviderManager + + +@pytest.fixture +def auth_state(monkeypatch): + """ + 使用独立管理器和内存配置替代持久化边界，禁止访问真实凭据和网络。 + """ + manager = object.__new__(LLMProviderManager) + manager.__init__() + config = { + "provider_auth": { + "chatgpt": { + "type": "oauth", + "access_token": "old-access", + "refresh_token": "old-refresh", + "expires_at": int(time.time()) - 60, + "account_id": "test-account", + }, + "github-copilot": {"type": "oauth", "access_token": "other-provider"}, + }, + } + + async def write_config(value): + """ + 模拟异步落库完成后才发布配置快照。 + """ + await asyncio.sleep(0) + config.clear() + config.update(copy.deepcopy(value)) + + monkeypatch.setattr(manager, "_read_agent_config", lambda: copy.deepcopy(config)) + monkeypatch.setattr(manager, "_write_agent_config", AsyncMock(side_effect=write_config)) + monkeypatch.setattr(manager, "_jwt_claims", lambda _token: {}) + monkeypatch.setattr(manager, "_build_async_request", lambda **_kwargs: pytest.fail("意外网络请求")) + return manager, config + + +def test_concurrent_requests_refresh_and_persist_once(auth_state, monkeypatch): + """ + 并发调用只使用一次旧令牌，所有等待者读取已持久化的新令牌和官方有效期。 + """ + manager, config = auth_state + now = int(time.time()) + + async def refresh(_token): + """ + 模拟网络等待，让未受保护的并发请求同时持有旧令牌。 + """ + await asyncio.sleep(0) + return {"access_token": "new-access", "refresh_token": "new-refresh", "expires_in": 864000} + + refresh_mock = AsyncMock(side_effect=refresh) + monkeypatch.setattr(manager, "_refresh_chatgpt_tokens", refresh_mock) + + async def resolve_all(): + """ + 同时模拟聊天、识别和模型目录调用。 + """ + return await asyncio.gather(*(manager._resolve_chatgpt_oauth() for _ in range(24))) + + results = asyncio.run(resolve_all()) + + refresh_mock.assert_awaited_once_with("old-refresh") + manager._write_agent_config.assert_awaited_once() + saved = config["provider_auth"]["chatgpt"] + assert all(result == saved for result in results) + assert saved["refresh_token"] == "new-refresh" + assert now + 864000 <= saved["expires_at"] <= int(time.time()) + 864000 + assert config["provider_auth"]["github-copilot"]["access_token"] == "other-provider" + + +def test_valid_token_does_not_refresh(auth_state, monkeypatch): + """ + 官方有效期内继续使用保存的授权，不提前触发轮换。 + """ + manager, config = auth_state + config["provider_auth"]["chatgpt"]["expires_at"] = int(time.time()) + 86400 + refresh = AsyncMock() + monkeypatch.setattr(manager, "_refresh_chatgpt_tokens", refresh) + result = asyncio.run(manager._resolve_chatgpt_oauth()) + assert result["access_token"] == "old-access" + refresh.assert_not_awaited() + manager._write_agent_config.assert_not_awaited() + + +@pytest.mark.parametrize("error", [TimeoutError("offline"), LLMProviderAuthError("official rejection")]) +def test_refresh_failure_preserves_saved_auth(auth_state, monkeypatch, error): + """ + 网络失败或官方拒绝不会清除授权、伪造有效期或在本次调用中重发。 + """ + manager, config = auth_state + before = copy.deepcopy(config) + refresh = AsyncMock(side_effect=error) + monkeypatch.setattr(manager, "_refresh_chatgpt_tokens", refresh) + with pytest.raises(type(error), match=str(error)): + asyncio.run(manager._resolve_chatgpt_oauth()) + assert config == before + refresh.assert_awaited_once() + manager._write_agent_config.assert_not_awaited() + assert not manager._chatgpt_refresh_lock.locked() + + +@pytest.mark.parametrize( + ("status", "payload", "expected"), + [ + (401, {"error": {"code": "refresh_token_reused", "message": "private-response"}}, "已被使用"), + (401, {"error": {"code": "refresh_token_expired"}}, "已过期"), + (401, {"error": {"code": "refresh_token_invalidated"}}, "已被撤销"), + (400, {"error": "invalid_grant", "error_description": "private-response"}, "已失效"), + (429, {"error": {"code": "private-response"}}, "HTTP 429"), + (503, ["private-response"], "HTTP 503"), + (403, None, "HTTP 403"), + ], +) +def test_refresh_error_uses_safe_official_reason(auth_state, monkeypatch, status, payload, expected): + """ + 官方拒绝保留可操作原因，网关或限流错误不误报注销，正文不出现在错误中。 + """ + manager, _config = auth_state + response = httpx2.Response( + status, + **({"json": payload} if payload is not None else {"text": "private-response"}), + request=httpx2.Request("POST", "https://auth.openai.com/oauth/token"), + ) + post = AsyncMock(return_value=response) + monkeypatch.setattr(manager, "_build_async_request", lambda **_kwargs: SimpleNamespace(post_res=post)) + with pytest.raises(LLMProviderAuthError, match=expected) as caught: + asyncio.run(manager._refresh_chatgpt_tokens("test-refresh")) + assert "private-response" not in str(caught.value) + assert "test-refresh" not in str(caught.value) + if status in (403, 429, 503): + assert "重新授权" not in str(caught.value) + post.assert_awaited_once() + + +@pytest.mark.parametrize("api_key", [None, "fallback-key"]) +def test_runtime_preserves_oauth_failure_without_api_key(auth_state, monkeypatch, api_key): + """ + 只有明确提供 API Key 才回退，否则将真实续期失败传给管理接口。 + """ + manager, _config = auth_state + provider = manager.get_provider("chatgpt") + monkeypatch.setattr(manager, "_get_provider_async", AsyncMock(return_value=provider)) + monkeypatch.setattr(manager, "_resolve_chatgpt_oauth", AsyncMock(side_effect=LLMProviderAuthError("已被使用"))) + if api_key: + result = asyncio.run(manager.resolve_runtime("chatgpt", "test-model", api_key=api_key)) + assert result["auth_mode"] == "api_key" + assert result["api_key"] == api_key + else: + result = asyncio.run(manager.provider_manage("chatgpt", "list_models")) + assert result["success"] is False + assert "已被使用" in result["message"]