plm-lottery/tests/unit/test_client_ip.py

"""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"