"""app.api.client_ip is shared by the login/registration throttles (B-33) and the SSE per-IP subscriber cap (B-38) — both depend on it correctly preferring X-Forwarded-For (Caddy reverse-proxies every request, see Caddyfile) over request.client.host, which would otherwise be the proxy's own address. Which *element* of that header it reads is a security property, not a detail: B-54 below is the whole reason all three controls hold at all.""" from starlette.requests import Request from app.api.client_ip import client_ip def _request(*, forwarded: str | None = None, client_host: str | None = "127.0.0.1") -> Request: headers = [(b"x-forwarded-for", forwarded.encode())] if forwarded else [] scope = { "type": "http", "headers": headers, "client": (client_host, 12345) if client_host else None, } return Request(scope) def test_client_ip_prefers_x_forwarded_for(): request = _request(forwarded="5.6.7.8", client_host="10.0.0.1") assert client_ip(request) == "5.6.7.8" def test_client_ip_takes_the_last_hop_of_a_forwarded_chain(): # B-54 """The hop closest to us — the one our own proxy appended. Exactly one trusted proxy sits in front of the app, so this is the real peer.""" request = _request(forwarded="5.6.7.8, 10.0.0.1, 172.17.0.1") assert client_ip(request) == "172.17.0.1" def test_client_ip_ignores_a_client_supplied_prefix(): # B-54 """Caddy *appends* to whatever the client sent, so the front of the header is attacker-controlled. Reading it from the front let anyone mint a fresh identity per request and walk straight through the login/registration throttles (B-33) and the SSE per-IP subscriber cap (B-38). Two requests spoofing different values must still key to the same real IP.""" first = _request(forwarded="1.1.1.1, 203.0.113.9") second = _request(forwarded="2.2.2.2, 203.0.113.9") assert client_ip(first) == client_ip(second) == "203.0.113.9" def test_client_ip_strips_whitespace(): request = _request(forwarded=" 5.6.7.8 , 10.0.0.1 ") assert client_ip(request) == "10.0.0.1" def test_client_ip_falls_back_when_the_header_is_empty(): """An empty or comma-only header used to yield "" — a single shared bucket every caller lands in, which is its own throttle-evasion trick.""" assert client_ip(_request(forwarded=" , ", client_host="10.0.0.1")) == "10.0.0.1" def test_client_ip_falls_back_to_request_client_without_the_header(): request = _request(forwarded=None, client_host="10.0.0.1") assert client_ip(request) == "10.0.0.1" def test_client_ip_falls_back_to_unknown_with_neither(): request = _request(forwarded=None, client_host=None) assert client_ip(request) == "unknown"