feat(auth): 发码也加同设备同IP每小时限流 — 堵换号绕开单号冷却的洞

/sms/send 增加 (device_id + IP) 每小时最多 10 次发码限流,放在真发之前(超限不烧短信)。
- 原单号 60s 冷却 / 单号每日上限都是单号维度,一机换手机号即可绕开 → 设备维度按机器封顶,挡短信轰炸/烧钱。
- 与路由上 IP 维度(10次/分钟)互补;测试号豁免。
- SmsSendRequest 加 device_id(空=按 IP 聚一桶);补回归测试。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
zzhyyyyy
2026-06-26 12:01:26 +08:00
parent a91a04c45d
commit 9ff91b81ee
3 changed files with 55 additions and 1 deletions
+17 -1
View File
@@ -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:
+4
View File
@@ -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):
+34
View File
@@ -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(内存计数跨用例累加),本用例临时打开并清空计数隔离。"""