chopratejas--headroom/headroom/proxy/forwarded_headers.py

"""Trusted-gateway gate for ``X-Forwarded-*`` headers — Phase F PR-F4.

The proxy must not blindly trust ``X-Forwarded-For``,
``X-Forwarded-Proto``, or ``X-Forwarded-Host`` from arbitrary clients —
a malicious upstream client can forge any of those values and spoof
their origin IP, scheme, or host. We trust them ONLY when the
connecting peer's IP is in a configured CIDR allow-list (i.e. behind
a known reverse proxy / API gateway).

Configuration
-------------

Single env var, comma-separated CIDR blocks::

    HEADROOM_PROXY_TRUSTED_GATEWAY_CIDRS=10.0.0.0/8,172.16.0.0/12,fd00::/8

Whitespace around the commas is tolerated. Empty / unset is the
**default** and the **most secure** setting — it means *no gateway is
trusted*, so every ``X-Forwarded-*`` header is ignored regardless of
peer.

Behaviour matrix
----------------

============================  =====================  ========
Allow-list state              Peer in list?          Result
============================  =====================  ========
unset / empty (default)       n/a                    headers IGNORED
configured                    yes                    headers HONORED
configured                    no                     headers IGNORED + ``forwarded_headers_rejected`` event
============================  =====================  ========

Public API
----------

* :func:`resolve_client_ip` — the IP to log / rate-limit / authorize on.
* :func:`trusted_forwarded_headers` — sanitized ``{proto, host, for}``
  dict; values are empty strings when the gate fails.

Both helpers cache their result on ``request.state`` so they run at
most once per request.

Constraints (per project memory)
--------------------------------

* configurable: env var only, no other config surface.
* no hardcodes: every CIDR comes from the env var.
* no regexes: parsing uses :mod:`ipaddress` from the stdlib.
* no silent fallbacks: a malformed CIDR raises ``ValueError`` at
  startup; every spoof rejection emits a structured log event.
"""

from __future__ import annotations

import ipaddress
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

logger = logging.getLogger(__name__)

__all__ = [
    "TRUSTED_GATEWAY_CIDRS_ENV",
    "load_trusted_gateway_cidrs",
    "peer_is_trusted_gateway",
    "resolve_client_ip",
    "trusted_forwarded_headers",
]


#: Environment variable that holds the comma-separated CIDR allow-list.
TRUSTED_GATEWAY_CIDRS_ENV = "HEADROOM_PROXY_TRUSTED_GATEWAY_CIDRS"


def _parse_cidr_list(
    raw: str,
) -> tuple[ipaddress.IPv4Network | ipaddress.IPv6Network, ...]:
    """Parse a comma-separated CIDR list. Empty / whitespace → empty tuple.

    Whitespace around commas is tolerated. Empty individual entries
    (e.g. trailing comma) are skipped. Malformed entries raise
    :class:`ValueError` — we *deliberately* do not silently skip bad
    CIDRs, because a config typo that quietly empties the allow-list
    would silently downgrade the proxy from "strict" to "more strict",
    masking the operator's intent.
    """
    return parse_cidr_list(raw)


def load_trusted_gateway_cidrs(
    raw: str | None = None,
) -> tuple[ipaddress.IPv4Network | ipaddress.IPv6Network, ...]:
    """Load and parse the trusted-gateway CIDR allow-list.

    ``raw`` is exposed for tests and direct callers; production code
    passes nothing and we read :data:`TRUSTED_GATEWAY_CIDRS_ENV` from
    the process environment. A malformed entry raises
    :class:`ValueError` — let it propagate so the failure is loud at
    startup instead of silently disabling the gate.
    """
    if raw is None:
        raw = os.environ.get(TRUSTED_GATEWAY_CIDRS_ENV, "")
    return _parse_cidr_list(raw)


def _normalize_ip(
    host: str,
) -> ipaddress.IPv4Address | ipaddress.IPv6Address | None:
    """Parse ``host`` into an IPv4/IPv6 address, unmapping ``::ffff:*``.

    IPv4-mapped IPv6 addresses (``::ffff:10.0.0.1``) — emitted by Linux
    dual-stack sockets — are normalized to their underlying IPv4 form
    so a CIDR allow-list of ``10.0.0.0/8`` matches them naturally.
    Returns ``None`` on malformed input; callers treat that as "not a
    trusted gateway".
    """
    return normalize_ip(host)


def _peer_host(request: Any) -> str | None:
    """Pull ``request.client.host`` defensively (TestClient may omit)."""
    client = getattr(request, "client", None)
    if client is None:
        return None
    return getattr(client, "host", None)


def _header_first(value: str) -> str:
    """Return the leftmost element of a comma-separated header value.

    ``X-Forwarded-For: client, proxy1, proxy2`` → ``"client"``. Empty
    input returns ``""``. We intentionally do NOT walk the chain — the
    leftmost hop is the only one whose authenticity the immediate
    gateway can vouch for, and beyond that we have no trust signal.
    """
    return header_first(value)


def _read_header(request: Any, name: str) -> str:
    """Read a header case-insensitively, ``""`` on miss."""
    headers = getattr(request, "headers", None)
    if headers is None:
        return ""
    try:
        value = headers.get(name)
    except AttributeError:
        return ""
    if value is None:
        return ""
    if isinstance(value, bytes):
        try:
            return value.decode("latin-1")
        except UnicodeDecodeError:  # pragma: no cover - defensive
            return ""
    return str(value)


def _emit_rejection_event(
    peer_host: str | None,
    fwd_for: str,
    fwd_proto: str,
    fwd_host: str,
) -> None:
    """One-line structured log for every spoof-rejection.

    Loud-by-design: an operator running a misconfigured network MUST
    see this so they can either widen their CIDR allow-list or fix the
    upstream proxy. The event name is stable for grep / Prometheus
    log-based alerts.
    """
    logger.warning(
        "forwarded_headers_rejected",
        extra={
            "event": "forwarded_headers_rejected",
            "peer_ip": peer_host or "",
            "forwarded_for": fwd_for,
            "forwarded_proto": fwd_proto,
            "forwarded_host": fwd_host,
        },
    )


def _resolve(request: Any) -> tuple[str, dict[str, str]]:
    """Compute (client_ip, sanitized_forwarded_dict) once.

    Cached on ``request.state.client_ip`` /
    ``request.state.forwarded`` so repeated calls within a single
    request are free.
    """
    state = getattr(request, "state", None)
    if state is not None:
        cached_ip = getattr(state, "client_ip", None)
        cached_fwd = getattr(state, "forwarded", None)
        if cached_ip is not None and cached_fwd is not None:
            return cached_ip, cached_fwd

    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:
            state.client_ip = client_ip
            state.forwarded = forwarded
        except Exception:  # pragma: no cover - defensive
            # Some test fakes use a frozen ``state`` namespace; don't
            # crash — the helpers still return the right value, just
            # without caching.
            pass
    return client_ip, forwarded


def resolve_client_ip(request: Request) -> str:
    """Return the client IP to use for logging / auth / rate-limit.

    Always falls back to ``request.client.host`` when the gate fails
    or no usable forwarded value is present. Returns ``""`` only if
    even ``request.client`` is ``None`` (TestClient / UDS).
    """
    ip, _ = _resolve(request)
    return ip


def trusted_forwarded_headers(request: Request) -> dict[str, str]:
    """Return the sanitized ``X-Forwarded-*`` triple.

    Keys: ``"for"``, ``"proto"``, ``"host"``. Every value is the empty
    string when the gateway gate fails, so callers can use simple
    truthiness checks (``if fwd["proto"]: ...``).
    """
    _, fwd = _resolve(request)
    # Defensive copy: callers writing into the dict must not poison
    # the request-state cache.
    return dict(fwd)