test(auth): 测试号豁免发码限流的回归测试

锁定「测试号(.env TEST_ACCOUNT_PHONE)不受 /sms/send 设备限流约束」——
防有人把发码限流挪到测试账号 early-return 之前,把 QA 联调也风控了。

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

alembic/versions/bcfcaf07152b_merge_meituan_coupon_image_meta_and_.py

@@ -0,0 +1,26 @@
+"""merge meituan_coupon_image_meta and comparison/ad heads
+
+Revision ID: bcfcaf07152b
+Revises: 4dc2af7ebe74, meituan_coupon_image_meta
+Create Date: 2026-06-24 20:34:09.475641
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'bcfcaf07152b'
+down_revision: Union[str, Sequence[str], None] = ('4dc2af7ebe74', 'meituan_coupon_image_meta')
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    pass
+
+
+def downgrade() -> None:
+    pass

app/admin/schemas/ad_config.py

@@ -15,6 +15,7 @@ class AdConfigOut(BaseModel):
     reward_enabled: bool
     compare_ad_enabled: bool
     coupon_ad_enabled: bool
+    withdrawal_ad_enabled: bool
 
 
 class AdConfigUpdate(BaseModel):
@@ -28,3 +29,4 @@ class AdConfigUpdate(BaseModel):
     reward_enabled: bool | None = None
     compare_ad_enabled: bool | None = None
     coupon_ad_enabled: bool | None = None
+    withdrawal_ad_enabled: bool | None = None

app/api/v1/ad.py

@@ -53,6 +53,9 @@ REASON_UNKNOWN_USER = 2  # user_id 不存在(可能伪造)
 
 REWARD_SCENE_REWARD_VIDEO = "reward_video"
 REWARD_SCENE_SIGNIN_BOOST = "signin_boost"
+# 提现看视频:看完才能提现的「硬门槛」广告,**不发金币**,只记一条幂等记录(收益由 eCPM 上报口径
+# ad_type="withdrawal_video" 单独统计)。故意不放进 SUPPORTED_REWARD_SCENES——它不走发币分支。
+REWARD_SCENE_WITHDRAWAL_AD = "withdrawal_ad"
 SUPPORTED_REWARD_SCENES = {REWARD_SCENE_REWARD_VIDEO, REWARD_SCENE_SIGNIN_BOOST}
 
 
@@ -126,6 +129,19 @@ def pangle_callback(request: Request, db: DbSession) -> PangleCallbackOut:
         return PangleCallbackOut(is_verify=True, reason=REASON_OK)
 
     try:
+        if reward_scene == REWARD_SCENE_WITHDRAWAL_AD:
+            # 提现看视频:已知合法场景,只记一条幂等记录、**不动钱包**(发不发币由 reward_scene 决定)。
+            # is_verify=true 让穿山甲不重试(区别于 unknown_scene 的 false)。
+            crud_ad.record_external_reward(
+                db, user_id, trans_id, coin=0, reward_scene=REWARD_SCENE_WITHDRAWAL_AD,
+                ad_session_id=ad_session_id, ecpm=ecpm,
+                reward_name=params.get("reward_name"), raw=raw[:1024],
+                status="withdrawal_ad",
+            )
+            logger.info(
+                "pangle callback withdrawal_ad user_id=%d trans_id=%s", user_id, trans_id,
+            )
+            return PangleCallbackOut(is_verify=True, reason=REASON_OK)
         if reward_scene not in SUPPORTED_REWARD_SCENES:
             rec = crud_ad.record_external_reward(
                 db, user_id, trans_id, coin=0, reward_scene=reward_scene[:32],

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,12 @@ logger = logging.getLogger("shagua.auth")
 
 router = APIRouter(prefix="/api/v1/auth", tags=["auth"])
 
+# 手机号登录防刷:同一设备(device_id) + 同一 IP 每小时最多的登录尝试次数(成功/失败都计)。
+SMS_LOGIN_MAX_PER_HOUR = 5
+# 发码防刷:同一设备(device_id) + 同一 IP 每小时最多的发码次数。比登录略宽(发码含正常重发);
+# 堵「换手机号绕开单号 60s 冷却 / 单号每日上限」的洞 —— 那两道是单号维度,一机换号能绕开。
+SMS_SEND_MAX_PER_HOUR_PER_DEVICE = 10
+
 
 def _login_response(
     user, *, onboarding_completed: bool, force_onboarding: bool = False
@@ -86,13 +92,26 @@ def jverify_login(req: JverifyLoginRequest, db: DbSession) -> TokenWithUser:
     summary="发送短信验证码",
     dependencies=[Depends(rate_limit(10, 60, "sms-send"))],  # 同 IP 每分钟≤10 次(防一 IP 刷不同号)
 )
-def sms_send(req: SmsSendRequest) -> SmsSendResponse:
+def sms_send(req: SmsSendRequest, request: Request) -> SmsSendResponse:
     # 测试账号:不真发短信(号码非真实手机号,真发会失败/浪费),直接放行让客户端进入填码界面。
     # 真正的"免验证码"在 /sms/login 跳过校验;每日上限也在 login 处算(send 不耗额度)。
+    # (也不受下面设备发码限流约束:QA 联调要反复发码。)
     if test_account.is_test_account(req.phone):
         logger.info("test_account sms_send short-circuit (不真发)")
         return SmsSendResponse(sent=True, mock=True, cooldown_sec=0)
 
+    # 防刷:同一设备(device_id) + 同一 IP 每小时最多 SMS_SEND_MAX_PER_HOUR_PER_DEVICE 次发码。
+    # 补「换手机号绕开单号 60s 冷却 / 单号每日上限」的洞(那两道是单号维度,一机换号能绕);设备维度按机器封顶,
+    # 挡短信轰炸/烧钱。放在真发(send_code)之前 → 超限直接拦下、不真发短信。与路由上 IP 维度(10次/分钟)互补。
+    enforce_rate_limit(
+        request,
+        scope="sms-send-device",
+        subject=req.device_id,
+        limit=SMS_SEND_MAX_PER_HOUR_PER_DEVICE,
+        window_sec=3600,
+        detail="操作过于频繁,请稍后再试",
+    )
+
     try:
         cooldown = send_code(req.phone)
     except SmsError as e:
@@ -109,9 +128,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 +143,19 @@ def sms_login(req: SmsLoginRequest, db: DbSession) -> TokenWithUser:
         # 不读/不写 onboarding_completion 表。
         return _login_response(user, onboarding_completed=False, force_onboarding=True)
 
+    # 防刷:同一设备(device_id) + 同一 IP 每小时最多 SMS_LOGIN_MAX_PER_HOUR 次登录尝试。放在验证码校验
+    # **之前** → 输错验证码的失败尝试也计数,才挡得住撞库/爆破。与路由上 IP 维度的 sms-login 限流(同 IP)互补。
+    # ⚠️ 按设备而非手机号 → 一台机器换不同手机号刷登录也受限(防一机狂登多号);device_id 空(老客户端)时
+    #    退化为该 IP 下所有空设备聚一桶,仍受限。
+    enforce_rate_limit(
+        request,
+        scope="sms-login-device",
+        subject=req.device_id,
+        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/api/v1/cps_redirect.py

@@ -83,7 +83,10 @@ def cps_landing(code: str, request: Request, db: Session = Depends(get_db)):
         redirect_uri = f"{settings.CPS_REDIRECT_BASE.rstrip('/')}/wx/oauth/cb"
         auth_url = wx_oauth.build_authorize_url(redirect_uri, "snsapi_base", f"base:{code}")
         return RedirectResponse(auth_url, status_code=302)
-    _record(db, link, request, "visit", openid=openid)
+    # ?authed=1 是 userinfo 授权后的自动回跳重载,不是用户主动打开:visit 已在授权前那次
+    # 落地(base 静默授权回来那次)记过,这里再记会让每个走"领券授权"的用户 visit 虚增 1。
+    if request.query_params.get("authed") != "1":
+        _record(db, link, request, "visit", openid=openid)
     if link.platform == "taobao":
         activity = db.get(CpsActivity, link.activity_id)
         image_url = (activity.image_url if activity else None) or _DEFAULT_TAOBAO_IMAGE
@@ -201,8 +204,9 @@ function copyToken(){
   if(UINFO_URL){location.href=UINFO_URL;return}
   doCopy();
 }
-// userinfo 授权回流(?authed=1):自动复制,把"点领券→授权→复制"衔接成一步无感
-if(location.search.indexOf('authed=1')>=0){doCopy()}
+// userinfo 授权回流(?authed=1):自动复制,把"点领券→授权→复制"衔接成一步无感;
+// 复制后用 replaceState 清掉 authed=1,避免刷新/微信返回(BFCache)重载时重复触发复制上报。
+if(location.search.indexOf('authed=1')>=0){doCopy();history.replaceState(null,'',location.pathname)}
 </script>
 </body>
 </html>"""

app/api/v1/platform.py

@@ -68,6 +68,7 @@ def ad_config(db: DbSession) -> AdConfigPublicOut:
         reward_enabled=c["reward_enabled"],
         compare_ad_enabled=c["compare_ad_enabled"],
         coupon_ad_enabled=c["coupon_ad_enabled"],
+        withdrawal_ad_enabled=c["withdrawal_ad_enabled"],
     )
 
 

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,
+        )

app/repositories/app_config.py

@@ -109,6 +109,7 @@ _AD_CONFIG_DEFAULTS: dict[str, Any] = {
     "reward_enabled": True,               # 福利激励视频开关
     "compare_ad_enabled": True,           # 比价广告开关
     "coupon_ad_enabled": True,            # 领券广告开关
+    "withdrawal_ad_enabled": True,        # 提现激励视频开关(关=直接放行提现,不挡用户的钱)
 }
 
 

app/schemas/auth.py

@@ -71,6 +71,10 @@ class JverifyLoginRequest(BaseModel):
 
 class SmsSendRequest(BaseModel):
     phone: str = Field(..., min_length=11, max_length=11, pattern=r"^1\d{10}$")
+    device_id: str = Field(
+        "", max_length=64,
+        description="硬件级设备标识(Android ANDROID_ID),用于发码防刷按 设备+IP 限流;空=按 IP 聚一桶",
+    )
 
 
 class SmsSendResponse(BaseModel):

app/schemas/platform.py

@@ -43,6 +43,7 @@ class AdConfigPublicOut(BaseModel):
     reward_enabled: bool  # 福利激励视频开关
     compare_ad_enabled: bool  # 比价广告开关
     coupon_ad_enabled: bool  # 领券广告开关
+    withdrawal_ad_enabled: bool  # 提现激励视频开关(关=客户端直接放行提现)
 
 
 class AppVersionOut(BaseModel):

tests/test_auth.py

@@ -74,6 +74,73 @@ 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_send_device_ip_rate_limit(client, monkeypatch) -> None:
+    """发码防刷:同一设备(device_id) + 同一 IP 每小时最多 N 次发码,超出 429。
+    用不同手机号(绕开单号 60s 冷却)证明限流按设备封顶 —— 堵「换号绕开单号冷却/每日上限」的洞。
+    conftest 默认关限流;本用例临时打开 + 调小阈值(避开同 IP 10/分钟那道)+ 清计数隔离。"""
+    from app.api.v1 import auth
+    from app.core import ratelimit
+
+    monkeypatch.setattr(ratelimit.settings, "RATE_LIMIT_ENABLED", True)
+    monkeypatch.setattr(auth, "SMS_SEND_MAX_PER_HOUR_PER_DEVICE", 3)
+    ratelimit._buckets.clear()
+
+    device = "dev-send-A"
+    # 同设备、每次换不同手机号(绕开单号冷却)发码,前 3 次 200
+    for i in range(3):
+        r = client.post(
+            "/api/v1/auth/sms/send",
+            json={"phone": f"137001370{i:02d}", "device_id": device},
+        )
+        assert r.status_code == 200, f"第 {i + 1} 次应放行: {r.text}"
+    # 第 4 次:同设备超限 → 429(即便又换了手机号)
+    r = client.post(
+        "/api/v1/auth/sms/send",
+        json={"phone": "13700137003", "device_id": device},
+    )
+    assert r.status_code == 429, r.text
+
+    # 换个设备 → 独立计数(限流键含 device_id),放行
+    r = client.post(
+        "/api/v1/auth/sms/send",
+        json={"phone": "13700137004", "device_id": "dev-send-B"},
+    )
+    assert r.status_code == 200, r.text
+
+
+def test_sms_login_device_ip_rate_limit(client, monkeypatch) -> None:
+    """防刷:同一设备(device_id) + 同一 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()  # 隔离:清掉跨用例累加的内存计数
+
+    device = "dev-rl-A"
+    # 前 N 次故意每次换手机号、固定设备:mock 校验放行 → 200。证明限流按设备而非手机号,同设备共用一个桶
+    for i in range(auth.SMS_LOGIN_MAX_PER_HOUR):
+        r = client.post(
+            "/api/v1/auth/sms/login",
+            json={"phone": f"135331533{i:02d}", "code": "123456", "device_id": device},
+        )
+        assert r.status_code == 200, f"第 {i + 1} 次应放行: {r.text}"
+    # 第 N+1 次:同设备同 IP 超限 → 429(即便又换了个手机号)
+    r = client.post(
+        "/api/v1/auth/sms/login",
+        json={"phone": "13533153399", "code": "123456", "device_id": device},
+    )
+    assert r.status_code == 429, r.text
+    assert "频繁" in r.json()["detail"]
+
+    # 同 IP 换个设备 → 独立计数(限流键含 device_id、非纯 IP/手机号),仍放行
+    r = client.post(
+        "/api/v1/auth/sms/login",
+        json={"phone": "13533153300", "code": "123456", "device_id": "dev-rl-B"},
+    )
+    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,54 @@ 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_device_ip_login_rate_limit(client, enabled, monkeypatch) -> None:
+    """测试号豁免 (设备+IP) 每小时登录限流。
+
+    普通设备同设备同 IP 到 SMS_LOGIN_MAX_PER_HOUR 次即被拦(见 test_auth.test_sms_login_device_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", "device_id": "dev-test-exempt"},
+        )
+        assert r.status_code == 200, (
+            f"测试号第 {i + 1} 次登录应放行(豁免每小时限流),实际 {r.status_code}: {r.text}"
+        )
+
+
+def test_exempt_from_device_ip_send_rate_limit(client, enabled, monkeypatch) -> None:
+    """测试号豁免 (设备+IP) 每小时**发码**限流(与登录那道同样靠早返回豁免)。
+
+    普通设备同设备到 SMS_SEND_MAX_PER_HOUR_PER_DEVICE 次发码即被拦(见 test_auth.test_sms_send_device_ip_rate_limit);
+    测试号发码本就短路(不真发)、且在限流之前早返回 → 同一设备连发远超上限仍全 200。
+    回归用:防有人把发码限流挪到测试账号 early-return 之前而把 QA 联调也风控了。"""
+    from app.api.v1 import auth
+    from app.core import ratelimit
+
+    monkeypatch.setattr(ratelimit.settings, "RATE_LIMIT_ENABLED", True)
+    monkeypatch.setattr(auth, "SMS_SEND_MAX_PER_HOUR_PER_DEVICE", 2)  # 调小:不豁免的话第 3 次就该被拦
+    ratelimit._buckets.clear()
+
+    for i in range(auth.SMS_SEND_MAX_PER_HOUR_PER_DEVICE + 2):
+        r = client.post(
+            "/api/v1/auth/sms/send",
+            json={"phone": TEST_PHONE, "device_id": "dev-test-send"},
+        )
+        assert r.status_code == 200, (
+            f"测试号第 {i + 1} 次发码应放行(豁免每小时限流),实际 {r.status_code}: {r.text}"
+        )
+
+
 # ============================ 计数单元逻辑 ============================
 
 def test_quota_resets_across_day(enabled, monkeypatch) -> None: