fix: honor X-Forwarded-For for rate limits behind trusted proxies (#221)

Parse client IPs from X-Forwarded-For only when the immediate peer is listed
in TRUSTED_PROXY_IPS, and enable ProxyHeadersMiddleware in that case.

Author: chriscarrollsmith

.env.example

@@ -26,6 +26,9 @@ DB_NAME=
 # Domain for production (used by Caddy for TLS)
 DOMAIN=
 
+# Comma-separated reverse-proxy peer IPs for X-Forwarded-For (e.g. 127.0.0.1,::1)
+# TRUSTED_PROXY_IPS=
+
 # Set to 0 to disable CSRF checks (not recommended in production)
 # CSRF_ENABLED=1
 

main.py

@@ -21,6 +21,7 @@
     require_unauthenticated_client,
 )
 from utils.core.auth import refresh_token_is_persistent, set_auth_cookies
+from utils.core.rate_limit import get_trusted_proxy_hosts
 from utils.core.csrf import (
     CSRF_COOKIE_NAME,
     UNSAFE_HTTP_METHODS,
@@ -63,6 +64,12 @@ async def lifespan(app: FastAPI):
 # Initialize the FastAPI app
 app: FastAPI = FastAPI(lifespan=lifespan)
 
+trusted_proxy_hosts = get_trusted_proxy_hosts()
+if trusted_proxy_hosts:
+    from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
+
+    app.add_middleware(ProxyHeadersMiddleware, trusted_hosts=list(trusted_proxy_hosts))
+
 # Mount static files (e.g., CSS, JS) and initialize Jinja2 templates
 app.mount("/static", StaticFiles(directory="static"), name="static")
 templates = Jinja2Templates(directory="templates")

tests/utils/test_rate_limit.py

@@ -1,9 +1,9 @@
 import importlib
 import time
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
 import utils.core.rate_limit as rate_limit_module
-from utils.core.rate_limit import RateLimitWindow
+from utils.core.rate_limit import RateLimitWindow, get_client_ip
 
 
 # ---------------------------------------------------------------------------
@@ -162,3 +162,53 @@ def test_module_limiters_honor_env_configuration(monkeypatch):
         assert rate_limit_module.forgot_password_email_limiter.window_seconds == 91
 
     importlib.reload(rate_limit_module)
+
+
+# ---------------------------------------------------------------------------
+# Client IP behind trusted proxies
+# ---------------------------------------------------------------------------
+
+
+def _make_request(
+    peer_host: str | None,
+    headers: dict[str, str] | None = None,
+) -> MagicMock:
+    request = MagicMock()
+    if peer_host is None:
+        request.client = None
+    else:
+        request.client = MagicMock(host=peer_host)
+    request.headers = headers or {}
+    return request
+
+
+def test_get_client_ip_ignores_x_forwarded_for_without_trusted_proxy(monkeypatch):
+    monkeypatch.delenv("TRUSTED_PROXY_IPS", raising=False)
+    request = _make_request(
+        "203.0.113.10",
+        headers={"x-forwarded-for": "198.51.100.7"},
+    )
+    assert get_client_ip(request) == "203.0.113.10"
+
+
+def test_get_client_ip_uses_x_forwarded_for_from_trusted_proxy(monkeypatch):
+    monkeypatch.setenv("TRUSTED_PROXY_IPS", "127.0.0.1")
+    request = _make_request(
+        "127.0.0.1",
+        headers={"x-forwarded-for": "198.51.100.7, 127.0.0.1"},
+    )
+    assert get_client_ip(request) == "198.51.100.7"
+
+
+def test_get_client_ip_falls_back_to_peer_when_x_forwarded_for_missing(
+    monkeypatch,
+):
+    monkeypatch.setenv("TRUSTED_PROXY_IPS", "127.0.0.1")
+    request = _make_request("127.0.0.1")
+    assert get_client_ip(request) == "127.0.0.1"
+
+
+def test_get_client_ip_unknown_when_peer_missing(monkeypatch):
+    monkeypatch.delenv("TRUSTED_PROXY_IPS", raising=False)
+    request = _make_request(None)
+    assert get_client_ip(request) == "unknown"

utils/core/rate_limit.py

@@ -2,6 +2,7 @@
 import time
 import threading
 import math
+import ipaddress
 from logging import getLogger
 from typing import Tuple
 
@@ -13,6 +14,58 @@
 load_dotenv()
 
 
+def get_trusted_proxy_hosts() -> tuple[str, ...]:
+    """
+    Return trusted reverse-proxy peer addresses from TRUSTED_PROXY_IPS.
+
+    Comma-separated list, e.g. ``127.0.0.1,::1,172.18.0.2``.
+    """
+    raw = os.environ.get("TRUSTED_PROXY_IPS", "")
+    return tuple(part.strip() for part in raw.split(",") if part.strip())
+
+
+def _peer_host(request: Request) -> str | None:
+    if request.client is None:
+        return None
+    return request.client.host
+
+
+def _parse_forwarded_client_ip(forwarded_for: str) -> str | None:
+    for candidate in forwarded_for.split(","):
+        value = candidate.strip()
+        if not value:
+            continue
+        try:
+            return str(ipaddress.ip_address(value))
+        except ValueError:
+            continue
+    return None
+
+
+def get_client_ip(request: Request) -> str:
+    """
+    Extract the client IP for rate limiting.
+
+    When the immediate peer is listed in TRUSTED_PROXY_IPS, the left-most
+    valid address in X-Forwarded-For is treated as the client. Otherwise only
+    request.client.host is used so clients cannot spoof their IP.
+    """
+    peer = _peer_host(request)
+    if peer is None:
+        return "unknown"
+
+    trusted_hosts = get_trusted_proxy_hosts()
+    if peer not in trusted_hosts:
+        return peer
+
+    forwarded_for = request.headers.get("x-forwarded-for")
+    if not forwarded_for:
+        return peer
+
+    client_ip = _parse_forwarded_client_ip(forwarded_for)
+    return client_ip or peer
+
+
 class RateLimitWindow:
     """
     Thread-safe sliding window rate limiter with bounded in-memory state.
@@ -172,18 +225,6 @@ def clear_all_rate_limiters() -> None:
 # --- Dependency helpers ---
 
 
-def get_client_ip(request: Request) -> str:
-    """
-    Extract client IP from the request.
-
-    Uses request.client.host only. Does NOT trust X-Forwarded-For
-    because this app has no trusted-proxy middleware.
-    """
-    if request.client:
-        return request.client.host
-    return "unknown"
-
-
 def _enforce_rate_limit(limiter: RateLimitWindow, key: str, scope: str) -> int:
     """
     Check the limiter for the given key. If limited, raise RateLimitError.