feat(auth): 短信登录防刷 — 同手机号同IP 每小时限 5 次

/sms/login 增加 (手机号+IP) 每小时最多 5 次限流,挡撞库爆破/高频重登;
与路由上原有 IP 维度限流(20次/分钟,跨号)互补。
- 计数放在验证码校验之前 → 输错码的失败尝试也计数
- 测试账号(.env TEST_ACCOUNT_PHONE)豁免,走自己的每日额度
- 复用 ratelimit 内存固定窗口,受 RATE_LIMIT_ENABLED 总开关控制(默认开)
- 新增 enforce_rate_limit(供路由内按请求体字段如手机号限流)
- 补 test_auth/test_test_account 回归测试(普通号到5次被拦、测试号豁免)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

app/api/v1/auth.py

@@ -12,11 +12,11 @@ from __future__ import annotations
 
 import logging
 
-from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi import APIRouter, Depends, HTTPException, Request, status
 
 from app.api.deps import CurrentUser, DbSession
 from app.core import test_account
-from app.core.ratelimit import rate_limit
+from app.core.ratelimit import enforce_rate_limit, rate_limit
 from app.core.security import TokenError, decode_token, issue_token_pair
 from app.integrations.jiguang import JiguangError, mask_phone, verify_and_get_phone
 from app.integrations.sms import SmsError, send_code, verify_code
@@ -38,6 +38,9 @@ logger = logging.getLogger("shagua.auth")
 
 router = APIRouter(prefix="/api/v1/auth", tags=["auth"])
 
+# 手机号登录防刷:同一手机号 + 同一 IP 每小时最多的登录尝试次数(成功/失败都计)。
+SMS_LOGIN_MAX_PER_HOUR = 5
+
 
 def _login_response(
     user, *, onboarding_completed: bool, force_onboarding: bool = False
@@ -109,9 +112,10 @@ def sms_send(req: SmsSendRequest) -> SmsSendResponse:
     summary="手机号+验证码登录",
     dependencies=[Depends(rate_limit(20, 60, "sms-login"))],  # 防撞库爆破(另有单码失败次数上限)
 )
-def sms_login(req: SmsLoginRequest, db: DbSession) -> TokenWithUser:
+def sms_login(req: SmsLoginRequest, request: Request, db: DbSession) -> TokenWithUser:
     # 测试账号:免验证码登录 + 每日上限 + 每次都走新手引导(详见 app/core/test_account.py)。
     # 放在最前面:命中即不校验验证码;先扣当日额度,超限直接拒,挡住有人猜到号后脚本刷。
+    # 测试账号走自己的每日额度、不受下面 (手机号+IP) 每小时限流约束(QA 需在一小时内反复登录联调)。
     if test_account.is_test_account(req.phone):
         if not test_account.try_consume_quota():
             raise HTTPException(status_code=429, detail="测试账号今日使用次数已达上限,请明天再试")
@@ -123,6 +127,17 @@ def sms_login(req: SmsLoginRequest, db: DbSession) -> TokenWithUser:
         # 不读/不写 onboarding_completion 表。
         return _login_response(user, onboarding_completed=False, force_onboarding=True)
 
+    # 防刷:同一手机号 + 同一 IP 每小时最多 SMS_LOGIN_MAX_PER_HOUR 次登录尝试。放在验证码校验**之前**
+    # → 输错验证码的失败尝试也计数,才挡得住撞库/爆破。与路由上 IP 维度的 sms-login 限流(同 IP 多号)互补。
+    enforce_rate_limit(
+        request,
+        scope="sms-login-phone",
+        subject=req.phone,
+        limit=SMS_LOGIN_MAX_PER_HOUR,
+        window_sec=3600,
+        detail="登录尝试过于频繁,请稍后再试",
+    )
+
     if not verify_code(req.phone, req.code):
         raise HTTPException(status_code=400, detail="invalid sms code")
 

app/core/ratelimit.py

@@ -57,3 +57,29 @@ def rate_limit(limit: int, window_sec: float, scope: str):
             )
 
     return _dep
+
+
+def enforce_rate_limit(
+    request: Request,
+    scope: str,
+    subject: str,
+    limit: int,
+    window_sec: float,
+    *,
+    detail: str = "操作过于频繁,请稍后再试",
+) -> None:
+    """在路由内部手动限流,按 (subject, 客户端 IP) 计数。
+
+    用于限流 key 需要请求体字段(如手机号)、Depends 阶段还拿不到 body 时
+    —— 此时无法用 [rate_limit] 依赖,改在 handler 解析完 body 后调用本函数。
+    key = `scope:subject:client_ip`;同一 (subject, IP) 在 window_sec 内超过 limit 次 → 抛 429。
+    受 [settings.RATE_LIMIT_ENABLED] 总开关控制(与 [rate_limit] 一致)。
+    """
+    if not settings.RATE_LIMIT_ENABLED:
+        return
+    key = f"{scope}:{subject}:{_client_ip(request)}"
+    if not _hit(key, limit, window_sec):
+        raise HTTPException(
+            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+            detail=detail,
+        )

tests/test_auth.py

@@ -74,6 +74,30 @@ def test_sms_send_too_frequent(client) -> None:
     assert client.post("/api/v1/auth/sms/send", json={"phone": phone}).status_code == 429
 
 
+def test_sms_login_phone_ip_rate_limit(client, monkeypatch) -> None:
+    """防刷:同一手机号 + 同一 IP 每小时最多 SMS_LOGIN_MAX_PER_HOUR 次登录尝试,超出 429。
+    conftest 默认 RATE_LIMIT_ENABLED=false(内存计数跨用例累加),本用例临时打开并清空计数隔离。"""
+    from app.api.v1 import auth
+    from app.core import ratelimit
+
+    monkeypatch.setattr(ratelimit.settings, "RATE_LIMIT_ENABLED", True)
+    ratelimit._buckets.clear()  # 隔离:清掉跨用例累加的内存计数
+
+    phone = "13533153300"
+    # 前 N 次:mock 校验放行(任意 6 位数字)→ 200 登录成功
+    for i in range(auth.SMS_LOGIN_MAX_PER_HOUR):
+        r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"})
+        assert r.status_code == 200, f"第 {i + 1} 次应放行: {r.text}"
+    # 第 N+1 次:同号同 IP 超限 → 429
+    r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"})
+    assert r.status_code == 429, r.text
+    assert "频繁" in r.json()["detail"]
+
+    # 同 IP 换个手机号 → 独立计数(限流键含手机号、非纯 IP),仍放行
+    r = client.post("/api/v1/auth/sms/login", json={"phone": "13533153301", "code": "123456"})
+    assert r.status_code == 200, r.text
+
+
 def test_sms_login_bad_code(client) -> None:
     """mock 下校验位数:5 位过 schema 但业务要 6 位 → 400。"""
     phone = "13700137000"

tests/test_test_account.py

@@ -142,6 +142,28 @@ def test_send_does_not_consume_quota(client, enabled, monkeypatch) -> None:
     assert client.post("/api/v1/auth/sms/login", json={"phone": TEST_PHONE, "code": "1234"}).status_code == 429
 
 
+# ============================ 登录限流豁免 ============================
+
+def test_exempt_from_phone_ip_login_rate_limit(client, enabled, monkeypatch) -> None:
+    """测试号豁免 (手机号+IP) 每小时登录限流。
+
+    普通号同号同 IP 到 SMS_LOGIN_MAX_PER_HOUR 次即被拦(见 test_auth.test_sms_login_phone_ip_rate_limit);
+    测试号走自己的每日额度、不受这道限流——这里临时打开 RATE_LIMIT_ENABLED,连登远超该上限仍全 200。
+    回归用:防有人把 enforce_rate_limit 挪到测试账号 early-return 之前而破坏豁免。"""
+    from app.api.v1 import auth
+    from app.core import ratelimit
+
+    monkeypatch.setattr(ratelimit.settings, "RATE_LIMIT_ENABLED", True)
+    ratelimit._buckets.clear()  # 隔离:清掉跨用例累加的内存计数
+
+    # 连登 (每小时上限 + 3) 次,远超普通号会被拦的阈值;测试号豁免 → 全部放行
+    for i in range(auth.SMS_LOGIN_MAX_PER_HOUR + 3):
+        r = client.post("/api/v1/auth/sms/login", json={"phone": TEST_PHONE, "code": "000000"})
+        assert r.status_code == 200, (
+            f"测试号第 {i + 1} 次登录应放行(豁免每小时限流),实际 {r.status_code}: {r.text}"
+        )
+
+
 # ============================ 计数单元逻辑 ============================
 
 def test_quota_resets_across_day(enabled, monkeypatch) -> None: