diff --git a/headroom/proxy/forwarded_headers.py b/headroom/proxy/forwarded_headers.py index c41b6ec2f..50ade8589 100644 --- a/headroom/proxy/forwarded_headers.py +++ b/headroom/proxy/forwarded_headers.py @@ -57,6 +57,15 @@ import logging import os from typing import TYPE_CHECKING, Any +from headroom.proxy.forwarded_policy import ( + ForwardedHeaderInputs, + header_first, + normalize_ip, + parse_cidr_list, + peer_is_trusted_gateway, + resolve_forwarded_headers, +) + if TYPE_CHECKING: from fastapi import Request @@ -87,20 +96,7 @@ def _parse_cidr_list( would silently downgrade the proxy from "strict" to "more strict", masking the operator's intent. """ - if not raw or not raw.strip(): - return () - nets: list[ipaddress.IPv4Network | ipaddress.IPv6Network] = [] - for chunk in raw.split(","): - entry = chunk.strip() - if not entry: - # Tolerate `"10.0.0.0/8,"` — trailing comma is benign. - continue - # ``strict=False`` so ``10.0.0.1/8`` is accepted as the network - # ``10.0.0.0/8`` instead of rejecting host bits — operators - # routinely paste a sample IP with a netmask and expect it to - # mean the network. - nets.append(ipaddress.ip_network(entry, strict=False)) - return tuple(nets) + return parse_cidr_list(raw) def load_trusted_gateway_cidrs( @@ -130,43 +126,7 @@ def _normalize_ip( Returns ``None`` on malformed input; callers treat that as "not a trusted gateway". """ - try: - addr = ipaddress.ip_address(host) - except ValueError: - return None - if isinstance(addr, ipaddress.IPv6Address) and addr.ipv4_mapped is not None: - return addr.ipv4_mapped - return addr - - -def peer_is_trusted_gateway( - peer_host: str | None, - cidrs: tuple[ipaddress.IPv4Network | ipaddress.IPv6Network, ...], -) -> bool: - """Return True iff ``peer_host`` is inside any of the allow-list CIDRs. - - Empty allow-list → always False (the strict-secure default). - ``None`` peer (TestClient / UDS) → False; we never trust a peer we - can't even identify. - """ - if not cidrs: - return False - if peer_host is None: - return False - addr = _normalize_ip(peer_host) - if addr is None: - return False - for net in cidrs: - # IP/network family must match; ipaddress raises TypeError if - # we pass an IPv4 address into an IPv6 network membership test - # in some versions, so we family-gate first. - if isinstance(addr, ipaddress.IPv4Address) and isinstance(net, ipaddress.IPv6Network): - continue - if isinstance(addr, ipaddress.IPv6Address) and isinstance(net, ipaddress.IPv4Network): - continue - if addr in net: - return True - return False + return normalize_ip(host) def _peer_host(request: Any) -> str | None: @@ -185,10 +145,7 @@ def _header_first(value: str) -> str: leftmost hop is the only one whose authenticity the immediate gateway can vouch for, and beyond that we have no trust signal. """ - if not value: - return "" - head, _, _ = value.partition(",") - return head.strip() + return header_first(value) def _read_header(request: Any, name: str) -> str: @@ -249,29 +206,22 @@ def _resolve(request: Any) -> tuple[str, dict[str, str]]: if cached_ip is not None and cached_fwd is not None: return cached_ip, cached_fwd - peer_host = _peer_host(request) or "" - fwd_for_raw = _read_header(request, "x-forwarded-for") - fwd_proto_raw = _read_header(request, "x-forwarded-proto") - fwd_host_raw = _read_header(request, "x-forwarded-host") - - cidrs = load_trusted_gateway_cidrs() - trusted = peer_is_trusted_gateway(peer_host or None, cidrs) - - if trusted: - client_ip = _header_first(fwd_for_raw) or peer_host - forwarded = { - "for": _header_first(fwd_for_raw), - "proto": fwd_proto_raw.strip(), - "host": fwd_host_raw.strip(), - } - else: - # Headers may be absent (legitimate direct client). Only emit - # the rejection event if the peer ACTUALLY tried to set one — - # otherwise we'd spam logs for every direct request. - if fwd_for_raw or fwd_proto_raw or fwd_host_raw: - _emit_rejection_event(peer_host or None, fwd_for_raw, fwd_proto_raw, fwd_host_raw) - client_ip = peer_host - forwarded = {"for": "", "proto": "", "host": ""} + inputs = ForwardedHeaderInputs( + peer_host=_peer_host(request) or "", + forwarded_for=_read_header(request, "x-forwarded-for"), + forwarded_proto=_read_header(request, "x-forwarded-proto"), + forwarded_host=_read_header(request, "x-forwarded-host"), + ) + resolution = resolve_forwarded_headers(inputs, load_trusted_gateway_cidrs()) + if resolution.rejected: + _emit_rejection_event( + inputs.peer_host or None, + inputs.forwarded_for, + inputs.forwarded_proto, + inputs.forwarded_host, + ) + client_ip = resolution.client_ip + forwarded = resolution.forwarded if state is not None: try: diff --git a/headroom/proxy/forwarded_policy.py b/headroom/proxy/forwarded_policy.py new file mode 100644 index 000000000..dc14cfb47 --- /dev/null +++ b/headroom/proxy/forwarded_policy.py @@ -0,0 +1,112 @@ +"""Pure policy for trusted forwarded headers.""" + +from __future__ import annotations + +import ipaddress +from dataclasses import dataclass + +Network = ipaddress.IPv4Network | ipaddress.IPv6Network +Address = ipaddress.IPv4Address | ipaddress.IPv6Address + + +@dataclass(frozen=True) +class ForwardedHeaderInputs: + """Raw connection and forwarded-header values before trust evaluation.""" + + peer_host: str + forwarded_for: str = "" + forwarded_proto: str = "" + forwarded_host: str = "" + + @property + def has_forwarded_headers(self) -> bool: + return bool(self.forwarded_for or self.forwarded_proto or self.forwarded_host) + + +@dataclass(frozen=True) +class ForwardedHeaderResolution: + """Deterministic forwarded-header trust decision.""" + + client_ip: str + forwarded: dict[str, str] + trusted: bool + rejected: bool + + +def parse_cidr_list(raw: str) -> tuple[Network, ...]: + """Parse a comma-separated CIDR list. Empty / whitespace returns empty.""" + if not raw or not raw.strip(): + return () + nets: list[Network] = [] + for chunk in raw.split(","): + entry = chunk.strip() + if not entry: + continue + nets.append(ipaddress.ip_network(entry, strict=False)) + return tuple(nets) + + +def normalize_ip(host: str) -> Address | None: + """Parse ``host`` as IP, normalizing IPv4-mapped IPv6 to IPv4.""" + try: + addr = ipaddress.ip_address(host) + except ValueError: + return None + if isinstance(addr, ipaddress.IPv6Address) and addr.ipv4_mapped is not None: + return addr.ipv4_mapped + return addr + + +def peer_is_trusted_gateway( + peer_host: str | None, + cidrs: tuple[Network, ...], +) -> bool: + """Return True iff ``peer_host`` is inside any allow-listed CIDR.""" + if not cidrs or peer_host is None: + return False + addr = normalize_ip(peer_host) + if addr is None: + return False + for net in cidrs: + if isinstance(addr, ipaddress.IPv4Address) and isinstance(net, ipaddress.IPv6Network): + continue + if isinstance(addr, ipaddress.IPv6Address) and isinstance(net, ipaddress.IPv4Network): + continue + if addr in net: + return True + return False + + +def header_first(value: str) -> str: + """Return the leftmost element of a comma-separated header value.""" + if not value: + return "" + head, _, _ = value.partition(",") + return head.strip() + + +def resolve_forwarded_headers( + inputs: ForwardedHeaderInputs, + cidrs: tuple[Network, ...], +) -> ForwardedHeaderResolution: + """Resolve client IP and sanitized forwarded headers from pure inputs.""" + trusted = peer_is_trusted_gateway(inputs.peer_host or None, cidrs) + if trusted: + forwarded_for = header_first(inputs.forwarded_for) + return ForwardedHeaderResolution( + client_ip=forwarded_for or inputs.peer_host, + forwarded={ + "for": forwarded_for, + "proto": inputs.forwarded_proto.strip(), + "host": inputs.forwarded_host.strip(), + }, + trusted=True, + rejected=False, + ) + + return ForwardedHeaderResolution( + client_ip=inputs.peer_host, + forwarded={"for": "", "proto": "", "host": ""}, + trusted=False, + rejected=inputs.has_forwarded_headers, + ) diff --git a/tests/test_forwarded_policy.py b/tests/test_forwarded_policy.py new file mode 100644 index 000000000..c323a1dc9 --- /dev/null +++ b/tests/test_forwarded_policy.py @@ -0,0 +1,70 @@ +"""Tests for pure trusted-forwarded-header policy.""" + +from __future__ import annotations + +from headroom.proxy.forwarded_policy import ( + ForwardedHeaderInputs, + header_first, + parse_cidr_list, + peer_is_trusted_gateway, + resolve_forwarded_headers, +) + + +def test_resolve_trusted_peer_honors_sanitized_forwarded_values() -> None: + cidrs = parse_cidr_list("10.0.0.0/8") + result = resolve_forwarded_headers( + ForwardedHeaderInputs( + peer_host="10.0.0.5", + forwarded_for="203.0.113.7, 10.0.0.99", + forwarded_proto=" https ", + forwarded_host=" api.example.com ", + ), + cidrs, + ) + + assert result.trusted is True + assert result.rejected is False + assert result.client_ip == "203.0.113.7" + assert result.forwarded == { + "for": "203.0.113.7", + "proto": "https", + "host": "api.example.com", + } + + +def test_resolve_untrusted_peer_rejects_forwarded_values() -> None: + cidrs = parse_cidr_list("10.0.0.0/8") + result = resolve_forwarded_headers( + ForwardedHeaderInputs( + peer_host="8.8.8.8", + forwarded_for="203.0.113.7", + forwarded_proto="https", + forwarded_host="api.example.com", + ), + cidrs, + ) + + assert result.trusted is False + assert result.rejected is True + assert result.client_ip == "8.8.8.8" + assert result.forwarded == {"for": "", "proto": "", "host": ""} + + +def test_resolve_direct_untrusted_peer_without_forwarded_values_is_not_rejection() -> None: + result = resolve_forwarded_headers(ForwardedHeaderInputs(peer_host="8.8.8.8"), ()) + + assert result.trusted is False + assert result.rejected is False + assert result.client_ip == "8.8.8.8" + assert result.forwarded == {"for": "", "proto": "", "host": ""} + + +def test_peer_trust_handles_ipv4_mapped_ipv6() -> None: + cidrs = parse_cidr_list("10.0.0.0/8") + + assert peer_is_trusted_gateway("::ffff:10.0.0.1", cidrs) is True + + +def test_header_first_uses_leftmost_forwarded_for_hop() -> None: + assert header_first("203.0.113.7, 10.0.0.99, 10.0.0.5") == "203.0.113.7"