6399039589
websocket.client.host is always the Cloudflare Tunnel machine's LAN IP for every internet-facing connection (the tunnel runs on a separate machine and terminates TLS there), which collapsed per-IP rate limiting into a single shared bucket for all remote visitors — the exact gap flagged in review. Cloudflare's edge sets CF-Connecting-IP itself, overwriting any client-supplied value, so it's safe to trust when present. Falls back to the raw socket peer for direct LAN/local access.
49 lines
1.8 KiB
Python
49 lines
1.8 KiB
Python
from unittest.mock import patch
|
|
|
|
from app.rate_limit import RateLimiter, resolve_client_ip
|
|
|
|
|
|
def test_allows_up_to_limit_then_blocks():
|
|
limiter = RateLimiter(max_requests=3, window_seconds=60)
|
|
assert limiter.allow("user-1") is True
|
|
assert limiter.allow("user-1") is True
|
|
assert limiter.allow("user-1") is True
|
|
assert limiter.allow("user-1") is False
|
|
|
|
|
|
def test_different_keys_tracked_independently():
|
|
limiter = RateLimiter(max_requests=1, window_seconds=60)
|
|
assert limiter.allow("user-1") is True
|
|
assert limiter.allow("user-2") is True
|
|
assert limiter.allow("user-1") is False
|
|
|
|
|
|
def test_hits_expire_after_window_elapses():
|
|
limiter = RateLimiter(max_requests=2, window_seconds=10)
|
|
|
|
with patch("app.rate_limit.time.monotonic", return_value=100.0):
|
|
assert limiter.allow("user-1") is True
|
|
assert limiter.allow("user-1") is True
|
|
assert limiter.allow("user-1") is False # at limit
|
|
|
|
with patch("app.rate_limit.time.monotonic", return_value=111.0):
|
|
# 11 seconds later, both prior hits (at t=100) are older than window_start (111 - 10 = 101)
|
|
assert limiter.allow("user-1") is True
|
|
|
|
|
|
def test_resolve_client_ip_prefers_cf_connecting_ip_over_socket_peer():
|
|
# The App CT sits behind a Cloudflare Tunnel on a separate machine — the
|
|
# raw socket peer is always the tunnel, never the visitor, for every
|
|
# internet-facing request.
|
|
headers = {"cf-connecting-ip": "203.0.113.7"}
|
|
assert resolve_client_ip(headers, "10.30.20.1") == "203.0.113.7"
|
|
|
|
|
|
def test_resolve_client_ip_falls_back_to_socket_peer_without_header():
|
|
# Direct LAN/local access (no Cloudflare in front) has no such header.
|
|
assert resolve_client_ip({}, "10.30.20.1") == "10.30.20.1"
|
|
|
|
|
|
def test_resolve_client_ip_falls_back_to_unknown_with_no_peer_or_header():
|
|
assert resolve_client_ip({}, None) == "unknown"
|