From 41a000304e1848d8f3d9418f815665cf29f2bfe7 Mon Sep 17 00:00:00 2001 From: zzhyyyyy <2685922758@qq.com> Date: Thu, 25 Jun 2026 17:28:59 +0800 Subject: [PATCH 1/4] =?UTF-8?q?feat(auth):=20=E7=9F=AD=E4=BF=A1=E7=99=BB?= =?UTF-8?q?=E5=BD=95=E9=98=B2=E5=88=B7=20=E2=80=94=20=E5=90=8C=E6=89=8B?= =?UTF-8?q?=E6=9C=BA=E5=8F=B7=E5=90=8CIP=20=E6=AF=8F=E5=B0=8F=E6=97=B6?= =?UTF-8?q?=E9=99=90=205=20=E6=AC=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit /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) --- app/api/v1/auth.py | 21 ++++++++++++++++++--- app/core/ratelimit.py | 26 ++++++++++++++++++++++++++ tests/test_auth.py | 24 ++++++++++++++++++++++++ tests/test_test_account.py | 22 ++++++++++++++++++++++ 4 files changed, 90 insertions(+), 3 deletions(-) diff --git a/app/api/v1/auth.py b/app/api/v1/auth.py index d5d21c2..07bb966 100644 --- a/app/api/v1/auth.py +++ b/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") diff --git a/app/core/ratelimit.py b/app/core/ratelimit.py index ce64f06..cff6545 100644 --- a/app/core/ratelimit.py +++ b/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, + ) diff --git a/tests/test_auth.py b/tests/test_auth.py index 390f488..539dfaa 100644 --- a/tests/test_auth.py +++ b/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" diff --git a/tests/test_test_account.py b/tests/test_test_account.py index 721ef56..15e0b12 100644 --- a/tests/test_test_account.py +++ b/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: -- 2.52.0 From a91a04c45dde08849f2eeb347976608480edfa02 Mon Sep 17 00:00:00 2001 From: zzhyyyyy <2685922758@qq.com> Date: Fri, 26 Jun 2026 10:55:36 +0800 Subject: [PATCH 2/4] =?UTF-8?q?fix(auth):=20=E7=99=BB=E5=BD=95=E9=99=90?= =?UTF-8?q?=E6=B5=81=E6=94=B9=E6=8C=89=E8=AE=BE=E5=A4=87(device=5Fid)?= =?UTF-8?q?=E8=80=8C=E9=9D=9E=E6=89=8B=E6=9C=BA=E5=8F=B7=20=E2=80=94=20?= =?UTF-8?q?=E9=98=B2=E4=B8=80=E6=9C=BA=E7=8B=82=E7=99=BB=E5=A4=9A=E5=8F=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit sms-login 防刷限流键从 (手机号+IP) 改为 (device_id+IP):同一台机器换不同手机号刷登录 也受同一桶约束(5次/小时,成功/失败都计);device_id 空(老客户端)时退化为该 IP 聚一桶,仍受限。 测试号仍豁免。回归测试更新为设备维度(同设备换号仍被拦、换设备独立放行)。 Co-Authored-By: Claude Opus 4.8 (1M context) --- app/api/v1/auth.py | 14 ++++++++------ tests/test_auth.py | 27 ++++++++++++++++++--------- tests/test_test_account.py | 15 +++++++++------ 3 files changed, 35 insertions(+), 21 deletions(-) diff --git a/app/api/v1/auth.py b/app/api/v1/auth.py index 07bb966..bcd4b2b 100644 --- a/app/api/v1/auth.py +++ b/app/api/v1/auth.py @@ -38,7 +38,7 @@ logger = logging.getLogger("shagua.auth") router = APIRouter(prefix="/api/v1/auth", tags=["auth"]) -# 手机号登录防刷:同一手机号 + 同一 IP 每小时最多的登录尝试次数(成功/失败都计)。 +# 手机号登录防刷:同一设备(device_id) + 同一 IP 每小时最多的登录尝试次数(成功/失败都计)。 SMS_LOGIN_MAX_PER_HOUR = 5 @@ -115,7 +115,7 @@ def sms_send(req: SmsSendRequest) -> SmsSendResponse: def sms_login(req: SmsLoginRequest, request: Request, db: DbSession) -> TokenWithUser: # 测试账号:免验证码登录 + 每日上限 + 每次都走新手引导(详见 app/core/test_account.py)。 # 放在最前面:命中即不校验验证码;先扣当日额度,超限直接拒,挡住有人猜到号后脚本刷。 - # 测试账号走自己的每日额度、不受下面 (手机号+IP) 每小时限流约束(QA 需在一小时内反复登录联调)。 + # 测试账号走自己的每日额度、不受下面 (设备+IP) 每小时限流约束(QA 需在一小时内反复登录联调)。 if test_account.is_test_account(req.phone): if not test_account.try_consume_quota(): raise HTTPException(status_code=429, detail="测试账号今日使用次数已达上限,请明天再试") @@ -127,12 +127,14 @@ def sms_login(req: SmsLoginRequest, request: Request, db: DbSession) -> TokenWit # 不读/不写 onboarding_completion 表。 return _login_response(user, onboarding_completed=False, force_onboarding=True) - # 防刷:同一手机号 + 同一 IP 每小时最多 SMS_LOGIN_MAX_PER_HOUR 次登录尝试。放在验证码校验**之前** - # → 输错验证码的失败尝试也计数,才挡得住撞库/爆破。与路由上 IP 维度的 sms-login 限流(同 IP 多号)互补。 + # 防刷:同一设备(device_id) + 同一 IP 每小时最多 SMS_LOGIN_MAX_PER_HOUR 次登录尝试。放在验证码校验 + # **之前** → 输错验证码的失败尝试也计数,才挡得住撞库/爆破。与路由上 IP 维度的 sms-login 限流(同 IP)互补。 + # ⚠️ 按设备而非手机号 → 一台机器换不同手机号刷登录也受限(防一机狂登多号);device_id 空(老客户端)时 + # 退化为该 IP 下所有空设备聚一桶,仍受限。 enforce_rate_limit( request, - scope="sms-login-phone", - subject=req.phone, + scope="sms-login-device", + subject=req.device_id, limit=SMS_LOGIN_MAX_PER_HOUR, window_sec=3600, detail="登录尝试过于频繁,请稍后再试", diff --git a/tests/test_auth.py b/tests/test_auth.py index 539dfaa..27eaf43 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -74,8 +74,8 @@ 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。 +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 @@ -83,18 +83,27 @@ def test_sms_login_phone_ip_rate_limit(client, monkeypatch) -> None: monkeypatch.setattr(ratelimit.settings, "RATE_LIMIT_ENABLED", True) ratelimit._buckets.clear() # 隔离:清掉跨用例累加的内存计数 - phone = "13533153300" - # 前 N 次:mock 校验放行(任意 6 位数字)→ 200 登录成功 + 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": phone, "code": "123456"}) + 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": phone, "code": "123456"}) + # 第 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 换个手机号 → 独立计数(限流键含手机号、非纯 IP),仍放行 - r = client.post("/api/v1/auth/sms/login", json={"phone": "13533153301", "code": "123456"}) + # 同 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 diff --git a/tests/test_test_account.py b/tests/test_test_account.py index 15e0b12..2637de2 100644 --- a/tests/test_test_account.py +++ b/tests/test_test_account.py @@ -144,11 +144,11 @@ def test_send_does_not_consume_quota(client, enabled, monkeypatch) -> None: # ============================ 登录限流豁免 ============================ -def test_exempt_from_phone_ip_login_rate_limit(client, enabled, monkeypatch) -> None: - """测试号豁免 (手机号+IP) 每小时登录限流。 +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_phone_ip_rate_limit); - 测试号走自己的每日额度、不受这道限流——这里临时打开 RATE_LIMIT_ENABLED,连登远超该上限仍全 200。 + 普通设备同设备同 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 @@ -156,9 +156,12 @@ def test_exempt_from_phone_ip_login_rate_limit(client, enabled, monkeypatch) -> monkeypatch.setattr(ratelimit.settings, "RATE_LIMIT_ENABLED", True) ratelimit._buckets.clear() # 隔离:清掉跨用例累加的内存计数 - # 连登 (每小时上限 + 3) 次,远超普通号会被拦的阈值;测试号豁免 → 全部放行 + # 同一设备连登 (每小时上限 + 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"}) + 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}" ) -- 2.52.0 From 9ff91b81ee73fbdafd6da7cb57155d2551bea3c4 Mon Sep 17 00:00:00 2001 From: zzhyyyyy <2685922758@qq.com> Date: Fri, 26 Jun 2026 12:01:26 +0800 Subject: [PATCH 3/4] =?UTF-8?q?feat(auth):=20=E5=8F=91=E7=A0=81=E4=B9=9F?= =?UTF-8?q?=E5=8A=A0=E5=90=8C=E8=AE=BE=E5=A4=87=E5=90=8CIP=E6=AF=8F?= =?UTF-8?q?=E5=B0=8F=E6=97=B6=E9=99=90=E6=B5=81=20=E2=80=94=20=E5=A0=B5?= =?UTF-8?q?=E6=8D=A2=E5=8F=B7=E7=BB=95=E5=BC=80=E5=8D=95=E5=8F=B7=E5=86=B7?= =?UTF-8?q?=E5=8D=B4=E7=9A=84=E6=B4=9E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit /sms/send 增加 (device_id + IP) 每小时最多 10 次发码限流,放在真发之前(超限不烧短信)。 - 原单号 60s 冷却 / 单号每日上限都是单号维度,一机换手机号即可绕开 → 设备维度按机器封顶,挡短信轰炸/烧钱。 - 与路由上 IP 维度(10次/分钟)互补;测试号豁免。 - SmsSendRequest 加 device_id(空=按 IP 聚一桶);补回归测试。 Co-Authored-By: Claude Opus 4.8 (1M context) --- app/api/v1/auth.py | 18 +++++++++++++++++- app/schemas/auth.py | 4 ++++ tests/test_auth.py | 34 ++++++++++++++++++++++++++++++++++ 3 files changed, 55 insertions(+), 1 deletion(-) diff --git a/app/api/v1/auth.py b/app/api/v1/auth.py index bcd4b2b..1ad66ab 100644 --- a/app/api/v1/auth.py +++ b/app/api/v1/auth.py @@ -40,6 +40,9 @@ 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( @@ -89,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: diff --git a/app/schemas/auth.py b/app/schemas/auth.py index 3133de0..3870a52 100644 --- a/app/schemas/auth.py +++ b/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): diff --git a/tests/test_auth.py b/tests/test_auth.py index 27eaf43..6288b11 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -74,6 +74,40 @@ 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(内存计数跨用例累加),本用例临时打开并清空计数隔离。""" -- 2.52.0 From 5b74a8507dbe3692ab656b94a82a4fb59b7a3bb4 Mon Sep 17 00:00:00 2001 From: zzhyyyyy <2685922758@qq.com> Date: Fri, 26 Jun 2026 12:40:58 +0800 Subject: [PATCH 4/4] =?UTF-8?q?test(auth):=20=E6=B5=8B=E8=AF=95=E5=8F=B7?= =?UTF-8?q?=E8=B1=81=E5=85=8D=E5=8F=91=E7=A0=81=E9=99=90=E6=B5=81=E7=9A=84?= =?UTF-8?q?=E5=9B=9E=E5=BD=92=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 锁定「测试号(.env TEST_ACCOUNT_PHONE)不受 /sms/send 设备限流约束」—— 防有人把发码限流挪到测试账号 early-return 之前,把 QA 联调也风控了。 Co-Authored-By: Claude Opus 4.8 (1M context) --- tests/test_test_account.py | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/tests/test_test_account.py b/tests/test_test_account.py index 2637de2..2155a05 100644 --- a/tests/test_test_account.py +++ b/tests/test_test_account.py @@ -167,6 +167,29 @@ def test_exempt_from_device_ip_login_rate_limit(client, enabled, monkeypatch) -> ) +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: -- 2.52.0