78 lines
2.9 KiB
Python
78 lines
2.9 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import time
|
||
|
|
|
||
|
|
from yuxi.channels.adapters.nextcloudtalk.webhook_server import _AuthRateLimiter
|
||
|
|
|
||
|
|
|
||
|
|
class TestAuthRateLimiter:
|
||
|
|
def test_initial_state_allows_access(self):
|
||
|
|
limiter = _AuthRateLimiter(max_requests=3, window_s=60, lockout_s=300)
|
||
|
|
assert limiter.check("192.168.1.1") is True
|
||
|
|
|
||
|
|
def test_after_max_failures_blocks(self):
|
||
|
|
limiter = _AuthRateLimiter(max_requests=3, window_s=60, lockout_s=5)
|
||
|
|
for _ in range(3):
|
||
|
|
limiter.record_failure("192.168.1.1")
|
||
|
|
assert limiter.check("192.168.1.1") is False
|
||
|
|
|
||
|
|
def test_lockout_reattempt(self):
|
||
|
|
limiter = _AuthRateLimiter(max_requests=2, window_s=60, lockout_s=1)
|
||
|
|
for _ in range(2):
|
||
|
|
limiter.record_failure("10.0.0.1")
|
||
|
|
assert limiter.check("10.0.0.1") is False
|
||
|
|
time.sleep(1.1)
|
||
|
|
assert limiter.check("10.0.0.1") is False
|
||
|
|
|
||
|
|
def test_different_ips_independent(self):
|
||
|
|
limiter = _AuthRateLimiter(max_requests=2, window_s=60, lockout_s=5)
|
||
|
|
for _ in range(2):
|
||
|
|
limiter.record_failure("10.0.0.1")
|
||
|
|
assert limiter.check("10.0.0.1") is False
|
||
|
|
assert limiter.check("10.0.0.2") is True
|
||
|
|
|
||
|
|
def test_stats(self):
|
||
|
|
limiter = _AuthRateLimiter(max_requests=3, window_s=60, lockout_s=5)
|
||
|
|
limiter.record_failure("10.0.0.1")
|
||
|
|
stats = limiter.stats()
|
||
|
|
assert stats["tracked_ips"] == 1
|
||
|
|
assert stats["locked_ips"] == 0
|
||
|
|
|
||
|
|
def test_stats_after_lockout(self):
|
||
|
|
limiter = _AuthRateLimiter(max_requests=2, window_s=60, lockout_s=5)
|
||
|
|
for _ in range(2):
|
||
|
|
limiter.record_failure("10.0.0.1")
|
||
|
|
limiter.check("10.0.0.1")
|
||
|
|
stats = limiter.stats()
|
||
|
|
assert stats["locked_ips"] == 1
|
||
|
|
|
||
|
|
def test_gc_cleans_expired_lockouts(self):
|
||
|
|
limiter = _AuthRateLimiter(max_requests=2, window_s=60, lockout_s=1)
|
||
|
|
for _ in range(2):
|
||
|
|
limiter.record_failure("10.0.0.1")
|
||
|
|
limiter.check("10.0.0.1")
|
||
|
|
time.sleep(1.1)
|
||
|
|
stats = limiter.stats()
|
||
|
|
assert stats["locked_ips"] == 0
|
||
|
|
|
||
|
|
def test_gc_cleans_expired_failures(self):
|
||
|
|
limiter = _AuthRateLimiter(max_requests=3, window_s=1, lockout_s=5)
|
||
|
|
limiter.record_failure("10.0.0.1")
|
||
|
|
time.sleep(1.1)
|
||
|
|
limiter.check("10.0.0.1")
|
||
|
|
stats = limiter.stats()
|
||
|
|
assert stats["tracked_ips"] == 0
|
||
|
|
|
||
|
|
def test_multiple_failures_in_window(self):
|
||
|
|
limiter = _AuthRateLimiter(max_requests=5, window_s=60, lockout_s=5)
|
||
|
|
for _ in range(4):
|
||
|
|
limiter.record_failure("10.0.0.1")
|
||
|
|
assert limiter.check("10.0.0.1") is True
|
||
|
|
|
||
|
|
def test_record_failure_timestamps(self):
|
||
|
|
limiter = _AuthRateLimiter(max_requests=3, window_s=60, lockout_s=5)
|
||
|
|
limiter.record_failure("10.0.0.1")
|
||
|
|
time.sleep(0.01)
|
||
|
|
limiter.record_failure("10.0.0.1")
|
||
|
|
limiter.record_failure("10.0.0.1")
|
||
|
|
assert limiter.check("10.0.0.1") is False
|