fix(security): use secrets for verification code and atomic token rotation

backend/packages/app/src/windup_app/server/user/service.py

@@ -9,7 +9,7 @@
 
 import hashlib
 import logging
-import random
+import secrets
 import string
 import uuid
 from datetime import datetime, timezone
@@ -83,8 +83,8 @@ def _hash_token(token: str) -> str:
 
 
 def _generate_code() -> str:
-    """生成 6 位数字验证码。"""
-    return "".join(random.choices(string.digits, k=6))
+    """生成 6 位数字验证码（密码学安全）。"""
+    return "".join(secrets.choice(string.digits) for _ in range(6))
 
 
 # -- User → UserView 转换 ------------------------------------------------
@@ -372,6 +372,24 @@ def validate_access_token(self, token: str) -> UserView | None:
             email=payload.get("email", ""),
         )
 
+    # -- Lua: 原子 检查-删除-存储 refresh token --------------------------------
+    # KEYS[1] = old_token_key, KEYS[2] = new_token_key
+    # ARGV[1] = ttl, ARGV[2] = user_id
+    # 返回: user_id (成功) 或 nil (旧 token 不存在/已被消费)
+    _ROTATE_TOKEN_SCRIPT = """
+    local old_key = KEYS[1]
+    local new_key = KEYS[2]
+    local ttl     = tonumber(ARGV[1])
+    local user_id = ARGV[2]
+    local cur = redis.call('GET', old_key)
+    if cur == false then
+        return nil
+    end
+    redis.call('DEL', old_key)
+    redis.call('SETEX', new_key, ttl, user_id)
+    return cur
+    """
+
     def refresh_tokens(self, refresh_token: str) -> LoginResult:
         """刷新 token。"""
         payload = decode_token(refresh_token)
@@ -382,24 +400,32 @@ def refresh_tokens(self, refresh_token: str) -> LoginResult:
         if not jti:
             raise BizException("token 无效", code=BizCode.UNAUTHORIZED)
 
+        # user_id 来自已验签的 JWT，可信
+        user_id = int(payload["sub"])
+        email = payload.get("email", "")
+
+        # 签发新 token
+        new_access = create_access_token(user_id, email)
+        new_refresh, new_jti = create_refresh_token(user_id, email)
+
+        # Lua 原子操作：GET old → 存在则 DEL old + SETEX new → 返回 user_id
         token_hash = _hash_token(jti)
-        redis_key = REFRESH_TOKEN_KEY.format(token_hash=token_hash)
-        user_id_str = self.redis.get(redis_key)
+        old_redis_key = REFRESH_TOKEN_KEY.format(token_hash=token_hash)
+        new_token_hash = _hash_token(new_jti)
+        new_redis_key = REFRESH_TOKEN_KEY.format(token_hash=new_token_hash)
+
+        user_id_str = self.redis.eval(
+            self._ROTATE_TOKEN_SCRIPT,
+            2,
+            old_redis_key,
+            new_redis_key,
+            REFRESH_TOKEN_EXPIRE_SECONDS,
+            str(user_id),
+        )
 
         if user_id_str is None:
             raise BizException("refresh token 已失效", code=BizCode.UNAUTHORIZED)
 
-        user_id = int(user_id_str)
-
-        # 撤销旧 token
-        self.redis.delete(redis_key)
-
-        # 签发新 token（需要 email，从旧 token payload 取）
-        email = payload.get("email", "")
-        new_access = create_access_token(user_id, email)
-        new_refresh, new_jti = create_refresh_token(user_id, email)
-        self._store_refresh_token(new_jti, user_id)
-
         logger.info("[WINDUP] token 已刷新 | user_id=%s", user_id)
         return LoginResult(
             user=UserView(id=user_id, email=email),

backend/tests/test_user_service.py

@@ -38,6 +38,7 @@ def mock_redis():
     redis_mock.get.return_value = None
     redis_mock.setex.return_value = True
     redis_mock.delete.return_value = True
+    redis_mock.eval.return_value = None  # Lua 脚本默认返回 None
     redis_mock.pipeline.return_value = MagicMock(
         execute=MagicMock(return_value=[True, True])
     )
@@ -316,26 +317,43 @@ def test_refresh_tokens(service, mock_redis):
     # 先创建一个 refresh token
     token, jti = create_refresh_token(1, "test@example.com")
 
-    # Mock Redis 返回 user_id
-    mock_redis.get.return_value = "1"
+    # Mock Redis eval 返回 user_id（Lua 脚本成功）
+    mock_redis.eval.return_value = "1"
 
     result = service.refresh_tokens(token)
 
     assert result.access_token is not None
     assert result.refresh_token is not None
     assert result.user.id == 1
+    # 验证调用了 eval（Lua 脚本），而不是 get
+    mock_redis.eval.assert_called_once()
 
 
 def test_refresh_tokens_revoked(service, mock_redis):
     token, jti = create_refresh_token(1, "test@example.com")
 
-    # Mock Redis 返回 None（已撤销）
-    mock_redis.get.return_value = None
+    # Mock Redis eval 返回 None（Lua 脚本：旧 token 不存在）
+    mock_redis.eval.return_value = None
 
     with pytest.raises(BizException, match="refresh token 已失效"):
         service.refresh_tokens(token)
 
 
+def test_refresh_tokens_concurrent_reuse(service, mock_redis):
+    """并发重放：同一个 refresh token 第二次调用应失败。"""
+    token, jti = create_refresh_token(1, "test@example.com")
+
+    # 第一次调用成功
+    mock_redis.eval.return_value = "1"
+    result1 = service.refresh_tokens(token)
+    assert result1.access_token is not None
+
+    # 第二次调用（并发重放）失败
+    mock_redis.eval.return_value = None
+    with pytest.raises(BizException, match="refresh token 已失效"):
+        service.refresh_tokens(token)
+
+
 # -- 修改密码测试 --------------------------------------------------------