"""Tests for RateLimitMiddleware honoring proxy headers (X-Forwarded-For).""" import pytest from src.backend.api.middleware.rate_limit import RateLimitMiddleware from src.backend.config import settings class _FakeRequest: """Minimal stand-in for starlette.requests.Request.""" def __init__(self, headers: dict[str, str], host: str | None = "10.0.0.1"): self.headers = headers class _Client: pass if host is None: self.client = None else: self.client = _Client() self.client.host = host @pytest.fixture(autouse=True) def default_trusted_proxy_count(monkeypatch): """Every test in this file assumes the default (1 trusted proxy).""" monkeypatch.setattr(settings, "trusted_proxy_count", 1) def test_client_ip_uses_last_trusted_hop_of_x_forwarded_for(): """With one trusted proxy, the client IP is the LAST element of XFF - the one appended by that proxy - not the first, which the caller supplies and can set to anything. """ req = _FakeRequest( headers={"x-forwarded-for": "203.0.113.5, 10.0.0.2"}, host="10.0.0.1", ) assert RateLimitMiddleware._client_ip(req) == "10.0.0.2" def test_client_ip_ignores_spoofed_first_hop(): """Rotating the client-supplied (leftmost) part of XFF changes nothing: the trusted, rightmost hop stays the same, so the bucket key is stable. """ req1 = _FakeRequest(headers={"x-forwarded-for": "1.1.1.1, 10.0.0.2"}) req2 = _FakeRequest(headers={"x-forwarded-for": "9.9.9.9, 10.0.0.2"}) assert RateLimitMiddleware._client_ip(req1) == RateLimitMiddleware._client_ip(req2) == "10.0.0.2" def test_client_ip_strips_whitespace(): req = _FakeRequest(headers={"x-forwarded-for": " 198.51.100.7 "}) assert RateLimitMiddleware._client_ip(req) == "198.51.100.7" def test_client_ip_falls_back_to_x_real_ip(): req = _FakeRequest(headers={"x-real-ip": "203.0.113.99"}, host="10.0.0.1") assert RateLimitMiddleware._client_ip(req) == "203.0.113.99" def test_client_ip_falls_back_to_request_client_host(): req = _FakeRequest(headers={}, host="172.18.0.5") assert RateLimitMiddleware._client_ip(req) == "172.18.0.5" def test_client_ip_returns_unknown_without_client(): req = _FakeRequest(headers={}, host=None) assert RateLimitMiddleware._client_ip(req) == "unknown" def test_x_forwarded_for_overrides_x_real_ip(): req = _FakeRequest( headers={ "x-forwarded-for": "203.0.113.5", "x-real-ip": "10.0.0.2", }, host="10.0.0.1", ) assert RateLimitMiddleware._client_ip(req) == "203.0.113.5" def test_empty_x_forwarded_for_falls_through(): req = _FakeRequest( headers={"x-forwarded-for": "", "x-real-ip": "203.0.113.42"}, host="10.0.0.1", ) assert RateLimitMiddleware._client_ip(req) == "203.0.113.42" def test_x_forwarded_for_shorter_than_trusted_proxy_count_falls_back(monkeypatch): """Two trusted proxies configured, but XFF has only one hop: falls back to request.client.host rather than trusting an under-populated header.""" monkeypatch.setattr(settings, "trusted_proxy_count", 2) req = _FakeRequest( headers={"x-forwarded-for": "203.0.113.5"}, host="172.18.0.9", ) assert RateLimitMiddleware._client_ip(req) == "172.18.0.9" def test_trusted_proxy_count_two_uses_second_to_last_hop(monkeypatch): monkeypatch.setattr(settings, "trusted_proxy_count", 2) req = _FakeRequest( headers={"x-forwarded-for": "1.1.1.1, 203.0.113.5, 10.0.0.2"}, host="10.0.0.1", ) assert RateLimitMiddleware._client_ip(req) == "203.0.113.5" def test_trusted_proxy_count_zero_ignores_all_proxy_headers(monkeypatch): monkeypatch.setattr(settings, "trusted_proxy_count", 0) req = _FakeRequest( headers={ "x-forwarded-for": "203.0.113.5, 10.0.0.2", "x-real-ip": "203.0.113.99", }, host="172.18.0.9", ) assert RateLimitMiddleware._client_ip(req) == "172.18.0.9" class TestLoginBucketPaths: """_is_login_bucket_request: which paths share the login rate-limit bucket.""" def test_login_path_is_login_bucket(self): assert RateLimitMiddleware._is_login_bucket_request("POST", "/api/auth/login") def test_login_path_get_is_not_login_bucket(self): assert not RateLimitMiddleware._is_login_bucket_request("GET", "/api/auth/login") def test_measurement_authorise_is_login_bucket(self): assert RateLimitMiddleware._is_login_bucket_request( "POST", "/api/measurements/42/authorise" ) def test_production_pause_resume_close_are_login_bucket(self): for action in ("pause", "resume", "close"): assert RateLimitMiddleware._is_login_bucket_request( "POST", f"/api/production-runs/7/{action}" ) def test_production_cycle_is_not_login_bucket(self): """Only the supervisor-gated actions, not every production-runs path.""" assert not RateLimitMiddleware._is_login_bucket_request( "POST", "/api/production-runs/7/cycle" ) def test_setup_paths_are_login_bucket_any_method(self): assert RateLimitMiddleware._is_login_bucket_request("POST", "/api/setup/init-db") assert RateLimitMiddleware._is_login_bucket_request("GET", "/api/setup/status") def test_bare_setup_page_is_not_login_bucket(self): """The HTML setup page itself carries no password.""" assert not RateLimitMiddleware._is_login_bucket_request("GET", "/api/setup") def test_unrelated_path_is_not_login_bucket(self): assert not RateLimitMiddleware._is_login_bucket_request("POST", "/api/recipes") class TestEffectiveLimit: """_effective_limit: dividing a configured limit across worker processes.""" def test_single_worker_keeps_configured_limit(self, monkeypatch): monkeypatch.setattr(settings, "web_concurrency", 1) assert RateLimitMiddleware._effective_limit(5) == 5 def test_divides_and_rounds_up(self, monkeypatch): monkeypatch.setattr(settings, "web_concurrency", 4) assert RateLimitMiddleware._effective_limit(5) == 2 # ceil(5/4) def test_never_below_one(self, monkeypatch): monkeypatch.setattr(settings, "web_concurrency", 100) assert RateLimitMiddleware._effective_limit(5) == 1 def test_non_positive_concurrency_treated_as_one(self, monkeypatch): monkeypatch.setattr(settings, "web_concurrency", 0) assert RateLimitMiddleware._effective_limit(5) == 5