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)

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

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)

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)