100 lines
3.0 KiB
Python
100 lines
3.0 KiB
Python
import sqlite3
|
|
|
|
from src.login_attempt_store import LoginAttemptStore, client_source
|
|
|
|
|
|
def test_forwarded_client_is_used_only_for_explicitly_trusted_proxies():
|
|
trusted = "127.0.0.0/8, 10.0.0.0/8"
|
|
|
|
assert client_source("198.51.100.4", "203.0.113.7", trusted) == "198.51.100.4"
|
|
assert (
|
|
client_source("127.0.0.1", "203.0.113.7, 10.1.2.3", trusted)
|
|
== "203.0.113.7"
|
|
)
|
|
assert client_source("127.0.0.1", "not-an-ip", trusted) == "127.0.0.1"
|
|
|
|
|
|
def test_failure_budget_is_shared_across_store_instances_and_expires(tmp_path):
|
|
now = [1_000.0]
|
|
database = tmp_path / "login-attempts.sqlite3"
|
|
first = LoginAttemptStore(
|
|
database,
|
|
clock=lambda: now[0],
|
|
max_failures=3,
|
|
window_seconds=60,
|
|
)
|
|
second = LoginAttemptStore(
|
|
database,
|
|
clock=lambda: now[0],
|
|
max_failures=3,
|
|
window_seconds=60,
|
|
)
|
|
|
|
assert first.retry_after("203.0.113.7") == 0
|
|
first.record_failure("203.0.113.7")
|
|
second.record_failure("203.0.113.7")
|
|
first.record_failure("203.0.113.7")
|
|
|
|
assert second.retry_after("203.0.113.7") == 60
|
|
now[0] = 1_060.0
|
|
assert first.retry_after("203.0.113.7") == 0
|
|
|
|
|
|
def test_recording_a_failure_prunes_expired_source_records(tmp_path):
|
|
now = [1_000.0]
|
|
store = LoginAttemptStore(
|
|
tmp_path / "login-attempts.sqlite3",
|
|
clock=lambda: now[0],
|
|
max_failures=3,
|
|
window_seconds=60,
|
|
)
|
|
store.record_failure("203.0.113.1")
|
|
store.record_failure("203.0.113.2")
|
|
|
|
now[0] = 1_060.0
|
|
store.record_failure("203.0.113.3")
|
|
|
|
with sqlite3.connect(store.path) as connection:
|
|
assert connection.execute("SELECT COUNT(*) FROM login_attempts").fetchone() == (1,)
|
|
|
|
|
|
def test_failure_ledger_evicts_oldest_sources_at_its_size_limit(tmp_path):
|
|
store = LoginAttemptStore(
|
|
tmp_path / "login-attempts.sqlite3",
|
|
clock=lambda: 1_000.0,
|
|
max_failures=3,
|
|
window_seconds=60,
|
|
max_entries=2,
|
|
)
|
|
|
|
for address in ("203.0.113.1", "203.0.113.2", "203.0.113.3"):
|
|
store.record_failure(address)
|
|
|
|
with sqlite3.connect(store.path) as connection:
|
|
assert connection.execute("SELECT COUNT(*) FROM login_attempts").fetchone() == (2,)
|
|
|
|
|
|
def test_named_admission_budget_is_atomic_across_instances_and_source_scoped(tmp_path):
|
|
now = [1_000.0]
|
|
database = tmp_path / "login-attempts.sqlite3"
|
|
first = LoginAttemptStore(
|
|
database,
|
|
clock=lambda: now[0],
|
|
max_failures=3,
|
|
window_seconds=60,
|
|
)
|
|
second = LoginAttemptStore(
|
|
database,
|
|
clock=lambda: now[0],
|
|
max_failures=3,
|
|
window_seconds=60,
|
|
)
|
|
|
|
assert first.admit("passkey_options", "203.0.113.7", limit=2) == 0
|
|
assert second.admit("passkey_options", "203.0.113.7", limit=2) == 0
|
|
assert first.admit("passkey_options", "203.0.113.7", limit=2) == 60
|
|
assert first.admit("passkey_options", "203.0.113.8", limit=2) == 0
|
|
|
|
now[0] = 1_060.0
|
|
assert second.admit("passkey_options", "203.0.113.7", limit=2) == 0
|