feat: add audit, throttle and session foundations
Audit entries written only inside the caller's transaction, database throttle counters that refuse to run inside a transaction that could roll back, and a database session store with the user key in its own column, end_sessions and idle and absolute limits. Ticket 1.3. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Sk72AxLwQnYttF6VFSxQLt
This commit is contained in:
co-authored by
Claude Sonnet 5.5
parent
003ce6e0c2
commit
73505ac717
@@ -0,0 +1,91 @@
|
||||
from datetime import UTC, datetime, timedelta
|
||||
|
||||
import pytest
|
||||
from django.db import transaction
|
||||
|
||||
from portal.accounts import throttle
|
||||
from portal.accounts.models import ThrottleCounter
|
||||
|
||||
T0 = datetime(2026, 1, 1, 12, 0, tzinfo=UTC)
|
||||
EP = "sign-in"
|
||||
|
||||
pytestmark = pytest.mark.django_db(transaction=True)
|
||||
|
||||
|
||||
def fail(n, at=T0, **kw):
|
||||
for _ in range(n):
|
||||
throttle.hit(EP, clock=lambda: at, **kw)
|
||||
|
||||
|
||||
def test_below_limit_passes():
|
||||
fail(4, user="alice")
|
||||
throttle.check(EP, user="alice", clock=lambda: T0)
|
||||
|
||||
|
||||
def test_user_is_throttled_but_others_pass():
|
||||
fail(5, user="Alice", ip="1.1.1.1")
|
||||
with pytest.raises(throttle.Throttled) as exc:
|
||||
throttle.check(EP, user="alice", clock=lambda: T0)
|
||||
assert exc.value.retry_after == 900
|
||||
throttle.check(EP, user="bob", ip="2.2.2.2", clock=lambda: T0)
|
||||
|
||||
|
||||
def test_ip_is_throttled_across_users():
|
||||
fail(5, user="a", ip="1.1.1.1")
|
||||
with pytest.raises(throttle.Throttled):
|
||||
throttle.check(EP, user="other", ip="1.1.1.1", clock=lambda: T0)
|
||||
throttle.check(EP, user="other", ip="3.3.3.3", clock=lambda: T0)
|
||||
|
||||
|
||||
def test_other_endpoint_is_unaffected():
|
||||
fail(5, user="alice")
|
||||
throttle.check("password-change", user="alice", clock=lambda: T0)
|
||||
|
||||
|
||||
def test_pause_elapses():
|
||||
fail(5, user="alice")
|
||||
later = T0 + timedelta(seconds=901)
|
||||
throttle.check(EP, user="alice", clock=lambda: later)
|
||||
# a new failure after the pause starts a fresh count
|
||||
throttle.hit(EP, user="alice", clock=lambda: later)
|
||||
assert ThrottleCounter.objects.get(subject="user:alice").count == 1
|
||||
|
||||
|
||||
def test_stale_window_restarts_count():
|
||||
fail(4, user="alice")
|
||||
later = T0 + timedelta(minutes=16)
|
||||
throttle.hit(EP, user="alice", clock=lambda: later)
|
||||
assert ThrottleCounter.objects.get().count == 1
|
||||
|
||||
|
||||
def test_reset_clears_user_but_never_the_ip():
|
||||
fail(5, user="alice", ip="1.1.1.1")
|
||||
throttle.reset(EP, user="alice")
|
||||
throttle.check(EP, user="alice", clock=lambda: T0)
|
||||
with pytest.raises(throttle.Throttled):
|
||||
throttle.check(EP, ip="1.1.1.1", clock=lambda: T0)
|
||||
|
||||
|
||||
def test_reset_has_no_ip_parameter():
|
||||
with pytest.raises(TypeError):
|
||||
throttle.reset(EP, user="alice", ip="1.1.1.1")
|
||||
|
||||
|
||||
def test_hit_inside_transaction_raises_and_counts_nothing():
|
||||
with pytest.raises(RuntimeError), transaction.atomic():
|
||||
throttle.hit(EP, user="alice", clock=lambda: T0)
|
||||
assert ThrottleCounter.objects.count() == 0
|
||||
|
||||
|
||||
def test_no_subject_raises_for_all_functions():
|
||||
with pytest.raises(ValueError):
|
||||
throttle.check(EP)
|
||||
with pytest.raises(ValueError):
|
||||
throttle.hit(EP)
|
||||
with pytest.raises(ValueError):
|
||||
throttle.reset(EP, user="")
|
||||
|
||||
|
||||
def test_check_never_writes():
|
||||
throttle.check(EP, user="alice", ip="1.1.1.1", clock=lambda: T0)
|
||||
assert ThrottleCounter.objects.count() == 0
|
||||
Reference in New Issue
Block a user