feat(ratelimit): env-tunable per-IP auth limiter budgets
Add RATELIMIT_OTC_REQUEST_MAX / RATELIMIT_VERIFY_MAX / RATELIMIT_CHECK_MAX (default to the existing secure values; non-positive/unparseable → default). Lets a test/PPE stack that drives the auth endpoints repeatedly from one IP lift the budget; production leaves them unset. Mirrors the existing OTC_REQUEST_COOLDOWN_SECONDS knob. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -19,11 +19,27 @@ lockouts). Both layers run together.
|
|||||||
"""
|
"""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import os
|
||||||
import threading
|
import threading
|
||||||
import time
|
import time
|
||||||
from collections import defaultdict, deque
|
from collections import defaultdict, deque
|
||||||
|
|
||||||
|
|
||||||
|
def _max_events(env_name: str, default: int) -> int:
|
||||||
|
"""Per-limiter budget, overridable via env (e.g. a test/PPE stack that
|
||||||
|
drives the auth endpoints repeatedly from one IP). Production leaves these
|
||||||
|
unset and gets the secure defaults below. A non-positive / unparseable
|
||||||
|
value falls back to the default."""
|
||||||
|
raw = os.environ.get(env_name, "").strip()
|
||||||
|
if not raw:
|
||||||
|
return default
|
||||||
|
try:
|
||||||
|
n = int(raw)
|
||||||
|
except ValueError:
|
||||||
|
return default
|
||||||
|
return n if n > 0 else default
|
||||||
|
|
||||||
|
|
||||||
class SlidingWindowLimiter:
|
class SlidingWindowLimiter:
|
||||||
"""Allow at most `max_events` per `window_seconds` per key.
|
"""Allow at most `max_events` per `window_seconds` per key.
|
||||||
|
|
||||||
@@ -66,12 +82,15 @@ class SlidingWindowLimiter:
|
|||||||
# * verify: 10 attempts / 5 min / IP across the auth verify surfaces.
|
# * verify: 10 attempts / 5 min / IP across the auth verify surfaces.
|
||||||
# * otc request: 5 sends / 5 min / IP (Turnstile is the primary gate;
|
# * otc request: 5 sends / 5 min / IP (Turnstile is the primary gate;
|
||||||
# this is defense in depth against a solved-challenge replay loop).
|
# this is defense in depth against a solved-challenge replay loop).
|
||||||
verify_limiter = SlidingWindowLimiter(max_events=10, window_seconds=300)
|
verify_limiter = SlidingWindowLimiter(
|
||||||
otc_request_limiter = SlidingWindowLimiter(max_events=5, window_seconds=300)
|
max_events=_max_events("RATELIMIT_VERIFY_MAX", 10), window_seconds=300)
|
||||||
|
otc_request_limiter = SlidingWindowLimiter(
|
||||||
|
max_events=_max_events("RATELIMIT_OTC_REQUEST_MAX", 5), window_seconds=300)
|
||||||
# /auth/passcode/check is an anonymous has-passcode oracle (audit 0026 L3).
|
# /auth/passcode/check is an anonymous has-passcode oracle (audit 0026 L3).
|
||||||
# It's a legitimate Login-flow affordance, so the budget is generous —
|
# It's a legitimate Login-flow affordance, so the budget is generous —
|
||||||
# enough for a human typing emails, tight enough to stop bulk scraping.
|
# enough for a human typing emails, tight enough to stop bulk scraping.
|
||||||
check_limiter = SlidingWindowLimiter(max_events=30, window_seconds=300)
|
check_limiter = SlidingWindowLimiter(
|
||||||
|
max_events=_max_events("RATELIMIT_CHECK_MAX", 30), window_seconds=300)
|
||||||
|
|
||||||
|
|
||||||
def _reset_all_for_tests() -> None:
|
def _reset_all_for_tests() -> None:
|
||||||
|
|||||||
@@ -0,0 +1,44 @@
|
|||||||
|
"""The per-IP limiter budgets are env-overridable (test/PPE stacks drive the
|
||||||
|
auth endpoints repeatedly from one IP); production leaves them unset and keeps
|
||||||
|
the secure defaults. A non-positive / unparseable value falls back."""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import importlib
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
|
||||||
|
def _reload_with(monkeypatch, **env):
|
||||||
|
for k, v in env.items():
|
||||||
|
if v is None:
|
||||||
|
monkeypatch.delenv(k, raising=False)
|
||||||
|
else:
|
||||||
|
monkeypatch.setenv(k, v)
|
||||||
|
import app.ratelimit as ratelimit
|
||||||
|
return importlib.reload(ratelimit)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(autouse=True)
|
||||||
|
def _restore():
|
||||||
|
yield
|
||||||
|
# Leave the module in its default state for other tests.
|
||||||
|
import app.ratelimit as ratelimit
|
||||||
|
importlib.reload(ratelimit)
|
||||||
|
|
||||||
|
|
||||||
|
def test_defaults_when_unset(monkeypatch):
|
||||||
|
rl = _reload_with(monkeypatch, RATELIMIT_OTC_REQUEST_MAX=None, RATELIMIT_VERIFY_MAX=None)
|
||||||
|
assert rl.otc_request_limiter.max_events == 5
|
||||||
|
assert rl.verify_limiter.max_events == 10
|
||||||
|
|
||||||
|
|
||||||
|
def test_env_override(monkeypatch):
|
||||||
|
rl = _reload_with(monkeypatch, RATELIMIT_OTC_REQUEST_MAX="1000", RATELIMIT_VERIFY_MAX="250")
|
||||||
|
assert rl.otc_request_limiter.max_events == 1000
|
||||||
|
assert rl.verify_limiter.max_events == 250
|
||||||
|
|
||||||
|
|
||||||
|
def test_bad_value_falls_back_to_default(monkeypatch):
|
||||||
|
rl = _reload_with(monkeypatch, RATELIMIT_OTC_REQUEST_MAX="nope", RATELIMIT_VERIFY_MAX="0")
|
||||||
|
assert rl.otc_request_limiter.max_events == 5 # unparseable → default
|
||||||
|
assert rl.verify_limiter.max_events == 10 # non-positive → default
|
||||||
Reference in New Issue
Block a user