45 lines
1.9 KiB
Python
45 lines
1.9 KiB
Python
"""P0-5 · 登录防爆破的 TokenObtainPairView 包装。
|
|
|
|
按 (username + IP) 记连续失败，5 次失败锁 15 分钟：
|
|
- 请求前已锁定 → 直接 429（即使密码正确也不再校验，避免计时侧信道）。
|
|
- 200 → 清零；401 → 失败 +1（达阈值加锁）。
|
|
"""
|
|
|
|
from rest_framework import status
|
|
from rest_framework.response import Response
|
|
from rest_framework_simplejwt.views import TokenObtainPairView
|
|
|
|
from apps.core import ratelimit as rl
|
|
|
|
|
|
class RateLimitedTokenObtainPairView(TokenObtainPairView):
|
|
def post(self, request, *args, **kwargs):
|
|
from rest_framework.exceptions import AuthenticationFailed
|
|
|
|
username = (request.data.get("username") or "").strip()
|
|
ip = rl.client_ip(request)
|
|
if username and rl.is_login_locked(username, ip):
|
|
return Response(
|
|
{"code": "login_locked", "detail": "登录失败次数过多，账号已临时锁定 15 分钟"},
|
|
status=status.HTTP_429_TOO_MANY_REQUESTS,
|
|
)
|
|
try:
|
|
resp = super().post(request, *args, **kwargs)
|
|
except AuthenticationFailed:
|
|
# simplejwt 凭证错误走抛异常（DRF 转 401），这里计失败数；
|
|
# 达阈值后本次直接 429，避免再给一次尝试机会。
|
|
if username:
|
|
fails = rl.record_login_failure(username, ip)
|
|
if fails >= rl.LOGIN_MAX_FAILURES:
|
|
return Response(
|
|
{"code": "login_locked",
|
|
"detail": "登录失败次数过多，账号已临时锁定 15 分钟"},
|
|
status=status.HTTP_429_TOO_MANY_REQUESTS,
|
|
)
|
|
raise
|
|
if not username:
|
|
return resp
|
|
if resp.status_code == status.HTTP_200_OK:
|
|
rl.clear_login_failures(username, ip)
|
|
return resp
|