103 lines
3.9 KiB
Python
103 lines
3.9 KiB
Python
"""登录失败限流单元测试。
|
|
|
|
守的是：锁定阈值与**滑动窗口**语义。若记录失败时不刷新 TTL，
|
|
「4 次失败 + 等 14 分钟 + 1 次失败」就不会触发锁定，
|
|
正好给慢速撞库留出缝隙。
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from app.core.config import get_settings
|
|
from app.core.constants import REDIS_KEY_LOGIN_FAIL
|
|
from app.services import login_limiter
|
|
|
|
|
|
def _ttl(username: str) -> int:
|
|
from app.core.redis_client import get_redis
|
|
|
|
return get_redis().ttl(REDIS_KEY_LOGIN_FAIL.format(username=username))
|
|
|
|
|
|
class TestThreshold:
|
|
"""阈值判定。"""
|
|
|
|
def test_初始状态未锁定且计数为零(self, fake_redis) -> None:
|
|
assert login_limiter.failure_count("u") == 0
|
|
assert login_limiter.is_locked("u") is False
|
|
|
|
def test_未达阈值时不锁定(self, fake_redis) -> None:
|
|
for _ in range(get_settings().login_fail_limit - 1):
|
|
login_limiter.record_failure("u")
|
|
assert login_limiter.is_locked("u") is False
|
|
|
|
def test_达到阈值即锁定(self, fake_redis) -> None:
|
|
for _ in range(get_settings().login_fail_limit):
|
|
login_limiter.record_failure("u")
|
|
assert login_limiter.is_locked("u") is True
|
|
|
|
def test_不同用户之间互不影响(self, fake_redis) -> None:
|
|
"""按用户名隔离，避免一个账号被攻击导致全员被锁。"""
|
|
for _ in range(get_settings().login_fail_limit):
|
|
login_limiter.record_failure("victim")
|
|
assert login_limiter.is_locked("victim") is True
|
|
assert login_limiter.is_locked("others") is False
|
|
|
|
|
|
class TestSlidingWindow:
|
|
"""滑动窗口：每次失败都刷新过期时间。"""
|
|
|
|
def test_锁定时长与配置一致(self, fake_redis) -> None:
|
|
"""容忍 1 秒偏差：Redis 的 TTL 向下取整，刚写入就查询可能少 1 秒。"""
|
|
login_limiter.record_failure("u")
|
|
expected = get_settings().login_lock_minutes * 60
|
|
assert _ttl("u") in (expected, expected - 1)
|
|
|
|
def test_再次失败会刷新过期时间(self, fake_redis) -> None:
|
|
"""把 TTL 改小模拟「距上次失败已过 14 分钟」，再失败一次应重新计满。
|
|
|
|
若不刷新，此处 TTL 仍是改小后的值，"慢速撞库"就能绕过锁定。
|
|
"""
|
|
login_limiter.record_failure("u")
|
|
fake_redis.expire(REDIS_KEY_LOGIN_FAIL.format(username="u"), 10)
|
|
|
|
login_limiter.record_failure("u")
|
|
assert _ttl("u") > 600
|
|
|
|
|
|
class TestAtomicWrite:
|
|
"""计数和 TTL 必须同事务生效。"""
|
|
|
|
def test_EXEC失败时不留下无TTL计数(self, fake_redis, monkeypatch) -> None:
|
|
key = REDIS_KEY_LOGIN_FAIL.format(username="u")
|
|
|
|
pipeline = fake_redis.pipeline(transaction=True)
|
|
|
|
def fail_execute():
|
|
raise RuntimeError("EXEC 失败")
|
|
|
|
monkeypatch.setattr(pipeline, "execute", fail_execute)
|
|
monkeypatch.setattr(fake_redis, "pipeline", lambda **_: pipeline)
|
|
with pytest.raises(RuntimeError, match="EXEC 失败"):
|
|
login_limiter.record_failure("u")
|
|
assert fake_redis.get(key) is None
|
|
|
|
|
|
class TestClearAndRemaining:
|
|
"""解锁与剩余时长。"""
|
|
|
|
def test_登录成功清除失败计数(self, fake_redis) -> None:
|
|
for _ in range(get_settings().login_fail_limit):
|
|
login_limiter.record_failure("u")
|
|
login_limiter.clear("u")
|
|
assert login_limiter.is_locked("u") is False
|
|
assert login_limiter.failure_count("u") == 0
|
|
|
|
def test_未锁定时剩余秒数为零(self, fake_redis) -> None:
|
|
assert login_limiter.remaining_lock_seconds("u") == 0
|
|
|
|
def test_锁定后剩余秒数为正且不超过锁定时长(self, fake_redis) -> None:
|
|
for _ in range(get_settings().login_fail_limit):
|
|
login_limiter.record_failure("u")
|
|
remaining = login_limiter.remaining_lock_seconds("u")
|
|
assert 0 < remaining <= get_settings().login_lock_minutes * 60
|