94 lines
3.5 KiB
Python
94 lines
3.5 KiB
Python
"""P0-5 · 登录防爆破 + API Key 限流回归。
|
||
|
||
- 主登录 `/api/v1/auth/token/`：5 次失败后锁定 15 分钟（429），正确密码也被锁；
|
||
成功登录清零（正常用户不受影响）。
|
||
- 商城登录 `/api/v1/storefront/login/`：同策略（按 phone + IP）。
|
||
- API Key：分钟窗口超 `rate_limit` → 429；`rate_limit<=0` 下界为 1。
|
||
"""
|
||
|
||
import pytest
|
||
from django.core.cache import cache
|
||
from rest_framework.test import APIClient
|
||
|
||
from apps.core import ratelimit as rl
|
||
from apps.openapi.models import APIKey
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _clean_cache():
|
||
cache.clear()
|
||
yield
|
||
cache.clear()
|
||
|
||
|
||
def test_main_login_locks_after_5_failures(db, django_user_model):
|
||
django_user_model.objects.create_user("victim", password="correct-pw")
|
||
c = APIClient()
|
||
for _ in range(4):
|
||
r = c.post("/api/v1/auth/token/",
|
||
{"username": "victim", "password": "wrong"},
|
||
format="json")
|
||
assert r.status_code == 401, r.content
|
||
# 第 5 次失败达阈值直接锁定（429），后续正确密码也被锁
|
||
fifth = c.post("/api/v1/auth/token/",
|
||
{"username": "victim", "password": "wrong"},
|
||
format="json")
|
||
assert fifth.status_code == 429, fifth.content
|
||
assert fifth.json()["code"] == "login_locked"
|
||
locked = c.post("/api/v1/auth/token/",
|
||
{"username": "victim", "password": "wrong"},
|
||
format="json")
|
||
assert locked.status_code == 429, locked.content
|
||
assert locked.json()["code"] == "login_locked"
|
||
# 正确密码也被锁
|
||
good = c.post("/api/v1/auth/token/",
|
||
{"username": "victim", "password": "correct-pw"},
|
||
format="json")
|
||
assert good.status_code == 429, good.content
|
||
|
||
|
||
def test_main_login_success_resets_counter(db, django_user_model):
|
||
django_user_model.objects.create_user("normal", password="pw123456")
|
||
c = APIClient()
|
||
for _ in range(4):
|
||
c.post("/api/v1/auth/token/",
|
||
{"username": "normal", "password": "wrong"}, format="json")
|
||
ok = c.post("/api/v1/auth/token/",
|
||
{"username": "normal", "password": "pw123456"}, format="json")
|
||
assert ok.status_code == 200, ok.content
|
||
# 计数已清：再错 4 次仍是 401 而非 429
|
||
for _ in range(4):
|
||
r = c.post("/api/v1/auth/token/",
|
||
{"username": "normal", "password": "wrong"}, format="json")
|
||
assert r.status_code == 401, r.content
|
||
|
||
|
||
def test_apikey_rate_limit_enforced(db, tenant):
|
||
from django.contrib.auth import get_user_model
|
||
|
||
User = get_user_model()
|
||
u = User.objects.create_user("keyboss", password="x")
|
||
key_obj, raw = APIKey.generate(
|
||
tenant=tenant, name="limited", scopes=["products:read"], created_by=u,
|
||
)
|
||
key_obj.rate_limit = 3
|
||
key_obj.save(update_fields=["rate_limit"])
|
||
|
||
c = APIClient()
|
||
c.credentials(HTTP_X_API_KEY=raw)
|
||
for _ in range(3):
|
||
r = c.get("/api/v1/openapi/v1/products/")
|
||
assert r.status_code == 200, r.content
|
||
over = c.get("/api/v1/openapi/v1/products/")
|
||
assert over.status_code == 429, over.content
|
||
# DRF 只在 exc.wait 非空时才回 Retry-After；不传 wait 只有 429 无头。
|
||
assert "Retry-After" in over, dict(over.items())
|
||
|
||
|
||
def test_effective_rate_limit_floor():
|
||
assert rl.effective_rate_limit(0) == 1
|
||
assert rl.effective_rate_limit(-5) == 1
|
||
assert rl.effective_rate_limit(None) == 1
|
||
assert rl.effective_rate_limit("abc") == 1
|
||
assert rl.effective_rate_limit(120) == 120
|