diff --git a/apps/api/app/api/routes/auth.py b/apps/api/app/api/routes/auth.py index 38438f931..5e8234484 100755 --- a/apps/api/app/api/routes/auth.py +++ b/apps/api/app/api/routes/auth.py @@ -13,7 +13,7 @@ import jwt from app.auth import AuthenticatedUser, blacklist_token, get_current_user from app.config import settings from app.dependencies import get_auth_email_service, get_auth_session_store, get_user_repository -from fastapi import APIRouter, Depends, Header, HTTPException, status +from fastapi import APIRouter, Depends, Header, HTTPException, Request, status from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from pydantic import BaseModel, EmailStr @@ -426,6 +426,7 @@ async def get_wechat_auth_url() -> WechatAuthUrlResponse: @router.post("/wechat/callback", response_model=WechatLoginResponse) async def wechat_callback( request: WechatCallbackRequest, + http_request: Request, user_repository: UserRepository = Depends(get_user_repository), ) -> WechatLoginResponse: """微信登录回调处理""" @@ -433,11 +434,30 @@ async def wechat_callback( from packages.application.auth.wechat_sync_use_case import WechatSyncRequest as SyncRequest from packages.application.auth.wechat_sync_use_case import WechatSyncUseCase + # 回调可观测性：记录 UA（区分微信内置浏览器 MicroMessenger）与 state， + # 便于排查"停留 open.weixin.qq.com / 回调失败"类问题（#1718） + user_agent = http_request.headers.get("User-Agent", "") + is_wechat_browser = "MicroMessenger" in user_agent + logger.info( + "[微信回调] 收到回调: state=%s code_len=%d UA=%r 微信内置浏览器=%s", + (request.state or "")[:8], + len(request.code or ""), + user_agent[:200], + is_wechat_browser, + ) + # 1. 用 code 换微信用户信息 oauth_service = get_wechat_oauth_service() wechat_user, err = oauth_service.handle_callback(request.code, request.state) if err: + # state 校验失败 / 微信 errcode 等错误原文已在 service 内 log，这里带上 UA 上下文 + logger.warning("[微信回调] 处理失败: err=%s 微信内置浏览器=%s", err, is_wechat_browser) raise HTTPException(status_code=400, detail=err) + logger.info( + "[微信回调] state 校验通过，微信用户信息获取成功: openid=%s unionid=%s", + wechat_user.openid[:8] if wechat_user.openid else "", + bool(wechat_user.unionid), + ) # 2. 同步登录/注册（复用 wechat-sync 逻辑） use_case = WechatSyncUseCase(user_repository=user_repository) diff --git a/packages/application/auth/wechat_oauth_service.py b/packages/application/auth/wechat_oauth_service.py index d8cd3d125..08e4416ce 100755 --- a/packages/application/auth/wechat_oauth_service.py +++ b/packages/application/auth/wechat_oauth_service.py @@ -20,6 +20,7 @@ import requests logger = logging.getLogger(__name__) STATE_TTL_SECONDS = 600 # state 有效期 10 分钟 +STATE_KEY_PREFIX = "wechat:state:" # Redis key 前缀（独立逻辑命名空间） class MemoryStateStore: @@ -53,6 +54,80 @@ class MemoryStateStore: del self._states[s] +class RedisStateStore: + """Redis state 存储（多实例/容器重启安全）。 + + 复用现有 Redis（celery broker 同实例），key 前缀 wechat:state:， + TTL 10 分钟，SET NX EX + GETDEL 保证一次性消费。 + Redis 不可用时降级为内存存储，保证登录流程不中断（单节点场景）。 + """ + + def __init__( + self, + redis_url: str = "", + ttl_seconds: int = STATE_TTL_SECONDS, + key_prefix: str = STATE_KEY_PREFIX, + client=None, + ): + self._ttl = ttl_seconds + self._prefix = key_prefix + self._fallback = MemoryStateStore(ttl_seconds=ttl_seconds) + self._redis = None + if client is not None: + # 测试/显式注入 + self._redis = client + return + try: + import redis + + self._redis = redis.Redis.from_url(redis_url, decode_responses=True) + self._redis.ping() + logger.info( + "微信 state 存储使用 Redis: %s db=%s", + self._redis.connection_pool.connection_kwargs.get("host"), + self._redis.connection_pool.connection_kwargs.get("db"), + ) + except Exception as e: # noqa: BLE001 — Redis 不可用降级内存，登录流程不中断 + logger.warning("微信 state Redis 不可用，降级为内存存储: %s", e) + self._redis = None + + def _key(self, state: str) -> str: + return f"{self._prefix}{state}" + + def put(self, state: str) -> None: + if self._redis is None: + self._fallback.put(state) + return + try: + # SET key 1 NX EX ttl：不存在才写入，自带过期 + self._redis.set(self._key(state), "1", nx=True, ex=self._ttl) + except Exception as e: # noqa: BLE001 + logger.warning("微信 state 写入 Redis 失败，降级内存: %s", e) + self._fallback.put(state) + + # Lua：原子读取并删除（单线程执行），兼容所有 Redis 版本（GETDEL 需 6.2+） + _CONSUME_LUA = """ +local v = redis.call('GET', KEYS[1]) +if v then redis.call('DEL', KEYS[1]) end +return v +""" + + def verify_and_consume(self, state: str) -> bool: + if self._redis is None: + return self._fallback.verify_and_consume(state) + try: + try: + val = self._redis.eval(self._CONSUME_LUA, 1, self._key(state)) + except Exception: # noqa: BLE001 — eval 不可用时退化 GET+DELETE + val = self._redis.get(self._key(state)) + if val is not None: + self._redis.delete(self._key(state)) + return val is not None + except Exception as e: # noqa: BLE001 + logger.warning("微信 state 校验 Redis 失败，降级内存: %s", e) + return self._fallback.verify_and_consume(state) + + @dataclass class WechatUserInfo: """微信用户信息""" @@ -158,6 +233,8 @@ class WechatOAuthService: "grant_type": "authorization_code", } token_resp = requests.get(token_url, params=token_params, timeout=10) + # 微信响应头不带 charset，requests 默认按 ISO-8859-1 解码会导致中文乱码 + token_resp.encoding = "utf-8" token_data = token_resp.json() if "errcode" in token_data and token_data["errcode"] != 0: @@ -176,6 +253,8 @@ class WechatOAuthService: "lang": "zh_CN", } user_resp = requests.get(user_url, params=user_params, timeout=10) + # 同上：显式 UTF-8 解码，保证中文昵称/unionid 等不乱码 + user_resp.encoding = "utf-8" user_data = user_resp.json() if "errcode" in user_data and user_data["errcode"] != 0: @@ -206,9 +285,23 @@ class WechatOAuthService: _oauth_service_singleton: WechatOAuthService | None = None +def _build_default_state_store(): + """默认 state 存储：优先 Redis（多实例/重启安全），不可用由 store 内部降级内存。""" + redis_url = "" + try: + from app.config import get_settings + + redis_url = get_settings().CELERY_BROKER_URL or get_settings().REDIS_URL + except Exception: # noqa: BLE001 — API 配置不可用时退回环境变量 + redis_url = os.environ.get("CELERY_BROKER_URL", "") or os.environ.get("REDIS_URL", "") + if redis_url: + return RedisStateStore(redis_url) + return MemoryStateStore() + + def get_wechat_oauth_service() -> WechatOAuthService: """获取微信 OAuth 服务单例（state store 跨请求共享）""" global _oauth_service_singleton if _oauth_service_singleton is None: - _oauth_service_singleton = WechatOAuthService() + _oauth_service_singleton = WechatOAuthService(state_store=_build_default_state_store()) return _oauth_service_singleton diff --git a/tests/unit/test_wechat_callback_logging_1718.py b/tests/unit/test_wechat_callback_logging_1718.py new file mode 100644 index 000000000..4f45ea735 --- /dev/null +++ b/tests/unit/test_wechat_callback_logging_1718.py @@ -0,0 +1,115 @@ +"""#1718：微信回调路由可观测性日志分支覆盖（UA/state/错误透传）。 + +直接驱动 wechat_callback 路由函数，mock OAuth service 与用户仓储： +- 成功路径：日志记录 UA、state 校验通过（MicroMessenger 内置浏览器） +- 失败路径：OAuth 返回错误时记 warning 并抛 400 +""" + +from __future__ import annotations + +import asyncio +import os +import sys +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest + +os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing") +os.environ.setdefault("DATABASE_URL", "sqlite:///test.db") + +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api")) + +from app.api.routes import auth as auth_route # noqa: E402 +from fastapi import HTTPException # noqa: E402 + + +class _FakeRequest: + def __init__(self, ua: str): + self.headers = {"User-Agent": ua} + + +def _wechat_user(): + return SimpleNamespace( + openid="openid-callback-1", + unionid="union-callback-1", + nickname="微信用户", + avatar_url="http://x/a.png", + ) + + +def _fake_oauth_factory(success: bool): + service = MagicMock() + if success: + service.handle_callback.return_value = (_wechat_user(), None) + else: + service.handle_callback.return_value = (None, "无效的 state 参数，请求可能已过期或被篡改") + return service + + +def test_wechat_callback_success_logs_ua_and_state(caplog): + fake_repo = MagicMock() + sync_response = SimpleNamespace( + access_token="at", + refresh_token="rt", + user_id="u-1", + nickname="微信用户", + avatar_url="", + is_new_user=False, + expires_in=1800, + ) + fake_use_case = MagicMock() + fake_use_case.execute.return_value = (sync_response, None) + + user = SimpleNamespace( + id="u-1", + phone_verified=True, + email_verified=True, + email="u@example.com", + ) + fake_repo.find_by_id.return_value = user + + request_obj = SimpleNamespace(code="code-1", state="state-1") + fake_http = _FakeRequest("Mozilla/5.0 (Linux; Android 13) MicroMessenger/8.0.40 WeChat/8.0.40") + + import packages.application.auth.wechat_oauth_service as oauth_mod + import packages.application.auth.wechat_sync_use_case as sync_mod + + orig_oauth = oauth_mod.get_wechat_oauth_service + orig_sync = sync_mod.WechatSyncUseCase + oauth_mod.get_wechat_oauth_service = MagicMock(return_value=_fake_oauth_factory(success=True)) + sync_mod.WechatSyncUseCase = MagicMock(return_value=fake_use_case) + try: + with caplog.at_level("INFO", logger="app.api.routes.auth"): + resp = asyncio.run(auth_route.wechat_callback(request_obj, fake_http, user_repository=fake_repo)) + finally: + oauth_mod.get_wechat_oauth_service = orig_oauth + sync_mod.WechatSyncUseCase = orig_sync + + assert resp.user_id == "u-1" + assert resp.binding_complete is True + log_text = " ".join(rec.getMessage() for rec in caplog.records) + assert "微信回调" in log_text + assert "MicroMessenger" in log_text or "微信内置浏览器=True" in log_text + + +def test_wechat_callback_failure_raises_400_with_detail(caplog): + request_obj = SimpleNamespace(code="code-bad", state="state-bad") + fake_http = _FakeRequest("Mozilla/5.0 Chrome/127") + fake_service = _fake_oauth_factory(success=False) + + import packages.application.auth.wechat_oauth_service as oauth_mod + + orig = oauth_mod.get_wechat_oauth_service + oauth_mod.get_wechat_oauth_service = MagicMock(return_value=fake_service) + try: + with caplog.at_level("WARNING", logger="app.api.routes.auth"): + with pytest.raises(HTTPException) as exc_info: + asyncio.run(auth_route.wechat_callback(request_obj, fake_http, user_repository=MagicMock())) + finally: + oauth_mod.get_wechat_oauth_service = orig + + assert exc_info.value.status_code == 400 + assert "state" in exc_info.value.detail + assert any("微信回调" in rec.getMessage() for rec in caplog.records) diff --git a/tests/unit/test_wechat_state_redis_1718.py b/tests/unit/test_wechat_state_redis_1718.py new file mode 100644 index 000000000..3ab463a28 --- /dev/null +++ b/tests/unit/test_wechat_state_redis_1718.py @@ -0,0 +1,262 @@ +"""#1718：微信 OAuth state 存储 Redis 化 + 中文昵称 UTF-8 解码修复。 + +覆盖（全 mock/fake，CI 无真实 redis 也产生覆盖）： +- RedisStateStore：put 用 SET NX EX、verify_and_consume 用 GETDEL 一次性消费、 + 重复消费返回 False、Redis 异常降级内存、client 注入 +- Redis 不可用（ping 失败）构造时降级内存，功能仍正常 +- GETDEL 不存在（老 Redis）走 GET+DELETE 兜底 +- handle_callback：微信 sns/userinfo 响应含中文 nickname，resp.encoding=utf-8 + 后解析不乱码；errcode 错误路径返回 errmsg +""" + +from __future__ import annotations + +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api")) + +from packages.application.auth import wechat_oauth_service as oauth # noqa: E402 + + +class _FakeRedisClient: + """最小内存版 redis client，模拟 SET NX EX / GETDEL / GET / DELETE / ping。""" + + def __init__(self): + self.data: dict[str, str] = {} + self.ttl: dict[str, int] = {} + self.has_getdel = True + + def ping(self): + return True + + def set(self, key, value, nx=False, ex=None): # noqa: ARG002 + if nx and key in self.data: + return None + self.data[key] = value + if ex is not None: + self.ttl[key] = ex + return True + + def get(self, key): + return self.data.get(key) + + def getdel(self, key): + return self.data.pop(key, None) + + def delete(self, key): + return 1 if self.data.pop(key, None) is not None else 0 + + def eval(self, script, numkeys, key): # noqa: ARG002 + # 模拟 Lua：原子 GET + DEL + return self.data.pop(key, None) + + +# ── RedisStateStore ───────────────────────────────────────────────────── + + +def test_redis_state_store_put_and_consume_once(): + client = _FakeRedisClient() + store = oauth.RedisStateStore(client=client) + store.put("state-abc") + # key 带前缀、TTL 写入 + assert client.data.get("wechat:state:state-abc") is not None + assert client.ttl.get("wechat:state:state-abc") == oauth.STATE_TTL_SECONDS + # 一次性消费：第一次 True，第二次 False + assert store.verify_and_consume("state-abc") is True + assert store.verify_and_consume("state-abc") is False + + +def test_redis_state_store_unknown_state_returns_false(): + store = oauth.RedisStateStore(client=_FakeRedisClient()) + assert store.verify_and_consume("never-put") is False + + +def test_redis_state_store_eval_missing_falls_back_to_get_delete(): + """eval 不可用（如禁用脚本）时退化 GET+DELETE，仍一次性消费。""" + client = _FakeRedisClient() + + def _no_eval(script, numkeys, *keys): # noqa: ARG002 + raise RuntimeError("unknown command EVAL") + + client.eval = _no_eval # type: ignore[method-assign] + store = oauth.RedisStateStore(client=client) + store.put("state-old") + assert store.verify_and_consume("state-old") is True + # GET+DELETE 也消费掉了 + assert "wechat:state:state-old" not in client.data + assert store.verify_and_consume("state-old") is False + + +def test_redis_state_store_put_exception_falls_back_to_memory(): + client = MagicMock() + client.set.side_effect = RuntimeError("redis write fail") + # eval/get 也失败，确保降级到内存 + client.eval.side_effect = RuntimeError("redis read fail") + client.get.side_effect = RuntimeError("redis read fail") + store = oauth.RedisStateStore(client=client) + + store.put("state-fb") # 写 Redis 失败 → 内存 + assert store.verify_and_consume("state-fb") is True # 内存命中 + assert store.verify_and_consume("state-fb") is False + + +def test_redis_state_store_consume_exception_falls_back_to_memory(): + client = MagicMock() + client.set.return_value = True # put 走 Redis + client.eval.side_effect = RuntimeError("redis down") + client.get.side_effect = RuntimeError("redis down") + store = oauth.RedisStateStore(client=client) + + store.put("state-fb2") # 成功写 Redis + # 校验时 Redis 挂了 → 降级内存（内存里没有，返回 False，不报错） + assert store.verify_and_consume("state-fb2") is False + + +def test_redis_state_store_constructor_ping_failure_falls_back(): + """构造时 ping 失败（Redis 不可用）→ 内存降级，功能正常。""" + fake_redis_mod = MagicMock() + fake_client = MagicMock() + fake_client.ping.side_effect = ConnectionError("refused") + fake_redis_mod.Redis.from_url.return_value = fake_client + + with patch.dict(sys.modules, {"redis": fake_redis_mod}): + store = oauth.RedisStateStore(redis_url="redis://nonexistent:6379/0") + + # Redis 不可用 → 内存存储仍工作 + store.put("state-mem") + assert store.verify_and_consume("state-mem") is True + assert store.verify_and_consume("state-mem") is False + + +# ── handle_callback：state 校验 + UTF-8 中文昵称 ──────────────────────── + + +def _configured_service(state_store=None): + store = state_store or oauth.MemoryStateStore() + return oauth.WechatOAuthService( + app_id="wx-test", + app_secret="secret-test", + redirect_uri="https://staging.xiaoxiajianji.com/auth/wechat/callback", + state_store=store, + ) + + +class _FakeResponse: + def __init__(self, payload): + self._payload = payload + self.encoding = None # 模拟微信响应头不带 charset + + def json(self): + # 模拟 requests 行为：按 self.encoding 解码。这里直接返回 payload， + # 但记录 encoding 是否被设置为 utf-8（断言修复生效） + self._decoded_with = self.encoding + return self._payload + + +def test_handle_callback_chinese_nickname_decoded_utf8(monkeypatch): + """微信 userinfo 返回中文昵称，service 设置 encoding=utf-8 后不乱码。""" + service = _configured_service() + state = "state-cn-1" + service._state_store.put(state) + + token_resp = _FakeResponse({"access_token": "at-1", "openid": "openid-cn", "unionid": "union-cn"}) + user_resp = _FakeResponse( + {"openid": "openid-cn", "unionid": "union-cn", "nickname": "微信小应🎬", "headimgurl": ""} + ) + responses = iter([token_resp, user_resp]) + monkeypatch.setattr(oauth.requests, "get", lambda *a, **k: next(responses)) + + info, err = service.handle_callback("code-cn", state) + assert err is None + assert info is not None + assert info.openid == "openid-cn" + assert info.nickname == "微信小应🎬" + # 两个响应都被显式设为 utf-8 + assert token_resp.encoding == "utf-8" + assert user_resp.encoding == "utf-8" + + +def test_handle_callback_state_invalid_returns_error(): + service = _configured_service() + info, err = service.handle_callback("code-x", "state-not-exist") + assert info is None + assert "state" in err + + +def test_handle_callback_wechat_errcode_returns_errmsg(monkeypatch): + """微信返回 errcode（如 code 已被消费 40029）时返回 errmsg 原文。""" + service = _configured_service() + state = "state-err-1" + service._state_store.put(state) + + err_resp = _FakeResponse({"errcode": 40029, "errmsg": "invalid code"}) + monkeypatch.setattr(oauth.requests, "get", lambda *a, **k: err_resp) + + info, err = service.handle_callback("bad-code", state) + assert info is None + assert "invalid code" in err + assert err_resp.encoding == "utf-8" + + +def test_generate_auth_url_stores_state_in_redis(): + """generate_auth_url 生成的 state 写入 Redis（而非仅内存）。""" + client = _FakeRedisClient() + service = oauth.WechatOAuthService( + app_id="wx-test", + app_secret="secret-test", + redirect_uri="https://example.com/cb", + state_store=oauth.RedisStateStore(client=client), + ) + url, state = service.generate_auth_url() + assert f"wechat:state:{state}" in client.data + assert "open.weixin.qq.com" in url + + +# ── _build_default_state_store 工厂分支 ───────────────────────────────── + + +def test_build_default_state_store_uses_redis_when_broker_configured(): + """API settings 有 CELERY_BROKER_URL 时返回 RedisStateStore。""" + store = oauth._build_default_state_store() + # CI/本地通常配置了 redis://localhost:6379/...；无论 Redis 是否可达， + # 返回类型应为 RedisStateStore（内部降级内存） + assert isinstance(store, oauth.RedisStateStore) or isinstance(store, oauth.MemoryStateStore) + + +def test_build_default_state_store_env_fallback(monkeypatch): + """app.config 不可用（如纯 worker 环境）时从环境变量取 redis url。""" + import builtins + + real_import = builtins.__import__ + + def _failing_import(name, *args, **kwargs): + if name == "app.config": + raise ImportError("no app.config") + return real_import(name, *args, **kwargs) + + monkeypatch.setattr(builtins, "__import__", _failing_import) + monkeypatch.setenv("CELERY_BROKER_URL", "redis://localhost:6379/9") + store = oauth._build_default_state_store() + assert isinstance(store, oauth.RedisStateStore) + + +def test_build_default_state_store_no_config_returns_memory(monkeypatch): + """无任何 redis 配置时返回 MemoryStateStore。""" + import builtins + + real_import = builtins.__import__ + + def _failing_import(name, *args, **kwargs): + if name == "app.config": + raise ImportError("no app.config") + return real_import(name, *args, **kwargs) + + monkeypatch.setattr(builtins, "__import__", _failing_import) + monkeypatch.delenv("CELERY_BROKER_URL", raising=False) + monkeypatch.delenv("REDIS_URL", raising=False) + store = oauth._build_default_state_store() + assert isinstance(store, oauth.MemoryStateStore)