Cap dashboard password auth rate buckets

hermes_cli/dashboard_auth/routes.py

@@ -20,7 +20,7 @@ from __future__ import annotations
 import logging
 import threading
 import time
-from collections import defaultdict, deque
+from collections import OrderedDict, deque
 from typing import Any, Deque, Dict
 from urllib.parse import quote, unquote, urlencode, urlparse, urlunparse
 
@@ -364,17 +364,43 @@ async def auth_callback(
 # Uses the ASGI peer; trusted proxy normalization must happen upstream.
 _PW_RATE_MAX_ATTEMPTS = 10
 _PW_RATE_WINDOW_SEC = 60.0
-_pw_attempts: Dict[str, Deque[float]] = defaultdict(deque)
+_PW_RATE_MAX_BUCKETS = 4096
+_pw_attempts: "OrderedDict[str, Deque[float]]" = OrderedDict()
 _pw_attempts_lock = threading.Lock()
 
 
+def _prune_password_rate_buckets_locked(cutoff: float) -> None:
+    """Drop expired/empty password-login limiter buckets.
+
+    Must be called with ``_pw_attempts_lock`` held.
+    """
+    for bucket_key in list(_pw_attempts.keys()):
+        bucket = _pw_attempts[bucket_key]
+        while bucket and bucket[0] < cutoff:
+            bucket.popleft()
+        if not bucket:
+            _pw_attempts.pop(bucket_key, None)
+
+
 def _password_rate_limited(ip: str) -> bool:
     """True if ``ip`` exceeded the budget; records the attempt when allowed. An empty IP shares
     one bucket — fail-safe toward throttling."""
     now = time.monotonic()
     cutoff = now - _PW_RATE_WINDOW_SEC
     with _pw_attempts_lock:
-        bucket = _pw_attempts[ip or "_unknown_"]
+        key = ip or "_unknown_"
+        max_buckets = max(1, int(_PW_RATE_MAX_BUCKETS))
+        if key not in _pw_attempts and len(_pw_attempts) >= max_buckets:
+            _prune_password_rate_buckets_locked(cutoff)
+        while key not in _pw_attempts and len(_pw_attempts) >= max_buckets:
+            _pw_attempts.popitem(last=False)
+
+        bucket = _pw_attempts.get(key)
+        if bucket is None:
+            bucket = deque()
+            _pw_attempts[key] = bucket
+        else:
+            _pw_attempts.move_to_end(key)
         while bucket and bucket[0] < cutoff:
             bucket.popleft()
         if len(bucket) >= _PW_RATE_MAX_ATTEMPTS:

tests/hermes_cli/test_dashboard_auth_password_login.py

@@ -30,9 +30,14 @@ from hermes_cli.dashboard_auth import (
     clear_providers,
     register_provider,
 )
+from hermes_cli.dashboard_auth import routes as auth_routes
 from hermes_cli.dashboard_auth.cookies import SESSION_AT_COOKIE, SESSION_RT_COOKIE
 from hermes_cli.dashboard_auth.login_page import render_login_html
-from hermes_cli.dashboard_auth.routes import _PW_RATE_MAX_ATTEMPTS, _reset_password_rate_limit
+from hermes_cli.dashboard_auth.routes import (
+    _PW_RATE_MAX_ATTEMPTS,
+    _PW_RATE_MAX_BUCKETS,
+    _reset_password_rate_limit,
+)
 from hermes_cli.web_server_lifecycle import _dashboard_forwarded_allow_ips
 from tests.hermes_cli.conftest_dashboard_auth import StubAuthProvider
 
@@ -403,6 +408,34 @@ class TestRateLimit:
         )
         assert allowed.status_code == 200
 
+    def test_distinct_ip_buckets_are_capped(self, monkeypatch):
+        _reset_password_rate_limit()
+        monkeypatch.setattr(auth_routes, "_PW_RATE_MAX_BUCKETS", 3)
+
+        for i in range(5):
+            assert auth_routes._password_rate_limited(f"192.0.2.{i}") is False
+
+        assert list(auth_routes._pw_attempts.keys()) == [
+            "192.0.2.2",
+            "192.0.2.3",
+            "192.0.2.4",
+        ]
+
+    def test_expired_buckets_pruned_before_evicting_live_bucket(self, monkeypatch):
+        _reset_password_rate_limit()
+        monkeypatch.setattr(auth_routes, "_PW_RATE_MAX_BUCKETS", 2)
+        now = time.monotonic()
+        auth_routes._pw_attempts["expired"] = auth_routes.deque([
+            now - auth_routes._PW_RATE_WINDOW_SEC - 1,
+        ])
+        auth_routes._pw_attempts["live"] = auth_routes.deque([now])
+
+        assert auth_routes._password_rate_limited("192.0.2.99") is False
+
+        assert "expired" not in auth_routes._pw_attempts
+        assert "live" in auth_routes._pw_attempts
+        assert "192.0.2.99" in auth_routes._pw_attempts
+
 
 @pytest.mark.parametrize("peer", [("203.0.113.7", 12345), None])
 def test_client_ip_uses_asgi_peer_not_forwarded_header(peer: tuple[str, int] | None) -> None:
@@ -418,7 +451,6 @@ def test_client_ip_uses_asgi_peer_not_forwarded_header(peer: tuple[str, int] | N
     })
     assert client_ip(request) == (peer[0] if peer else "")
 
-
 # ---------------------------------------------------------------------------
 # Login page rendering
 # ---------------------------------------------------------------------------