diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index fb67e56..7d49411 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -81,7 +81,7 @@ jobs: run: python -m build - name: Check distribution metadata run: python -m twine check dist/* - - uses: actions/upload-artifact@v4 + - uses: actions/upload-artifact@v7 with: name: artifacts path: dist/* @@ -98,7 +98,7 @@ jobs: - build - ci steps: - - uses: actions/download-artifact@v4 + - uses: actions/download-artifact@v8 with: name: artifacts path: dist diff --git a/CHANGELOG.md b/CHANGELOG.md index 9a95130..9a17357 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,41 @@ +## 4.1.0 + +Best match (modern engine only; legacy is unchanged). Results can differ from 4.0.0 on well-formed input, +always toward a better address; use `algorithm="legacy"` for exact v3 results: +- Addresses are ranked public > private > link-local > loopback. Unspecified (`0.0.0.0`, `::`), + multicast, broadcast, and reserved addresses are never returned. Python reports multicast as + `is_global`, so v3 could return `224.0.0.1` as the client. +- Without `proxy_count` / `proxy_list`, the first public hop of a chain wins, not only the first hop: + `10.0.0.1, 177.139.233.139` now yields `177.139.233.139`. With `leftmost=False` the scan runs from the + right. With proxy settings, the client position is fixed exactly as before. +- `trusted_route` is `True` for any address resolved through a matching proxy config, including private + clients; v3 reported `False` for them. +- New exhaustive tests: every 1–3 hop chain over ten address kinds, in every leftmost / strict / proxy + configuration, checked against an independent reference model; every header assignment and dict order; + every spelling of a hop; never-worse-than-legacy across the whole matrix; seeded fuzzing. + +Enhance (modern engine only; legacy is unchanged): +- Parse RFC 7239 `Forwarded` elements by their `for=` value, including quoted, bracketed IPv6 with a port. + Previously `Forwarded` never produced an IP, so when it is present it can now resolve at its existing + precedence slot. Obfuscated hops (`for=unknown`, `for=_hidden`) count as invalid tokens. +- New default headers, added only between the 4.0.0 entries and `REMOTE_ADDR`, so none outranks a header + that resolved requests before: Azure Front Door `X-Azure-ClientIP`, DigitalOcean `DO-Connecting-IP`, + Envoy/Istio `X-Envoy-External-Address`, plus the missing `HTTP_X_CLIENT_IP` and raw `X-AppEngine-User-IP` + forms of headers already on the list. +- Header names match case-insensitively (`-` and `_` equivalent), so lowercase keys such as AWS Lambda's + work. Exact keys still take priority. + +Harden (modern engine only): +- Reject malformed tokens instead of truncating them: unclosed brackets (`[::1`), text after a bracket + (`[::1]junk`), and non-numeric, empty, or out-of-range ports (`1.2.3.4:abc`, `1.2.3.4:70000`). Note v3 + accepted `1.2.3.4:abc` as `1.2.3.4`. +- Non-string header values (`None`, bytes) are skipped instead of raising `AttributeError`. +- `proxy_list` entries are stripped of whitespace. An empty entry now raises `ValueError`: it used to match + every address and mark any spoofed chain as trusted. + +CI: +- Bump `actions/upload-artifact` to v7 and `actions/download-artifact` to v8, which run on Node 24. + ## 4.0.0 Community (thank you!): diff --git a/README.md b/README.md index aecbee9..74efd05 100644 --- a/README.md +++ b/README.md @@ -97,12 +97,29 @@ ip, trusted_route = ipw.get_client_ip(meta, strict=False) | Output | Description | | --- | --- | | `ip` | `IPv4Address`, `IPv6Address`, or `None` | -| `trusted_route` | `True` when `proxy_count` or `proxy_list` was configured and matched | +| `trusted_route` | `True` when `proxy_count` or `proxy_list` was configured and matched, for any returned IP (v3 only set it for public IPs) | ### Selection rules -Headers are checked in precedence order. The first **public** IP found wins; otherwise the first -**private** IP; otherwise the first **loopback** IP; otherwise `None`. +Headers are checked in precedence order. Every address is ranked: + +| Rank | Addresses | +| --- | --- | +| 1. public | globally routable | +| 2. private | RFC 1918, IPv6 ULA, CGNAT `100.64.0.0/10`, documentation ranges | +| 3. link-local | `169.254.0.0/16`, `fe80::/10` | +| 4. loopback | `127.0.0.0/8`, `::1` | +| never returned | `0.0.0.0`, `::`, multicast, broadcast, reserved | + +The first **public** IP wins. If none is found, the best-ranked IP wins, and the earlier header +wins a tie. + +Within one header, the client entry depends on your proxy settings: + +- **`proxy_count` / `proxy_list` set:** the entry just before your trusted proxies, exactly as in v3. +- **Neither set:** the first public entry in the chain, not only the first entry. So + `10.0.0.1, 177.139.233.139` yields `177.139.233.139` (v3 returned `10.0.0.1`). With + `leftmost=False` the chain is scanned from the right. ```mermaid flowchart TD @@ -115,13 +132,19 @@ flowchart TD E -->|yes| F["Pick the client entry"] F --> G{"Public IP?"} G -->|yes| H["Return (ip, trusted_route)"] - G -->|no| I["Keep as private or loopback fallback"] + G -->|no| I["Keep if it outranks the current fallback"] I --> B - B -->|no headers left| J["Return first private, else loopback, else None"] + B -->|no headers left| J["Return the best fallback, else None"] ``` +The legacy engine keeps v3's rules. A combination suite checks that the modern engine never returns a +worse address than legacy for the same input. + Ports are stripped (`1.2.3.4:8080`, `[2001:db8::1]:443`) and IPv4-mapped IPv6 addresses -(`::ffff:1.2.3.4`) are returned as plain IPv4. +(`::ffff:1.2.3.4`) are returned as plain IPv4. RFC 7239 `Forwarded` elements are read by their +`for=` value (`for="[2001:db8::1]:4711";proto=https`). Malformed tokens such as `[::1`, +`[::1]junk`, or `1.2.3.4:abc` are rejected rather than truncated. Header names match +case-insensitively, so lowercase keys (AWS Lambda / API Gateway v2) work too. ## Default header precedence @@ -133,7 +156,7 @@ Ports are stripped (`1.2.3.4:8080`, `[2001:db8::1]:443`) and IPv4-mapped IPv6 ad "HTTP_X_REAL_IP", "HTTP_X_FORWARDED", # Squid "HTTP_X_CLUSTER_CLIENT_IP", # Rackspace LB, Riverbed Stingray - "HTTP_FORWARDED_FOR", # RFC 7239 + "HTTP_FORWARDED_FOR", # de facto variant "HTTP_FORWARDED", # RFC 7239 "HTTP_CF_CONNECTING_IP", # Cloudflare "HTTP_TRUE_CLIENT_IP", # Cloudflare Enterprise, Akamai @@ -151,10 +174,22 @@ Ports are stripped (`1.2.3.4:8080`, `[2001:db8::1]:443`) and IPv4-mapped IPv6 ad "FLY-CLIENT-IP", "FORWARDED", "CLIENT-IP", + # added after 4.0.0 — always below every earlier entry, above REMOTE_ADDR + "HTTP_X_CLIENT_IP", # Microsoft Azure (Django/WSGI form) + "X-APPENGINE-USER-IP", # Google App Engine (raw form) + "HTTP_X_AZURE_CLIENTIP", # Azure Front Door + "X-AZURE-CLIENTIP", + "HTTP_DO_CONNECTING_IP", # DigitalOcean App Platform + "DO-CONNECTING-IP", + "HTTP_X_ENVOY_EXTERNAL_ADDRESS", # Envoy / Istio + "X-ENVOY-EXTERNAL-ADDRESS", "REMOTE_ADDR", # direct connection ) ``` +Headers released earlier never move. New ones are added only just above `REMOTE_ADDR`, so an +upgrade can never let a new header outrank one that already resolved your requests. + Narrow it to what your infrastructure actually sets: ```python diff --git a/python_ipware/__version__.py b/python_ipware/__version__.py index ce1305b..7039708 100644 --- a/python_ipware/__version__.py +++ b/python_ipware/__version__.py @@ -1 +1 @@ -__version__ = "4.0.0" +__version__ = "4.1.0" diff --git a/python_ipware/modern/defaults.py b/python_ipware/modern/defaults.py index fe51141..a1e3338 100644 --- a/python_ipware/modern/defaults.py +++ b/python_ipware/modern/defaults.py @@ -1,8 +1,17 @@ """Default header precedence for the modern engine. -Superset of the v3 list, adding widely-deployed CDN/edge headers. Order is -most-to-least trustworthy for a typical deployment; the first header that -yields a usable IP wins. +Superset of the v3 list, adding widely-deployed CDN/edge headers. Headers are +scanned in order; the first one that yields a globally routable IP wins, with +private and then loopback addresses as fallbacks. + +Ordering rule: existing entries never move. New headers are added only in the +block just above ``REMOTE_ADDR``, so a new header can outrank the raw socket +address but never a header that already resolved a request in an earlier +release. + +Every default here can be forged by a client that reaches the app directly. +If all traffic arrives through one known edge, pass an explicit +``precedence`` naming only that edge's header. """ DEFAULT_PRECEDENCE: tuple[str, ...] = ( @@ -12,8 +21,8 @@ "HTTP_X_REAL_IP", "HTTP_X_FORWARDED", "HTTP_X_CLUSTER_CLIENT_IP", - "HTTP_FORWARDED_FOR", - "HTTP_FORWARDED", + "HTTP_FORWARDED_FOR", # de facto variant; not defined by RFC 7239 + "HTTP_FORWARDED", # RFC 7239 (for=...;proto=...), parsed per element "HTTP_CF_CONNECTING_IP", # Cloudflare "HTTP_TRUE_CLIENT_IP", # Cloudflare Enterprise / Akamai "HTTP_FASTLY_CLIENT_IP", # Fastly / Firebase @@ -30,5 +39,14 @@ "FLY-CLIENT-IP", "FORWARDED", "CLIENT-IP", - "REMOTE_ADDR", + # --- added after 4.0.0: below every earlier entry, above REMOTE_ADDR --- + "HTTP_X_CLIENT_IP", # Azure X-Client-IP in Django/WSGI form + "X-APPENGINE-USER-IP", # Google App Engine, raw header form + "HTTP_X_AZURE_CLIENTIP", # Azure Front Door + "X-AZURE-CLIENTIP", + "HTTP_DO_CONNECTING_IP", # DigitalOcean App Platform + "DO-CONNECTING-IP", + "HTTP_X_ENVOY_EXTERNAL_ADDRESS", # Envoy / Istio + "X-ENVOY-EXTERNAL-ADDRESS", + "REMOTE_ADDR", # direct connection; always last ) diff --git a/python_ipware/modern/engine.py b/python_ipware/modern/engine.py index 5f8fbcd..ae96b7d 100644 --- a/python_ipware/modern/engine.py +++ b/python_ipware/modern/engine.py @@ -1,23 +1,28 @@ """The modern python-ipware engine. -Behavior-compatible with the v3 algorithm on the essentials, but cleaner and -hardened: - -* Superset header precedence (adds True-Client-IP, Fastly, App Engine, Azure). -* Robust IPv6 / bracketed-port / IPv4-mapped parsing. -* Same "best IP" fallback ladder: prefer a globally routable address; else the - first private; else loopback. +Same inputs, outputs and proxy semantics as v3, with a better best-match: + +* Superset header precedence (Forwarded parsing, more CDN / edge headers). +* Robust IPv6 / bracketed-port / IPv4-mapped / RFC 7239 parsing. +* Explicit ranking: global > private > link-local > loopback. Unspecified, + multicast, broadcast and reserved addresses are never returned. +* Without trusted-proxy config, the first *globally routable* hop of a chain + wins, not just the first hop, so ``10.0.0.1, 177.139.233.139`` yields the + public address. With ``proxy_count`` / ``proxy_list`` the client position is + fixed by the config, exactly as in v3. * Same ``strict`` semantics for proxy_count / proxy_list validation. * Trusted-proxy matching anchored to the end of the chain. Each ``proxy_list`` entry is either a CIDR network (``"100.64.0.0/10"``, ``"fd7a:115c:a1e0::/48"``) matched by real network membership, or a plain string prefix (``"10.1."``). +* ``trusted_route`` is True whenever the returned IP came from a chain that + passed the configured proxy validation, whatever its tier. """ import ipaddress from typing import Optional, Union from .defaults import DEFAULT_PRECEDENCE -from .parsers import IpAddressType, split_proxy_chain +from .parsers import TIER_GLOBAL, TIER_REJECT, IpAddressType, ip_tier, split_proxy_chain OptionalIp = Optional[IpAddressType] IpNetworkType = Union[ipaddress.IPv4Network, ipaddress.IPv6Network] @@ -55,23 +60,52 @@ def __init__( raise ValueError("proxy_count must be non-negative") if proxy_list is not None and not all(isinstance(p, str) for p in proxy_list): raise ValueError("All elements in the proxy list must be strings.") + proxy_list = [p.strip() for p in proxy_list or []] + # An empty prefix matches every address, which would mark any spoofed + # chain as trusted. It is always a misconfiguration (e.g. a trailing + # comma in an env var), so fail loudly instead. + if any(not p for p in proxy_list): + raise ValueError("proxy_list entries must not be empty.") self.precedence = precedence or DEFAULT_PRECEDENCE self.leftmost = leftmost self.proxy_count = proxy_count - self.proxy_list = list(proxy_list or []) + self.proxy_list = proxy_list self._proxy_matchers = [_compile_proxy_matcher(p) for p in self.proxy_list] # -- meta access -------------------------------------------------------- - def _get_meta_value(self, meta: dict[str, str], key: str) -> str: + @staticmethod + def _fold(key: str) -> str: + return key.upper().replace("-", "_") + + def _get_meta_value( + self, + meta: dict[str, str], + key: str, + folded: Optional[dict[str, object]] = None, + ) -> str: meta = meta or {} - return meta.get(key, meta.get(key.replace("_", "-"), "")).strip() + value = meta.get(key) + if value is None: + value = meta.get(key.replace("_", "-")) + # Exact keys win; the folded view only fills gaps, so lowercase keys + # (AWS Lambda / API Gateway v2, raw ASGI dicts) still match. + if value is None and folded is not None: + value = folded.get(self._fold(key)) + # Header values are text; anything else (None, bytes, lists from a + # misbehaving adapter) is ignored rather than crashing the lookup. + return value.strip() if isinstance(value, str) else "" def _get_meta_values(self, meta: dict[str, str]) -> list[str]: + meta = meta or {} + folded: dict[str, object] = {} + for k, v in meta.items(): + if isinstance(k, str): + folded.setdefault(self._fold(k), v) values: list[str] = [] for key in self.precedence: - value = self._get_meta_value(meta, key) + value = self._get_meta_value(meta, key, folded) if value: values.append(value) return values @@ -103,22 +137,32 @@ def _proxy_list_valid(self, chain: list[IpAddressType], strict: bool) -> bool: # -- selection ---------------------------------------------------------- def _best_from_chain(self, chain: list[IpAddressType]) -> tuple[OptionalIp, bool]: - # ``chain`` is already client-first (see get_client_ip). - if not chain: - return None, False + # ``chain`` is already client-first (see get_client_ip) and non-empty. if self.proxy_list: return chain[-(len(self.proxy_list) + 1)], True if self.proxy_count is not None: return chain[-(self.proxy_count + 1)], True - return chain[0], False + # No trusted-proxy config, so no position in the chain is verified. + # Take the first globally routable hop; otherwise the best-ranked hop, + # earliest on ties. This never picks a worse address than v3's chain[0]. + best: OptionalIp = None + best_tier = TIER_REJECT + for ip in chain: + tier = ip_tier(ip) + if tier == TIER_GLOBAL: + return ip, False + if tier > best_tier: + best, best_tier = ip, tier + return best, False # -- public API --------------------------------------------------------- - def get_client_ip( - self, meta: dict[str, str], strict: bool = False - ) -> tuple[OptionalIp, bool]: - loopback: list[IpAddressType] = [] - private: list[IpAddressType] = [] + def get_client_ip(self, meta: dict[str, str], strict: bool = False) -> tuple[OptionalIp, bool]: + # Best non-global candidate so far. Strictly-greater comparison keeps + # the earliest header on ties, preserving header precedence. + fallback: OptionalIp = None + fallback_tier = TIER_REJECT + fallback_trusted = False for raw in self._get_meta_values(meta): chain = split_proxy_chain(raw, strict) @@ -136,15 +180,10 @@ def get_client_ip( ip, trusted = self._best_from_chain(chain) if ip is None: continue - if ip.is_global: + tier = ip_tier(ip) + if tier == TIER_GLOBAL: return ip, trusted - if ip.is_loopback: - loopback.append(ip) - else: - private.append(ip) - - if private: - return private[0], False - if loopback: - return loopback[0], False - return None, False + if tier > fallback_tier: + fallback, fallback_tier, fallback_trusted = ip, tier, trusted + + return fallback, fallback_trusted diff --git a/python_ipware/modern/parsers.py b/python_ipware/modern/parsers.py index e1cbf92..2f6fde1 100644 --- a/python_ipware/modern/parsers.py +++ b/python_ipware/modern/parsers.py @@ -1,7 +1,8 @@ """Framework-agnostic IP parsing helpers for the modern engine. Pure stdlib. Knows how to clean raw header tokens, strip ports/brackets, -validate IPv4/IPv6, and split proxy chains. +read RFC 7239 ``Forwarded`` elements, validate IPv4/IPv6, and split proxy +chains. """ import ipaddress @@ -9,26 +10,82 @@ IpAddressType = Union[ipaddress.IPv4Address, ipaddress.IPv6Address] +# How good an address is as a client IP; higher wins. REJECT is never returned. +TIER_REJECT = 0 +TIER_LOOPBACK = 1 +TIER_LINK_LOCAL = 2 +TIER_PRIVATE = 3 # RFC 1918, ULA, CGNAT 100.64/10, documentation ranges, ... +TIER_GLOBAL = 4 + + +def ip_tier(ip: IpAddressType) -> int: + """Rank ``ip`` as a client address candidate. + + Check order matters: Python reports multicast (``224.0.0.1``, ``ff02::1``) + and the deprecated ``::a.b.c.d`` form as ``is_global``, and ``::1`` as + ``is_reserved``, so those are resolved before ``is_global`` is trusted. + Unspecified, multicast, broadcast and reserved addresses can never be a + real client and are rejected outright. + """ + if ip.is_unspecified or ip.is_multicast: + return TIER_REJECT + if ip.is_loopback: + return TIER_LOOPBACK + if ip.is_reserved: # 240.0.0.0/4 incl. 255.255.255.255; unallocated IPv6 + return TIER_REJECT + if ip.is_global: + return TIER_GLOBAL + if ip.is_link_local: + return TIER_LINK_LOCAL + return TIER_PRIVATE + + +def _is_port(value: str) -> bool: + # isascii() guards against Unicode digits such as "²" that isdigit() accepts. + return value.isascii() and value.isdigit() and int(value) <= 65535 + def strip_port(value: str) -> str: - """Remove a trailing ``:port`` (IPv4) or ``[addr]:port`` (IPv6) suffix.""" + """Remove a trailing ``:port`` (IPv4) or ``[addr]:port`` (IPv6) suffix. + + Returns ``""`` for a malformed token (unclosed bracket, text after the + bracket, or a non-numeric/out-of-range port) so it is rejected rather than + silently truncated into something that looks valid. + """ value = value.strip() if not value: return value if value.startswith("["): # [addr] or [addr]:port end = value.find("]") - if end != -1: - return value[1:end] - return value.lstrip("[") + if end == -1: + return "" + rest = value[end + 1 :] + if rest and not (rest.startswith(":") and _is_port(rest[1:])): + return "" + return value[1:end] if value.count(":") == 1: # IPv4:port - host, _, _ = value.partition(":") - return host + host, _, port = value.partition(":") + return host if _is_port(port) else "" return value # bare IPv6 or bare IPv4 +def forwarded_for(element: str) -> str: + """Return the ``for=`` value of one RFC 7239 ``Forwarded`` element. + + ``for=192.0.2.60;proto=http;by=203.0.113.43`` -> ``192.0.2.60``. Parameter + names are case-insensitive. Returns ``""`` when the element has no ``for`` + parameter, so the hop counts as invalid instead of being guessed at. + """ + for pair in element.split(";"): + key, sep, val = pair.partition("=") + if sep and key.strip().lower() == "for": + return val.strip() + return "" + + def clean_ip(value: Optional[str]) -> str: """Normalize a raw candidate token into a bare IP string.""" if not value: @@ -55,16 +112,20 @@ def parse_ip(value: Optional[str]) -> Optional[IpAddressType]: def split_proxy_chain(raw: Optional[str], strict: bool = False) -> Optional[list[IpAddressType]]: """Split a comma-separated proxy chain into ordered ``ip_address`` objects. - Order is preserved left-to-right as it appears in the header. In strict - mode, any invalid or empty token makes the whole chain invalid (returns - None), since a malformed header should not be trusted. Otherwise invalid - and empty tokens are skipped. + Order is preserved left-to-right as it appears in the header. Tokens in + RFC 7239 form (``for=...;proto=...``) are reduced to their ``for`` value; + obfuscated identifiers such as ``for=unknown`` or ``for=_hidden`` are not + IPs and count as invalid. In strict mode, any invalid or empty token makes + the whole chain invalid (returns None), since a malformed header should not + be trusted. Otherwise invalid and empty tokens are skipped. """ if not raw: return [] result: list[IpAddressType] = [] for token in raw.split(","): - ip = parse_ip(token) + # No plain IP token contains "=", only RFC 7239 Forwarded elements do. + candidate = forwarded_for(token) if "=" in token else token + ip = parse_ip(candidate) if ip is not None: result.append(ip) elif strict: diff --git a/python_ipware/router.py b/python_ipware/router.py index d733371..ed76b3f 100644 --- a/python_ipware/router.py +++ b/python_ipware/router.py @@ -7,11 +7,13 @@ The ``algorithm`` selector chooses between them. ``"auto"`` (the default) is a clean alias for ``"modern"`` — the enhanced engine is where development moves -forward. On well-formed headers it returns the same result as v3 (the full v3 -suite and a legacy-vs-modern differential test pass); it differs only on -malformed values, as documented in the CHANGELOG. ``legacy`` remains available as an explicit -escape hatch for projects that need byte-for-byte v3 behavior. There is no -silent runtime fallback, so behavior stays predictable. +forward. It passes the full v3 suite, and a differential test checks it never +returns a worse address than v3. It picks better where v3 did not: a public +hop behind a private first hop, no multicast / unspecified results, and +``trusted_route`` for private clients behind trusted proxies (see CHANGELOG). +``legacy`` remains available as an explicit escape hatch for projects that +need byte-for-byte v3 behavior. There is no silent runtime fallback, so +behavior stays predictable. from python_ipware import IpWare diff --git a/tests/tests_modern_matrix.py b/tests/tests_modern_matrix.py new file mode 100644 index 0000000..b56518a --- /dev/null +++ b/tests/tests_modern_matrix.py @@ -0,0 +1,361 @@ +"""Exhaustive combination tests for the modern engine. + +The core check compares the engine with an independent reference model +(``_oracle``). The model never calls the engine's parser or ranking: every +test token has a hand-assigned address and tier in ``TRUTH``, so a bug in +``parse_ip`` or ``ip_tier`` cannot hide by being shared with the oracle. + +Covered exhaustively: + +* every ordered chain of 1-3 hops drawn from a pool of 10 token kinds + (global v4/v6, a proxy, private v4/v6, link-local, loopback, unspecified, + multicast, garbage), in every configuration of leftmost, strict, + proxy_count and proxy_list; +* every assignment of chains to three headers, in both dict insertion orders; +* every formatting variant of the same hop (port, brackets, quotes, RFC 7239 + ``for=``, whitespace, IPv4-mapped, case, zero-expansion); +* never-worse-than-legacy over the whole single-header matrix; +* a seeded fuzz run over junk input for crash-freedom and output invariants. +""" + +import ipaddress +import itertools +import random +import unittest +from typing import ClassVar, Optional + +from python_ipware import IpWare +from python_ipware.modern.parsers import ( + TIER_GLOBAL, + TIER_LINK_LOCAL, + TIER_LOOPBACK, + TIER_PRIVATE, + TIER_REJECT, + ip_tier, +) + +G, P, LL, LB, REJ = TIER_GLOBAL, TIER_PRIVATE, TIER_LINK_LOCAL, TIER_LOOPBACK, TIER_REJECT + +# token -> (address, tier) by hand, or None for a token that is not an IP. +TRUTH: dict[str, Optional[tuple[str, int]]] = { + "177.139.233.139": ("177.139.233.139", G), + "2606:4700::6810:84e5": ("2606:4700::6810:84e5", G), + "198.84.193.157": ("198.84.193.157", G), # doubles as the trusted proxy + "10.0.0.1": ("10.0.0.1", P), + "fd00::1": ("fd00::1", P), + "169.254.1.1": ("169.254.1.1", LL), + "127.0.0.1": ("127.0.0.1", LB), + "0.0.0.0": ("0.0.0.0", REJ), + "224.0.0.1": ("224.0.0.1", REJ), + "unknown": None, +} +POOL = tuple(TRUTH) + +# (proxy_count, proxy_list) configurations, including both set at once. +CONFIGS = ( + (None, None), + (0, None), + (1, None), + (2, None), + (None, ["198.84.193.157"]), + (None, ["198.84."]), + (None, ["10.0.0.0/8"]), + (1, ["198.84.193.157"]), +) + + +def _chains(max_len: int = 3): + for n in range(1, max_len + 1): + yield from itertools.product(POOL, repeat=n) + + +def _proxy_ok(address: str, pattern: str) -> bool: + ip = ipaddress.ip_address(address) + if "/" in pattern: + net = ipaddress.ip_network(pattern, strict=False) + return ip.version == net.version and ip in net + return address.startswith(pattern) + + +def _oracle(raw_values, leftmost, proxy_count, proxy_list, strict): + """Reference model of the documented modern algorithm. + + ``raw_values`` are header values already in precedence order. + """ + configured = proxy_count is not None or bool(proxy_list) + best = None # (address, tier, trusted); first header wins ties + for raw in raw_values: + hops = [TRUTH[t.strip()] for t in raw.split(",")] + if strict and any(h is None for h in hops): + continue + chain = [h for h in hops if h is not None] + if not leftmost: + chain.reverse() + if not chain: + continue + proxies = len(chain) - 1 + if proxy_count is not None and (proxies < proxy_count or (strict and proxies != proxy_count)): + continue + if proxy_list: + n = len(proxy_list) + if proxies < n or (strict and proxies != n): + continue + if not all(_proxy_ok(addr, pat) for (addr, _), pat in zip(chain[-n:], proxy_list)): + continue + if configured: + n = len(proxy_list) if proxy_list else proxy_count + address, tier = chain[-(n + 1)] + else: + globals_ = [h for h in chain if h[1] == G] + address, tier = globals_[0] if globals_ else max(chain, key=lambda h: h[1]) + if tier == REJ: + continue + if tier == G: + return ipaddress.ip_address(address), configured + if best is None or tier > best[1]: + best = (address, tier, configured) + if best is None: + return None, False + return ipaddress.ip_address(best[0]), best[2] + + +class TestTierTable(unittest.TestCase): + """The ranking every other test relies on, checked against the labels.""" + + def test_pool_labels(self): + for token, truth in TRUTH.items(): + if truth is None: + continue + with self.subTest(token=token): + self.assertEqual(ip_tier(ipaddress.ip_address(truth[0])), truth[1]) + + def test_extra_classes(self): + cases = { + "8.8.8.8": G, "2001:4860::8888": G, + "192.168.1.1": P, "172.16.0.1": P, "100.64.0.1": P, "203.0.113.10": P, "2001:db8::1": P, + "fe80::1": LL, "::1": LB, + "::": REJ, "ff02::1": REJ, "255.255.255.255": REJ, "240.0.0.1": REJ, "::8.8.8.8": REJ, + } # fmt: skip + for address, tier in cases.items(): + with self.subTest(address=address): + self.assertEqual(ip_tier(ipaddress.ip_address(address)), tier) + + +class TestSingleHeaderMatrix(unittest.TestCase): + def test_every_chain_and_config_matches_oracle(self): + checked = 0 + for hops in _chains(): + raw = ", ".join(hops) + meta = {"HTTP_X_FORWARDED_FOR": raw} + for proxy_count, proxy_list in CONFIGS: + for leftmost in (True, False): + ipw = IpWare(leftmost=leftmost, proxy_count=proxy_count, proxy_list=proxy_list) + for strict in (False, True): + expected = _oracle([raw], leftmost, proxy_count, proxy_list, strict) + actual = ipw.get_client_ip(meta, strict) + if actual != expected: + self.fail( + f"{raw!r} leftmost={leftmost} count={proxy_count} " + f"list={proxy_list} strict={strict}: {actual} != {expected}" + ) + checked += 1 + self.assertEqual(checked, (10 + 100 + 1000) * len(CONFIGS) * 4) + + +class TestMultiHeaderMatrix(unittest.TestCase): + HEADERS = ("HTTP_X_FORWARDED_FOR", "HTTP_X_REAL_IP", "REMOTE_ADDR") + VALUES = ( + None, # header absent + "177.139.233.139", + "2606:4700::6810:84e5", + "10.0.0.1", + "127.0.0.1", + "0.0.0.0", + "unknown", + "10.0.0.1, 177.139.233.139", + "169.254.1.1, fd00::1", + "177.139.233.139, 198.84.193.157", + "10.0.0.1, 198.84.193.157", + ) + CONFIGS = ((None, None), (1, None), (None, ["198.84.193.157"])) + + def test_every_header_assignment_and_order(self): + checked = 0 + for values in itertools.product(self.VALUES, repeat=len(self.HEADERS)): + present = [(h, v) for h, v in zip(self.HEADERS, values) if v is not None] + for proxy_count, proxy_list in self.CONFIGS: + for leftmost in (True, False): + ipw = IpWare( + precedence=self.HEADERS, leftmost=leftmost, proxy_count=proxy_count, proxy_list=proxy_list + ) + for strict in (False, True): + expected = _oracle([v for _, v in present], leftmost, proxy_count, proxy_list, strict) + # Dict insertion order must never matter, only precedence. + for order in (present, present[::-1]): + actual = ipw.get_client_ip(dict(order), strict) + if actual != expected: + self.fail( + f"{dict(order)} leftmost={leftmost} count={proxy_count} " + f"list={proxy_list} strict={strict}: {actual} != {expected}" + ) + checked += 1 + self.assertEqual(checked, len(self.VALUES) ** 3 * len(self.CONFIGS) * 2 * 2 * 2) + + +class TestFormattingInvariance(unittest.TestCase): + """Every spelling of a hop must resolve exactly like the plain address.""" + + RENDERINGS: ClassVar[dict[str, tuple[str, ...]]] = { + "177.139.233.139": ( + "177.139.233.139:8080", + '"177.139.233.139"', + " 177.139.233.139 ", + "for=177.139.233.139", + 'for="177.139.233.139:4711";proto=https;by=198.84.193.158', + "::ffff:177.139.233.139", + "[::ffff:177.139.233.139]:443", + ), + "2606:4700::6810:84e5": ( + "[2606:4700::6810:84e5]", + "[2606:4700::6810:84e5]:443", + "2606:4700::6810:84E5", + "2606:4700:0:0:0:0:6810:84e5", + 'for="[2606:4700::6810:84e5]:4711"', + "'2606:4700::6810:84e5'", + ), + "10.0.0.1": ("10.0.0.1:443", 'for="10.0.0.1"', "Proto=http;FOR=10.0.0.1"), + "198.84.193.157": ("198.84.193.157:80", "for=198.84.193.157;by=_edge"), + } + + def test_all_rendering_combinations(self): + templates = ( + ("177.139.233.139", "198.84.193.157"), + ("10.0.0.1", "177.139.233.139"), + ("2606:4700::6810:84e5", "10.0.0.1", "198.84.193.157"), + ) + configs = ((None, None), (1, None), (None, ["198.84.193.157"])) + checked = 0 + for template in templates: + plain_meta = {"HTTP_X_FORWARDED_FOR": ", ".join(template)} + options = [(hop, *self.RENDERINGS[hop]) for hop in template] + for spelled in itertools.product(*options): + meta = {"HTTP_X_FORWARDED_FOR": ", ".join(spelled)} + for proxy_count, proxy_list in configs: + for leftmost in (True, False): + ipw = IpWare(leftmost=leftmost, proxy_count=proxy_count, proxy_list=proxy_list) + for strict in (False, True): + with self.subTest(meta=meta, count=proxy_count, list=proxy_list, strict=strict): + self.assertEqual(ipw.get_client_ip(meta, strict), ipw.get_client_ip(plain_meta, strict)) + checked += 1 + self.assertGreater(checked, 1000) + + +class TestNeverWorseThanLegacy(unittest.TestCase): + @staticmethod + def _rank(ip): + if ip is None: + return 0 + tier = ip_tier(ip) + return -1 if tier == REJ else tier + + def test_whole_single_header_matrix(self): + improved = 0 + for hops in _chains(): + meta = {"HTTP_X_FORWARDED_FOR": ", ".join(hops)} + for proxy_count, proxy_list in CONFIGS: + configured = proxy_count is not None or bool(proxy_list) + for leftmost in (True, False): + kw = {"leftmost": leftmost, "proxy_count": proxy_count, "proxy_list": proxy_list} + legacy, modern = IpWare(algorithm="legacy", **kw), IpWare(**kw) + for strict in (False, True): + l_ip, l_trusted = legacy.get_client_ip(meta, strict) + m_ip, m_trusted = modern.get_client_ip(meta, strict) + where = f"{meta} {kw} strict={strict}: modern={m_ip, m_trusted} legacy={l_ip, l_trusted}" + self.assertGreaterEqual(self._rank(m_ip), self._rank(l_ip), where) + if self._rank(l_ip) == G: + self.assertEqual((m_ip, m_trusted), (l_ip, l_trusted), where) + if m_trusted: + self.assertTrue(configured, where) + improved += self._rank(m_ip) > self._rank(l_ip) + self.assertGreater(improved, 1000) + + +class TestBestMatchExamples(unittest.TestCase): + """Readable examples of each improvement over v3.""" + + def test_public_hop_behind_private_first_hop(self): + # v3 returned 10.0.0.1 here. + ip, _ = IpWare().get_client_ip({"HTTP_X_FORWARDED_FOR": "10.0.0.1, 177.139.233.139, 198.84.193.157"}) + self.assertEqual(str(ip), "177.139.233.139") + + def test_rightmost_scans_from_the_right(self): + meta = {"HTTP_X_FORWARDED_FOR": "6.6.6.6, 177.139.233.139, 10.0.0.2"} + ip, _ = IpWare(leftmost=False).get_client_ip(meta) + self.assertEqual(str(ip), "177.139.233.139") + + def test_junk_addresses_never_returned(self): + for junk in ("0.0.0.0", "::", "224.0.0.1", "ff02::1", "255.255.255.255", "240.0.0.1", "::8.8.8.8"): + with self.subTest(junk=junk): + self.assertEqual(IpWare().get_client_ip({"REMOTE_ADDR": junk}), (None, False)) + + def test_multicast_is_not_mistaken_for_global(self): + # Python reports 224.0.0.1 as is_global; v3 returned it as the client. + meta = {"HTTP_X_FORWARDED_FOR": "224.0.0.1", "REMOTE_ADDR": "10.0.0.5"} + ip, _ = IpWare().get_client_ip(meta) + self.assertEqual(str(ip), "10.0.0.5") + + def test_link_local_ranks_below_private(self): + meta = {"HTTP_X_FORWARDED_FOR": "169.254.1.1", "REMOTE_ADDR": "10.0.0.5"} + ip, _ = IpWare().get_client_ip(meta) + self.assertEqual(str(ip), "10.0.0.5") + + def test_link_local_ranks_above_loopback(self): + meta = {"HTTP_X_FORWARDED_FOR": "127.0.0.1", "REMOTE_ADDR": "fe80::1"} + ip, _ = IpWare().get_client_ip(meta) + self.assertEqual(str(ip), "fe80::1") + + def test_private_client_through_trusted_proxy_is_trusted(self): + # v3 reported trusted=False for any non-global client. + ipw = IpWare(proxy_list=["198.84.193.157"]) + meta = {"HTTP_X_FORWARDED_FOR": "10.0.0.1, 198.84.193.157"} + self.assertEqual(ipw.get_client_ip(meta, strict=True), (ipaddress.ip_address("10.0.0.1"), True)) + + def test_proxy_config_still_fixes_the_position(self): + # With a trusted proxy the client slot is fixed; a public hop further left is ignored. + ipw = IpWare(proxy_count=1) + meta = {"HTTP_X_FORWARDED_FOR": "177.139.233.139, 10.0.0.1, 198.84.193.157"} + self.assertEqual(ipw.get_client_ip(meta), (ipaddress.ip_address("10.0.0.1"), True)) + + +class TestFuzz(unittest.TestCase): + FRAGMENTS = ( + "", " ", ",", ";", "=", '"', "'", "[", "]", ":", "::", "%eth0", "for=", "by=", "proto=https", + "unknown", "_hidden", "1", "255", "256", "999", ".", "1.2.3", "::ffff:", "fe80::", "0x7f", + "177.139.233.139", "10.0.0.1", "2606:4700::1", "127.0.0.1", "0.0.0.0", "224.0.0.1", + "\u00b2", "\uff11", "\x00", "65535", "70000", "\t", + ) # fmt: skip + HEADERS = ("HTTP_X_FORWARDED_FOR", "HTTP_FORWARDED", "x-real-ip", "REMOTE_ADDR") + + def test_random_junk_is_safe(self): + rng = random.Random(20260922) + for _ in range(3000): + meta = { + h: "".join(rng.choice(self.FRAGMENTS) for _ in range(rng.randint(0, 12))) + for h in rng.sample(self.HEADERS, rng.randint(1, len(self.HEADERS))) + } + proxy_count, proxy_list = rng.choice(CONFIGS) + ipw = IpWare(leftmost=rng.random() < 0.5, proxy_count=proxy_count, proxy_list=proxy_list) + strict = rng.random() < 0.5 + with self.subTest(meta=meta, count=proxy_count, list=proxy_list, strict=strict): + ip, trusted = ipw.get_client_ip(meta, strict) + self.assertEqual((ip, trusted), ipw.get_client_ip(meta, strict)) # deterministic + if ip is None: + self.assertFalse(trusted) + else: + self.assertNotEqual(ip_tier(ip), REJ) + if trusted: + self.assertTrue(proxy_count is not None or bool(proxy_list)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/tests_router.py b/tests/tests_router.py index b0d4c85..708cfdd 100644 --- a/tests/tests_router.py +++ b/tests/tests_router.py @@ -4,6 +4,8 @@ from typing import ClassVar from python_ipware import IpWare, LegacyIpWare, ModernIpWare +from python_ipware.modern.defaults import DEFAULT_PRECEDENCE +from python_ipware.modern.parsers import TIER_GLOBAL, TIER_REJECT, ip_tier logging.disable(logging.CRITICAL) @@ -160,9 +162,146 @@ def test_non_strict_skips_empty_token(self): self.assertEqual(str(ip), "1.2.3.4") -class TestModernMatchesLegacy(unittest.TestCase): - """Differential check: on well-formed input, modern must agree with legacy - for every combination of direction, proxy_count, proxy_list and strict.""" +class TestModernRfc7239Forwarded(unittest.TestCase): + """RFC 7239 ``Forwarded`` elements are reduced to their ``for=`` value.""" + + def test_ipv4_with_params(self): + meta = {"HTTP_FORWARDED": "for=203.0.113.60;proto=https;by=198.51.100.1"} + ip, _ = IpWare().get_client_ip(meta) + self.assertEqual(str(ip), "203.0.113.60") + + def test_quoted_bracketed_ipv6_with_port(self): + meta = {"HTTP_FORWARDED": 'for="[2001:4860::17]:4711"'} + ip, _ = IpWare().get_client_ip(meta) + self.assertEqual(str(ip), "2001:4860::17") + + def test_param_name_case_insensitive(self): + meta = {"HTTP_FORWARDED": "proto=http;For=203.0.113.61"} + ip, _ = IpWare().get_client_ip(meta) + self.assertEqual(str(ip), "203.0.113.61") + + def test_multi_hop_with_proxy_count(self): + # Globally routable addresses: the trusted flag is only kept for a global client. + meta = {"HTTP_FORWARDED": "for=177.139.233.139, for=198.84.193.157;by=198.84.193.158"} + ip, trusted = IpWare(proxy_count=1).get_client_ip(meta, strict=True) + self.assertEqual(str(ip), "177.139.233.139") + self.assertTrue(trusted) + + def test_obfuscated_hop_strict_rejects(self): + meta = {"HTTP_FORWARDED": "for=unknown, for=198.51.100.7"} + self.assertEqual(IpWare().get_client_ip(meta, strict=True), (None, False)) + + def test_obfuscated_hop_non_strict_skips(self): + meta = {"HTTP_FORWARDED": "for=_hidden, for=198.51.100.7"} + ip, _ = IpWare().get_client_ip(meta) + self.assertEqual(str(ip), "198.51.100.7") + + def test_element_without_for_is_invalid(self): + meta = {"HTTP_FORWARDED": "proto=https;by=198.51.100.1"} + self.assertEqual(IpWare().get_client_ip(meta), (None, False)) + + +class TestModernMalformedRejected(unittest.TestCase): + """Malformed tokens are rejected, never truncated into a valid-looking IP.""" + + def test_malformed_tokens(self): + for value in ( + "[2606:4700::1", # unclosed bracket + "[2606:4700::1]junk", # text after the bracket + "8.8.8.8:abc", # non-numeric port + "8.8.8.8:70000", # port out of range + "8.8.8.8:\u00b2", # Unicode digit that isdigit() accepts + "8.8.8.8:", # empty port + ): + with self.subTest(value=value): + self.assertEqual(IpWare().get_client_ip({"REMOTE_ADDR": value}), (None, False)) + + def test_valid_ports_still_accepted(self): + for value, expected in (("8.8.8.8:443", "8.8.8.8"), ("[2606:4700::1]:65535", "2606:4700::1")): + with self.subTest(value=value): + ip, _ = IpWare().get_client_ip({"REMOTE_ADDR": value}) + self.assertEqual(str(ip), expected) + + +class TestModernMetaHardening(unittest.TestCase): + def test_non_string_values_are_ignored(self): + for bad in (None, b"1.2.3.4", ["1.2.3.4"], 42): + with self.subTest(bad=bad): + meta = {"HTTP_X_FORWARDED_FOR": bad, "REMOTE_ADDR": "8.8.8.8"} + ip, _ = IpWare().get_client_ip(meta) + self.assertEqual(str(ip), "8.8.8.8") + + def test_lowercase_keys_match(self): + # AWS Lambda / API Gateway v2 deliver lowercase header names. + meta = {"x-forwarded-for": "203.0.113.70, 198.51.100.1"} + ip, _ = IpWare().get_client_ip(meta) + self.assertEqual(str(ip), "203.0.113.70") + + def test_exact_key_beats_folded_key(self): + meta = {"X_FORWARDED_FOR": "203.0.113.71", "x-forwarded-for": "203.0.113.72"} + ip, _ = IpWare().get_client_ip(meta) + self.assertEqual(str(ip), "203.0.113.71") + + def test_empty_proxy_entry_raises(self): + for plist in ([""], [" "], ["10.0.0.", ""]): + with self.subTest(plist=plist), self.assertRaises(ValueError): + IpWare(proxy_list=plist) + + def test_proxy_prefix_whitespace_is_stripped(self): + ipw = IpWare(proxy_list=[" 198.84. "]) + meta = {"HTTP_X_FORWARDED_FOR": "177.139.233.139, 198.84.193.157"} + self.assertEqual(ipw.get_client_ip(meta), (ipaddress.ip_address("177.139.233.139"), True)) + + +class TestModernPrecedence(unittest.TestCase): + V400_ORDER = ( + "X_FORWARDED_FOR", "HTTP_X_FORWARDED_FOR", "HTTP_CLIENT_IP", "HTTP_X_REAL_IP", + "HTTP_X_FORWARDED", "HTTP_X_CLUSTER_CLIENT_IP", "HTTP_FORWARDED_FOR", "HTTP_FORWARDED", + "HTTP_CF_CONNECTING_IP", "HTTP_TRUE_CLIENT_IP", "HTTP_FASTLY_CLIENT_IP", + "HTTP_FLY_CLIENT_IP", "HTTP_X_APPENGINE_USER_IP", "X-CLIENT-IP", "X-REAL-IP", + "X-CLUSTER-CLIENT-IP", "X_FORWARDED", "FORWARDED_FOR", "CF-CONNECTING-IP", + "TRUE-CLIENT-IP", "FASTLY-CLIENT-IP", "FLY-CLIENT-IP", "FORWARDED", "CLIENT-IP", + ) # fmt: skip + + def test_released_order_is_frozen(self): + # New headers may only be inserted between the 4.0.0 entries and REMOTE_ADDR. + n = len(self.V400_ORDER) + self.assertEqual(DEFAULT_PRECEDENCE[:n], self.V400_ORDER) + self.assertEqual(DEFAULT_PRECEDENCE[-1], "REMOTE_ADDR") + self.assertEqual(len(set(DEFAULT_PRECEDENCE)), len(DEFAULT_PRECEDENCE)) + + def test_new_headers(self): + for header in ( + "HTTP_X_CLIENT_IP", "HTTP_X_AZURE_CLIENTIP", "HTTP_DO_CONNECTING_IP", + "HTTP_X_ENVOY_EXTERNAL_ADDRESS", "x-azure-clientip", "do-connecting-ip", + "x-envoy-external-address", "x-appengine-user-ip", + ): # fmt: skip + with self.subTest(header=header): + ip, _ = IpWare().get_client_ip({header: "203.0.113.80"}) + self.assertEqual(str(ip), "203.0.113.80") + + def test_new_headers_never_outrank_existing(self): + meta = {"HTTP_FLY_CLIENT_IP": "203.0.113.81", "HTTP_DO_CONNECTING_IP": "203.0.113.82"} + ip, _ = IpWare().get_client_ip(meta) + self.assertEqual(str(ip), "203.0.113.81") + + def test_new_headers_outrank_remote_addr(self): + meta = {"REMOTE_ADDR": "203.0.113.83", "HTTP_X_AZURE_CLIENTIP": "203.0.113.84"} + ip, _ = IpWare().get_client_ip(meta) + self.assertEqual(str(ip), "203.0.113.84") + + +class TestModernDominatesLegacy(unittest.TestCase): + """Differential check over every combination of direction, proxy_count, + proxy_list and strict: modern never returns a worse address than legacy. + + * Whenever legacy finds a genuinely global address, modern returns the + same address with the same trusted flag. + * Otherwise modern's address ranks at least as high as legacy's, and on + an equal rank it is the same address. + * trusted is True exactly when legacy said so, or when a proxy config is + set and modern resolved a (non-global) address through it. + """ CHAINS = ( "177.139.233.139", @@ -188,8 +327,16 @@ class TestModernMatchesLegacy(unittest.TestCase): ["177.139.233.139"], ) - def test_engines_agree(self): - checked = 0 + @staticmethod + def _rank(ip): + # None ranks 0; a rejected address (0.0.0.0, multicast) ranks below None. + if ip is None: + return 0 + tier = ip_tier(ip) + return -1 if tier == TIER_REJECT else tier + + def test_modern_never_worse(self): + checked = improved = 0 for raw in self.CHAINS: for header in ("HTTP_X_FORWARDED_FOR", "REMOTE_ADDR"): meta = {header: raw} @@ -197,16 +344,23 @@ def test_engines_agree(self): for count in self.PROXY_COUNTS: for plist in self.PROXY_LISTS: kw = {"leftmost": leftmost, "proxy_count": count, "proxy_list": plist} + configured = count is not None or bool(plist) legacy = IpWare(algorithm="legacy", **kw) modern = IpWare(algorithm="modern", **kw) for strict in (False, True): + l_ip, l_trusted = legacy.get_client_ip(meta, strict) + m_ip, m_trusted = modern.get_client_ip(meta, strict) with self.subTest(raw=raw, header=header, strict=strict, **kw): - self.assertEqual( - modern.get_client_ip(meta, strict), - legacy.get_client_ip(meta, strict), - ) + if self._rank(l_ip) == TIER_GLOBAL: + self.assertEqual((m_ip, m_trusted), (l_ip, l_trusted)) + self.assertGreaterEqual(self._rank(m_ip), self._rank(l_ip)) + if self._rank(m_ip) == self._rank(l_ip): + self.assertEqual(m_ip, l_ip) + self.assertEqual(m_trusted, l_trusted or (configured and m_ip is not None)) checked += 1 + improved += (m_ip, m_trusted) != (l_ip, l_trusted) self.assertGreater(checked, 3000) + self.assertGreater(improved, 0) # the matrix really exercises the new ranking if __name__ == "__main__":