The per-IP rate limits could be bypassed entirely: uvicorn ran with --forwarded-allow-ips "*", which trusts the leftmost X-Forwarded-For value (client-controlled), and Render forwards the inbound header rather than stripping it. Rotating the header handed out a fresh rate-limit bucket per request, so the login/register limit (10/5min) and guest-minting limit (30/5min) were no throttle at all — unbounded password guessing and guest-row creation. Confirmed live: fixed IP -> 429 after 10; rotating spoofed header -> no 429 across 14 attempts. Two-layer fix: - limits._client_ip now derives the client IP from the hop the trusted edge appends (rightmost of X-Forwarded-For), which a client can't spoof past; tunable via AIDND_TRUSTED_PROXY_HOPS. Dropped --forwarded-allow-ips "*". - New per-account login throttle (email-keyed, 8 fails / 15 min, cleared on success): stops distributed guessing against one account that a per-IP limit can't, since it can't be diluted across many source addresses. Also close an SSRF on the BYOK endpoint_url (hosted mode only): the connection test and turn/chat streams now refuse a URL that resolves to a non-public address (private/loopback/link-local metadata/reserved), checked at request time so it resists a DNS record flipping to a private IP. No-op locally, where reaching localhost Ollama is intended. Tests: test_ratelimit_hardening.py (8), test_netguard.py (13). 172 pass. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015CYEJKobJ2Re4Dv7qUoSA7
109 lines
4.0 KiB
Python
109 lines
4.0 KiB
Python
"""Regression tests for the X-Forwarded-For rate-limit bypass and the
|
|
per-account login throttle added to close it.
|
|
|
|
Background: uvicorn's --forwarded-allow-ips "*" trusted the LEFTMOST
|
|
X-Forwarded-For entry, which the client controls, so rotating the header
|
|
handed out a fresh rate-limit bucket per request. _client_ip now reads the
|
|
hop the trusted edge appends (rightmost), and login has an email-keyed throttle
|
|
that no IP trick can dilute.
|
|
|
|
python -m pytest tests/test_ratelimit_hardening.py -v
|
|
"""
|
|
import os
|
|
import tempfile
|
|
|
|
_tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
|
|
_tmp.close()
|
|
os.environ["AIDND_DB_PATH"] = _tmp.name
|
|
os.environ.pop("AIDND_DATABASE_URL", None)
|
|
os.environ.pop("DATABASE_URL", None)
|
|
|
|
import pytest
|
|
|
|
from app import auth, limits
|
|
|
|
|
|
class _Req:
|
|
"""Minimal stand-in for starlette's Request: a header lookup and a peer."""
|
|
|
|
def __init__(self, xff: str | None, peer: str | None = "10.0.0.1"):
|
|
self.headers = {} if xff is None else {"x-forwarded-for": xff}
|
|
self.client = None if peer is None else type("C", (), {"host": peer})()
|
|
|
|
|
|
# ---------- _client_ip: the spoof-resistant hop ----------
|
|
|
|
def test_client_ip_takes_appended_rightmost_hop(monkeypatch):
|
|
monkeypatch.setattr(limits, "TRUSTED_PROXY_HOPS", 1)
|
|
# Attacker prepends a fake IP; the edge appends the real one on the right.
|
|
req = _Req("203.0.113.9, 198.51.100.77")
|
|
assert limits._client_ip(req) == "198.51.100.77"
|
|
|
|
|
|
def test_client_ip_ignores_spoofed_leftmost(monkeypatch):
|
|
monkeypatch.setattr(limits, "TRUSTED_PROXY_HOPS", 1)
|
|
# Whatever the client stuffs to the left, the keyed IP stays the real hop —
|
|
# so rotating it no longer mints a new bucket.
|
|
a = limits._client_ip(_Req("1.1.1.1, 198.51.100.77"))
|
|
b = limits._client_ip(_Req("2.2.2.2, 198.51.100.77"))
|
|
c = limits._client_ip(_Req("evil, junk, 198.51.100.77"))
|
|
assert a == b == c == "198.51.100.77"
|
|
|
|
|
|
def test_client_ip_honours_extra_trusted_hops(monkeypatch):
|
|
monkeypatch.setattr(limits, "TRUSTED_PROXY_HOPS", 2)
|
|
# Two trusted hops: real client is second from the right.
|
|
req = _Req("9.9.9.9, 203.0.113.5, 198.51.100.77")
|
|
assert limits._client_ip(req) == "203.0.113.5"
|
|
|
|
|
|
def test_client_ip_falls_back_to_socket_peer():
|
|
assert limits._client_ip(_Req(None, peer="172.16.0.4")) == "172.16.0.4"
|
|
assert limits._client_ip(_Req(None, peer=None)) == "unknown"
|
|
|
|
|
|
# ---------- per-account login throttle ----------
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _multi_user(monkeypatch):
|
|
monkeypatch.setattr(auth, "MULTI_USER", True)
|
|
# Isolate the module-level failure map for each test.
|
|
from collections import defaultdict, deque
|
|
monkeypatch.setattr(limits, "_login_fails", defaultdict(deque))
|
|
|
|
|
|
def test_login_throttle_blocks_after_limit():
|
|
email = "victim@example.com"
|
|
# Up to the limit: allowed, each a recorded failure.
|
|
for _ in range(limits.LOGIN_FAIL_LIMIT):
|
|
limits.check_login_allowed(email) # does not raise
|
|
limits.note_login_failure(email)
|
|
# One more crosses the line.
|
|
with pytest.raises(limits.HTTPException) as exc:
|
|
limits.check_login_allowed(email)
|
|
assert exc.value.status_code == 429
|
|
|
|
|
|
def test_login_throttle_is_per_account():
|
|
for _ in range(limits.LOGIN_FAIL_LIMIT):
|
|
limits.note_login_failure("a@example.com")
|
|
with pytest.raises(limits.HTTPException):
|
|
limits.check_login_allowed("a@example.com")
|
|
# A different account is unaffected — this is not an IP bucket.
|
|
limits.check_login_allowed("b@example.com") # must not raise
|
|
|
|
|
|
def test_successful_login_clears_the_streak():
|
|
email = "typo@example.com"
|
|
for _ in range(limits.LOGIN_FAIL_LIMIT):
|
|
limits.note_login_failure(email)
|
|
limits.note_login_success(email)
|
|
limits.check_login_allowed(email) # streak wiped — must not raise
|
|
|
|
|
|
def test_throttle_is_noop_in_local_mode(monkeypatch):
|
|
monkeypatch.setattr(auth, "MULTI_USER", False)
|
|
for _ in range(limits.LOGIN_FAIL_LIMIT * 3):
|
|
limits.note_login_failure("solo@example.com")
|
|
limits.check_login_allowed("solo@example.com") # never throttled locally
|