From e47a68e0213b8b0ca46430ee6c5df5445f3a6f85 Mon Sep 17 00:00:00 2001 From: yellowcooln <12516003+yellowcooln@users.noreply.github.com> Date: Tue, 8 Sep 2026 09:29:31 -0400 Subject: [PATCH] fix(auth): bound and expire login throttle state --- repeater/web/auth_endpoints.py | 61 +++++++++++++---- tests/test_login_throttle_storage.py | 97 ++++++++++++++++++++++++++++ 2 files changed, 146 insertions(+), 12 deletions(-) create mode 100644 tests/test_login_throttle_storage.py diff --git a/repeater/web/auth_endpoints.py b/repeater/web/auth_endpoints.py index 06e60d9..ffa99a2 100644 --- a/repeater/web/auth_endpoints.py +++ b/repeater/web/auth_endpoints.py @@ -2,6 +2,7 @@ Authentication endpoints for login and token management """ +import hashlib import logging import math import threading @@ -35,20 +36,30 @@ class _LoginThrottle: self.base_backoff_sec = base_backoff_sec self.max_backoff_sec = max_backoff_sec self.window_sec = window_sec + self.max_states = 4096 self._time_fn = time_fn or time.monotonic self._lock = threading.Lock() self._ip_states = {} self._user_states = {} self._global_state = {"failures": 0, "last_failure": 0.0, "blocked_until": 0.0} - def _state(self, bucket: dict, key: str): + @staticmethod + def _keys(client_ip: str, username: str) -> tuple[bytes, bytes]: + # Fixed-size digests also bound storage for oversized attacker input. + user_key = hashlib.sha256( + ((username or "").strip().lower() or "").encode("utf-8") + ).digest() + ip_key = hashlib.sha256((client_ip or "").encode("utf-8")).digest() + return ip_key, user_key + + def _state(self, bucket: dict, key: bytes): if key not in bucket: bucket[key] = {"failures": 0, "last_failure": 0.0, "blocked_until": 0.0} return bucket[key] def _maybe_decay(self, state: dict, now: float) -> None: last = state.get("last_failure", 0.0) - if last and (now - last) > self.window_sec: + if (now - last) > self.window_sec and state.get("blocked_until", 0.0) <= now: state["failures"] = 0 state["blocked_until"] = 0.0 @@ -68,21 +79,48 @@ class _LoginThrottle: return 0 return max(1, math.ceil(blocked_until - now)) + def _get_retry_after(self, ip_key: bytes, user_key: bytes, now: float) -> int: + # Rejected requests must not allocate attacker-controlled keys. + global_retry = self._retry_after(self._global_state, now) + if global_retry: + return global_retry + for bucket in (self._ip_states, self._user_states): + expired = [ + key + for key, state in bucket.items() + if now - state["last_failure"] > self.window_sec and state["blocked_until"] <= now + ] + for key in expired: + del bucket[key] + retry = 0 + for bucket, key in ((self._ip_states, ip_key), (self._user_states, user_key)): + if key in bucket: + retry = max(retry, self._retry_after(bucket[key], now)) + elif len(bucket) >= self.max_states: + # Fail closed until space expires; evicting active failures would + # let a rotating-key attacker reset another client's backoff. + expires = min( + max(state["last_failure"] + self.window_sec, state["blocked_until"]) + for state in bucket.values() + ) + retry = max(retry, max(1, math.ceil(expires - now))) + return retry + def get_retry_after(self, client_ip: str, username: str) -> int: now = self._time_fn() - user_key = (username or "").strip().lower() or "" - ip_key = client_ip or "" + ip_key, user_key = self._keys(client_ip, username) with self._lock: - ip_retry = self._retry_after(self._state(self._ip_states, ip_key), now) - user_retry = self._retry_after(self._state(self._user_states, user_key), now) - global_retry = self._retry_after(self._global_state, now) - return max(ip_retry, user_retry, global_retry) + return self._get_retry_after(ip_key, user_key, now) def register_failure(self, client_ip: str, username: str) -> int: now = self._time_fn() - user_key = (username or "").strip().lower() or "" - ip_key = client_ip or "" + ip_key, user_key = self._keys(client_ip, username) with self._lock: + # Recheck under the lock: other requests may have filled or blocked + # the store while this request was validating credentials. + retry = self._get_retry_after(ip_key, user_key, now) + if retry: + return retry self._record_failure(self._state(self._ip_states, ip_key), self.per_ip_threshold, now) self._record_failure( self._state(self._user_states, user_key), self.per_user_threshold, now @@ -94,8 +132,7 @@ class _LoginThrottle: return max(ip_retry, user_retry, global_retry) def register_success(self, client_ip: str, username: str) -> None: - user_key = (username or "").strip().lower() or "" - ip_key = client_ip or "" + ip_key, user_key = self._keys(client_ip, username) with self._lock: self._ip_states.pop(ip_key, None) self._user_states.pop(user_key, None) diff --git a/tests/test_login_throttle_storage.py b/tests/test_login_throttle_storage.py new file mode 100644 index 0000000..3fe9a81 --- /dev/null +++ b/tests/test_login_throttle_storage.py @@ -0,0 +1,97 @@ +import pytest + +from repeater.web.auth_endpoints import _LoginThrottle + + +@pytest.mark.parametrize("blocked", ["ip", "user", "global"]) +def test_blocked_requests_do_not_allocate_states(blocked): + throttle = _LoginThrottle( + per_ip_threshold=1 if blocked == "ip" else 99, + per_user_threshold=1 if blocked == "user" else 99, + global_threshold=1 if blocked == "global" else 99, + time_fn=lambda: 1000, + ) + assert throttle.register_failure("ip", "user") > 0 + for i in range(100): + ip = "ip" if blocked == "ip" else f"ip-{i}" + user = "user" if blocked == "user" else f"user-{i}" + assert throttle.get_retry_after(ip, user) > 0 + assert len(throttle._ip_states) == 1 + assert len(throttle._user_states) == 1 + + +def test_retained_keys_have_bounded_size_and_success_clears_them(): + throttle = _LoginThrottle() + ip, user = "i" * 100000, "u" * 100000 + assert throttle.register_failure(ip, user) == 0 + assert max(map(len, throttle._ip_states)) <= 64 + assert max(map(len, throttle._user_states)) <= 64 + throttle.register_success(ip, f" {user.upper()} ") + assert not throttle._ip_states + assert not throttle._user_states + + +def test_unfailed_lookups_do_not_allocate_states(): + throttle = _LoginThrottle() + for i in range(100): + assert throttle.get_retry_after(f"ip-{i}", f"user-{i}") == 0 + assert not throttle._ip_states + assert not throttle._user_states + + +@pytest.mark.parametrize("start", [0, 1000]) +def test_expired_states_are_removed_without_revisiting_keys(start): + now = [start] + throttle = _LoginThrottle(time_fn=lambda: now[0]) + throttle.register_failure("old-ip", "old-user") + now[0] += throttle.window_sec + 1 + assert throttle.get_retry_after("new-ip", "new-user") == 0 + assert not throttle._ip_states + assert not throttle._user_states + assert throttle._global_state["failures"] == 0 + + +def test_capacity_fails_closed_without_evicting_active_failures(): + now = [1000] + throttle = _LoginThrottle( + per_ip_threshold=2, + per_user_threshold=2, + global_threshold=10000, + time_fn=lambda: now[0], + ) + # A small cap exercises the same bounded-store path as the default limit. + throttle.max_states = 3 + for i in range(3): + assert throttle.register_failure(f"ip-{i}", f"user-{i}") == 0 + for i in range(3, 100): + assert throttle.get_retry_after(f"ip-{i}", f"user-{i}") > 0 + # A failure already in flight must not bypass the capacity bound either. + assert throttle.register_failure(f"ip-{i}", f"user-{i}") > 0 + assert len(throttle._ip_states) == 3 + assert len(throttle._user_states) == 3 + assert throttle.register_failure("ip-0", " USER-0 ") > 0 + now[0] += throttle.window_sec + 1 + assert throttle.get_retry_after("new-ip", "new-user") == 0 + assert throttle.register_failure("new-ip", "new-user") == 0 + assert len(throttle._ip_states) == 1 + assert len(throttle._user_states) == 1 + + +def test_expiry_does_not_shorten_active_backoff(): + now = [1000] + throttle = _LoginThrottle( + per_ip_threshold=1, + per_user_threshold=1, + global_threshold=1, + window_sec=5, + base_backoff_sec=20, + max_backoff_sec=20, + time_fn=lambda: now[0], + ) + assert throttle.register_failure("ip", "user") == 20 + now[0] += 6 + assert throttle.get_retry_after("ip", "user") == 14 + now[0] += 15 + assert throttle.get_retry_after("ip", "user") == 0 + assert not throttle._ip_states + assert not throttle._user_states