From 054bf0eb31526d20188b76ee305edb176ea2f660 Mon Sep 17 00:00:00 2001 From: Val Neekman Date: Tue, 22 Sep 2026 16:24:38 -0400 Subject: [PATCH] Deep-review fixes: exact/boundary proxy matching, deterministic header folding, config validation - proxy_list: complete IPs match exactly (v3 prefix-matched, so 1.2.3.4 trusted 1.2.3.45); prefixes match on octet/group boundaries; IPv6 entries case/zero-insensitive; mapped/NAT64 CIDRs match unwrapped IPv4 hops. - Folded header keys: dash spelling wins regardless of dict order, so a client underscore header cannot shadow the proxy's; conflicts -> absent. - Constructor rejects bare-string proxy_list/precedence, non-IP entries, bool/non-int proxy_count; copies caller lists. Non-mapping meta -> TypeError. - RFC 7239 Forwarded parsing is quote-aware (no smuggled hops). - Tiers: 0.0.0.0/8 rejected; fec0::/10 and 64:ff9b:1::/48 private. - Remove dead is_valid_ip. - Tests: new tests_modern_config.py, wider matrix configs (IPv6, two-hop, count/list mismatch); 100% line+branch coverage of modern engine. --- CHANGELOG.md | 21 ++- README.md | 14 ++ python_ipware/modern/engine.py | 171 ++++++++++++++------ python_ipware/modern/parsers.py | 88 ++++++++-- tests/tests_modern_config.py | 274 ++++++++++++++++++++++++++++++++ tests/tests_modern_matrix.py | 23 ++- 6 files changed, 518 insertions(+), 73 deletions(-) create mode 100644 tests/tests_modern_config.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 54218b0..630af22 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -28,7 +28,9 @@ Enhance (modern engine only; legacy is unchanged): 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. + work. Exact keys still take priority. When several spellings fold to the same header, the dash spelling + wins, whatever the dict order, so a client-sent `x_forwarded_for` cannot shadow the proxy's + `x-forwarded-for`. Dash spellings that disagree are treated as absent. Harden (modern engine only): - Reject malformed tokens instead of truncating them: unclosed brackets (`[::1`), text after a bracket @@ -37,6 +39,23 @@ Harden (modern engine only): - 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. +- Trusted-proxy matching is tighter. A complete IP entry is matched exactly: v3 prefix-matched it, so + `"1.2.3.4"` also trusted `1.2.3.45`, letting that host forge the client IP. Prefixes match on whole + octets or groups (`"10.1"` matches `10.1.x.x`, not `10.100.x.x`). IPv6 entries are case- and + zero-insensitive. An entry ending in `:` stays a prefix, so `"2001:db8::"` behaves as in 4.0.0. + IPv4-mapped and NAT64 CIDR entries (`::ffff:10.0.0.0/104`) match the unwrapped IPv4 hops. +- Configuration errors raise `ValueError` at construction instead of misbehaving silently: `proxy_list` + or `precedence` passed as a bare string (each character became an entry), a non-IP `proxy_list` entry + such as `"foo"`, or a `proxy_count` that is a bool, a float, or a string. `precedence` and `proxy_list` + are copied, so later changes to the caller's lists have no effect. +- A `meta` that is not a mapping raises `TypeError` with a clear message. Non-string keys are skipped. +- RFC 7239 `Forwarded` parsing is quote-aware: a `,` or `;` inside a quoted value no longer splits a hop, + so `ext="x,8.8.8.8"` cannot smuggle in a fake address or a fake proxy hop. +- More address classes: `0.0.0.0/8` is never returned. Deprecated site-local `fec0::/10` ranks as + private, since Python reports it as global. RFC 8215 local-use NAT64 `64:ff9b:1::/48` ranks as private, + since Python reports it as reserved. +- The unused `is_valid_ip` helper was removed from `python_ipware.modern.parsers`. It was never exported. +- Test suite: 100% line and branch coverage of the modern engine. CI: - Bump `actions/upload-artifact` to v7 and `actions/download-artifact` to v8, which run on Node 24. diff --git a/README.md b/README.md index 6ff2731..dcb7479 100644 --- a/README.md +++ b/README.md @@ -214,6 +214,20 @@ ipw = IpWare(precedence=("HTTP_CF_CONNECTING_IP", "HTTP_X_FORWARDED_FOR", "REMOT If your server sits behind known proxies, pass their IPs or prefixes: +Each entry can be (modern engine): + +- **a complete IP**, matched exactly: `"198.84.193.157"` never matches `198.84.193.15x`, and IPv6 + spelling (case, leading zeros) does not matter; +- **a CIDR network** (IPv4 or IPv6), matched by membership. This is the recommended form for IPv6; +- **an IP prefix**, matched on whole octets or groups: `"10.1"` and `"10.1."` match `10.1.x.x` but not + `10.100.x.x`. IPv6 prefixes compare against the compressed form (`2001:db8::5`), so prefer CIDR. + +IPv4-mapped (`::ffff:a.b.c.d`) and NAT64 (`64:ff9b::a.b.c.d`) hops are unwrapped to IPv4 before matching. +Misconfiguration raises `ValueError` at construction: a bare string instead of a list, empty or +non-IP entries, an invalid CIDR, or a negative or non-integer `proxy_count`. `proxy_count` and +`proxy_list` may both be set: the list fixes the client position, and the count is a hop-count +requirement (a minimum, or exact when `strict=True`). + ```python ipw = IpWare(proxy_list=["198.84.193.157"]) # one proxy ipw = IpWare(proxy_list=["198.84.193.157", "198.84.193.158"]) # two proxies diff --git a/python_ipware/modern/engine.py b/python_ipware/modern/engine.py index ae96b7d..7556caa 100644 --- a/python_ipware/modern/engine.py +++ b/python_ipware/modern/engine.py @@ -12,42 +12,108 @@ 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."``). + entry is a CIDR network (``"100.64.0.0/10"``, ``"fd7a:115c:a1e0::/48"``), a + complete IP matched exactly, or an IP prefix matched on octet / group + boundaries (``"10.1."``, ``"10.1"`` -> 10.1.x.x only). * ``trusted_route`` is True whenever the returned IP came from a chain that passed the configured proxy validation, whatever its tier. +* Misconfiguration fails loudly at construction instead of silently trusting + or never matching. """ import ipaddress -from typing import Optional, Union +from collections.abc import Mapping +from typing import Any, Optional, Union from .defaults import DEFAULT_PRECEDENCE -from .parsers import TIER_GLOBAL, TIER_REJECT, IpAddressType, ip_tier, split_proxy_chain +from .parsers import ( + TIER_GLOBAL, + TIER_REJECT, + IpAddressType, + IpNetworkType, + ip_tier, + split_proxy_chain, + unwrap_ipv4, + unwrap_ipv4_network, +) OptionalIp = Optional[IpAddressType] -IpNetworkType = Union[ipaddress.IPv4Network, ipaddress.IPv6Network] ProxyMatcher = Union[str, IpNetworkType] +_PREFIX_CHARS = frozenset("0123456789abcdef.:") + def _compile_proxy_matcher(pattern: str) -> ProxyMatcher: - """CIDR entries become networks; anything else stays a string prefix.""" - if "/" not in pattern: - return pattern - try: - # strict=False accepts host bits set, e.g. "10.0.0.5/24" -> 10.0.0.0/24. - return ipaddress.ip_network(pattern.strip(), strict=False) - except ValueError as exc: - msg = f"Invalid CIDR in proxy_list: {pattern!r}" - raise ValueError(msg) from exc + """Compile one ``proxy_list`` entry. + + * ``"10.0.0.0/8"`` -> network, matched by membership. + * ``"198.84.193.157"`` (a complete IP) -> exact /32 or /128 network, so it + can never match ``198.84.193.15x`` and IPv6 spelling (case, zeros) does + not matter. + * ``"198.84."`` / ``"10.1"`` / ``"2001:db8:"`` -> text prefix, matched on + octet / group boundaries only. An entry ending in ``:`` is always a + prefix, since ``"2001:db8::"`` historically meant "anything under it". + """ + if "/" in pattern: + try: + # strict=False accepts host bits set, e.g. "10.0.0.5/24" -> 10.0.0.0/24. + return unwrap_ipv4_network(ipaddress.ip_network(pattern, strict=False)) + except ValueError as exc: + msg = f"Invalid CIDR in proxy_list: {pattern!r}" + raise ValueError(msg) from exc + if not pattern.endswith(":"): + try: + return ipaddress.ip_network(unwrap_ipv4(ipaddress.ip_address(pattern))) + except ValueError: + pass + prefix = pattern.lower() + if not set(prefix) <= _PREFIX_CHARS: + msg = f"proxy_list entry is not an IP, CIDR or IP prefix: {pattern!r}" + raise ValueError(msg) + return prefix def _proxy_matches(ip: IpAddressType, matcher: ProxyMatcher) -> bool: if isinstance(matcher, str): - return str(ip).startswith(matcher) + text = str(ip) + if not text.startswith(matcher): + return False + # Boundary check: "10.1" matches 10.1.x.x but not 10.100.x.x. + rest = text[len(matcher) :] + return not rest or matcher[-1] in ".:" or rest[0] in ".:%" # Membership across IP versions is simply False, never an error. return ip.version == matcher.version and ip in matcher +def _fold(key: str) -> str: + return key.upper().replace("-", "_") + + +def _build_folded(meta: Mapping[Any, Any]) -> dict[str, object]: + """Case- and dash-insensitive view of ``meta`` for lowercase-key adapters. + + When several spellings fold to the same name, the result must not depend + on dict order. Real header names use dashes, while an underscore spelling + in a raw header dict can only come from the client, so dash spellings win. + If several dash spellings disagree, the header is treated as absent. + """ + groups: dict[str, list[tuple[str, object]]] = {} + for key, value in meta.items(): + if isinstance(key, str): + groups.setdefault(_fold(key), []).append((key, value)) + folded: dict[str, object] = {} + for name, items in groups.items(): + dashed = [v for k, v in items if "-" in k] + pool = dashed or [v for _, v in items] + distinct: list[object] = [] + for value in pool: + if value not in distinct: + distinct.append(value) + if len(distinct) == 1: + folded[name] = distinct[0] + return folded + + class ModernIpWare: def __init__( self, @@ -56,18 +122,32 @@ def __init__( proxy_count: Optional[int] = None, proxy_list: Optional[list[str]] = None, ) -> None: - if proxy_count is not None and proxy_count < 0: - 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.") + if proxy_count is not None and ( + isinstance(proxy_count, bool) or not isinstance(proxy_count, int) or proxy_count < 0 + ): + raise ValueError("proxy_count must be a non-negative integer") + # A bare string is iterable and would silently become one prefix per + # character ("10.0.0.1" -> "1", "0", ...), trusting almost anything. + if isinstance(proxy_list, str) or ( + proxy_list is not None and not all(isinstance(p, str) for p in proxy_list) + ): + raise ValueError("proxy_list must be a list of 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.") + # proxy_count and proxy_list may both be set and may differ (v3 API): + # the count is a hop-count requirement (minimum, or exact when strict) + # and the list pins the client position. See _best_from_chain. + if isinstance(precedence, str) or ( + precedence is not None and not all(isinstance(h, str) for h in precedence) + ): + raise ValueError("precedence must be a sequence of header-name strings.") - self.precedence = precedence or DEFAULT_PRECEDENCE + # Copy, so later changes to the caller's objects cannot leak in. + self.precedence = tuple(precedence) if precedence else DEFAULT_PRECEDENCE self.leftmost = leftmost self.proxy_count = proxy_count self.proxy_list = proxy_list @@ -75,39 +155,26 @@ def __init__( # -- meta access -------------------------------------------------------- - @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 {} - 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) + def _get_meta_values(self, meta: Optional[Mapping[Any, Any]]) -> list[str]: + if meta is None: + return [] + if not isinstance(meta, Mapping): + msg = f"meta must be a mapping of header names to values, got {type(meta).__name__}" + raise TypeError(msg) + folded = _build_folded(meta) values: list[str] = [] for key in self.precedence: - value = self._get_meta_value(meta, key, folded) - if value: - values.append(value) + 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: + value = folded.get(_fold(key)) + # Header values are text; anything else (None, bytes, lists from + # a misbehaving adapter) is ignored rather than crashing. + if isinstance(value, str) and value.strip(): + values.append(value.strip()) return values # -- validation --------------------------------------------------------- @@ -157,7 +224,7 @@ def _best_from_chain(self, chain: list[IpAddressType]) -> tuple[OptionalIp, bool # -- public API --------------------------------------------------------- - def get_client_ip(self, meta: dict[str, str], strict: bool = False) -> tuple[OptionalIp, bool]: + def get_client_ip(self, meta: Optional[Mapping[Any, Any]], 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 diff --git a/python_ipware/modern/parsers.py b/python_ipware/modern/parsers.py index c7f8b87..52d9695 100644 --- a/python_ipware/modern/parsers.py +++ b/python_ipware/modern/parsers.py @@ -9,9 +9,17 @@ from typing import Optional, Union IpAddressType = Union[ipaddress.IPv4Address, ipaddress.IPv6Address] +IpNetworkType = Union[ipaddress.IPv4Network, ipaddress.IPv6Network] # RFC 6052 well-known NAT64 prefix; the low 32 bits are the IPv4 client. _NAT64_WKP = ipaddress.IPv6Network("64:ff9b::/96") +# RFC 8215 local-use NAT64 prefix. The IPv4 position depends on the operator's +# chosen prefix length, so it is not unwrapped; it is ranked as private. +_NAT64_LOCAL = ipaddress.IPv6Network("64:ff9b:1::/48") +# IPv4-mapped IPv6 space. Hops in it are unwrapped to IPv4. +_IPV4_MAPPED = ipaddress.IPv6Network("::ffff:0:0/96") +# RFC 1122 "this network": never a valid source address. +_THIS_NETWORK = ipaddress.IPv4Network("0.0.0.0/8") # How good an address is as a client IP; higher wins. REJECT is never returned. TIER_REJECT = 0 @@ -24,12 +32,13 @@ 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. NAT64 well-known-prefix addresses - never reach here: ``parse_ip`` unwraps them to IPv4 first. + Check order matters: Python reports multicast (``224.0.0.1``, ``ff02::1``), + site-local ``fec0::/10`` and the deprecated ``::a.b.c.d`` form as + ``is_global``, and ``::1`` and ``64:ff9b:1::/48`` as ``is_reserved``, so + those are resolved before ``is_reserved`` / ``is_global`` are trusted. + Unspecified, ``0.0.0.0/8``, multicast, broadcast and reserved addresses can + never be a real client and are rejected outright. NAT64 well-known-prefix + addresses never reach here: ``parse_ip`` unwraps them to IPv4 first. Global/private classification comes from the running Python's ``ipaddress`` tables, which changed in 3.12 (e.g. 6to4 ``2002::/16`` is @@ -37,10 +46,16 @@ def ip_tier(ip: IpAddressType) -> int: """ if ip.is_unspecified or ip.is_multicast: return TIER_REJECT + if isinstance(ip, ipaddress.IPv4Address) and ip in _THIS_NETWORK: + return TIER_REJECT if ip.is_loopback: return TIER_LOOPBACK + if isinstance(ip, ipaddress.IPv6Address) and ip in _NAT64_LOCAL: + return TIER_PRIVATE # Python calls it reserved (it sits in ::/8) if ip.is_reserved: # 240.0.0.0/4 incl. 255.255.255.255; unallocated IPv6 return TIER_REJECT + if isinstance(ip, ipaddress.IPv6Address) and ip.is_site_local: + return TIER_PRIVATE # deprecated fec0::/10, which Python calls global if ip.is_global: return TIER_GLOBAL if ip.is_link_local: @@ -80,14 +95,44 @@ def strip_port(value: str) -> str: return value # bare IPv6 or bare IPv4 +def split_unquoted(value: str, sep: str) -> list[str]: + """Split ``value`` on ``sep``, ignoring separators inside double quotes. + + RFC 7239 values may be quoted strings (``by="a,b"``), and a separator + inside one must not start a new hop or parameter. Backslash escapes inside + quotes are honoured. An unclosed quote swallows the rest of the value, so + that token fails to parse instead of leaking a fragment as a fake hop. + """ + if '"' not in value: + return value.split(sep) + parts: list[str] = [] + buf: list[str] = [] + quoted = escaped = False + for ch in value: + if escaped: + escaped = False + elif quoted and ch == "\\": + escaped = True + elif ch == '"': + quoted = not quoted + elif ch == sep and not quoted: + parts.append("".join(buf)) + buf = [] + continue + buf.append(ch) + parts.append("".join(buf)) + return parts + + 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. + names are case-insensitive, and ``;`` inside quoted values is not a + separator. Returns ``""`` when the element has no ``for`` parameter, so the + hop counts as invalid instead of being guessed at. """ - for pair in element.split(";"): + for pair in split_unquoted(element, ";"): key, sep, val = pair.partition("=") if sep and key.strip().lower() == "for": return val.strip() @@ -113,9 +158,13 @@ def parse_ip(value: Optional[str]) -> Optional[IpAddressType]: if not cleaned: return None try: - ip = ipaddress.ip_address(cleaned) + return unwrap_ipv4(ipaddress.ip_address(cleaned)) except ValueError: return None + + +def unwrap_ipv4(ip: IpAddressType) -> IpAddressType: + """Return the IPv4 address embedded in an IPv4-mapped or NAT64-WKP address.""" if isinstance(ip, ipaddress.IPv6Address): if ip.ipv4_mapped is not None: return ip.ipv4_mapped @@ -124,6 +173,19 @@ def parse_ip(value: Optional[str]) -> Optional[IpAddressType]: return ip +def unwrap_ipv4_network(net: IpNetworkType) -> IpNetworkType: + """IPv4 network embedded in a /96-or-longer IPv4-mapped or NAT64-WKP network. + + Hops in those ranges are unwrapped to IPv4 before matching, so a trusted + proxy written as ``::ffff:10.0.0.0/104`` must match the IPv4 hop too. + """ + if isinstance(net, ipaddress.IPv6Network) and net.prefixlen >= 96: + for embedding in (_IPV4_MAPPED, _NAT64_WKP): + if net.subnet_of(embedding): + return ipaddress.IPv4Network((int(net.network_address) & 0xFFFFFFFF, net.prefixlen - 96)) + return net + + def split_proxy_chain(raw: Optional[str], strict: bool = False) -> Optional[list[IpAddressType]]: """Split a comma-separated proxy chain into ordered ``ip_address`` objects. @@ -137,7 +199,7 @@ def split_proxy_chain(raw: Optional[str], strict: bool = False) -> Optional[list if not raw: return [] result: list[IpAddressType] = [] - for token in raw.split(","): + for token in split_unquoted(raw, ","): # No plain IP token contains "=", only RFC 7239 Forwarded elements do. candidate = forwarded_for(token) if "=" in token else token ip = parse_ip(candidate) @@ -146,7 +208,3 @@ def split_proxy_chain(raw: Optional[str], strict: bool = False) -> Optional[list elif strict: return None return result - - -def is_valid_ip(value: Optional[str]) -> bool: - return parse_ip(value) is not None diff --git a/tests/tests_modern_config.py b/tests/tests_modern_config.py new file mode 100644 index 0000000..35ed70b --- /dev/null +++ b/tests/tests_modern_config.py @@ -0,0 +1,274 @@ +"""Modern engine: configuration validation, proxy matchers, meta handling, +RFC 7239 quoting and address classes. Each test pins a behavior that the +exhaustive matrices in tests_modern_matrix.py cannot reach.""" + +import ipaddress +import unittest +from typing import ClassVar + +from python_ipware import IpWare +from python_ipware.modern.parsers import TIER_PRIVATE, TIER_REJECT, ip_tier, split_proxy_chain, split_unquoted + +ip = ipaddress.ip_address +CLIENT = ip("8.8.8.8") + + +def trusted_via(proxy_entry, hop, strict=True): + """Resolve ``8.8.8.8, `` with a single trusted proxy entry.""" + meta = {"HTTP_X_FORWARDED_FOR": f"8.8.8.8, {hop}"} + return IpWare(proxy_list=[proxy_entry]).get_client_ip(meta, strict) + + +class TestConstructorValidation(unittest.TestCase): + def test_rejected_configs(self): + cases = [ + {"proxy_count": -1}, + {"proxy_count": True}, + {"proxy_count": 1.5}, + {"proxy_count": "1"}, + {"proxy_list": "10.0.0.1"}, # bare string, would explode per character + {"proxy_list": [1]}, + {"proxy_list": [None]}, + {"proxy_list": ["10.", 5]}, + {"proxy_list": [""]}, + {"proxy_list": [" "]}, + {"proxy_list": ["foo"]}, + {"proxy_list": ["10.0.0.0/33"]}, + {"proxy_list": ["300.1.0.0/16"]}, + {"precedence": "REMOTE_ADDR"}, + {"precedence": ("REMOTE_ADDR", 5)}, + ] + for kwargs in cases: + with self.subTest(kwargs=kwargs), self.assertRaises(ValueError): + IpWare(**kwargs) + + def test_accepted_configs(self): + for kwargs in ( + {"proxy_count": 0}, + {"proxy_list": []}, + {"proxy_list": ("10.0.0.1",)}, + {"proxy_list": [" 10.0.0.0/8 ", "2001:db8::/32", "fd00:", "198.84"]}, + {"proxy_count": 2, "proxy_list": ["198.84."]}, # may differ, see TestCountAndList + {"precedence": ["REMOTE_ADDR"]}, + ): + with self.subTest(kwargs=kwargs): + IpWare(**kwargs) + + def test_caller_objects_are_copied(self): + precedence, proxies = ["REMOTE_ADDR"], ["198.84.193.157"] + ipw = IpWare(precedence=precedence, proxy_list=proxies) + precedence.append("HTTP_X_REAL_IP") + proxies.append("10.") + self.assertEqual(ipw.engine.precedence, ("REMOTE_ADDR",)) + self.assertEqual(ipw.engine.proxy_list, ["198.84.193.157"]) + + def test_empty_precedence_means_defaults(self): + ipw = IpWare(precedence=()) + self.assertEqual(str(ipw.get_client_ip({"HTTP_X_REAL_IP": "8.8.8.8"})[0]), "8.8.8.8") + + def test_repr(self): + self.assertEqual(repr(IpWare()), "IpWare(algorithm='auto' -> 'modern')") + self.assertEqual(repr(IpWare(algorithm="legacy")), "IpWare(algorithm='legacy' -> 'legacy')") + + +class TestProxyMatchers(unittest.TestCase): + def test_full_ip_entry_is_exact(self): + # v3 prefix-matched, so "1.2.3.4" trusted 1.2.3.45 and let it forge the client. + self.assertEqual(trusted_via("1.2.3.4", "1.2.3.45"), (None, False)) + self.assertEqual(trusted_via("1.2.3.4", "1.2.3.4"), (CLIENT, True)) + + def test_ipv4_prefix_respects_octet_boundary(self): + cases = { + ("10.1", "10.1.0.1"): True, + ("10.1", "10.100.0.1"): False, + ("10.1", "10.10.0.1"): False, + ("10.1.", "10.1.0.1"): True, + ("10.1.", "10.10.0.1"): False, + ("198.84.193", "198.84.193.157"): True, + ("198.84.19", "198.84.193.157"): False, + } + for (entry, hop), trusted in cases.items(): + with self.subTest(entry=entry, hop=hop): + self.assertEqual(trusted_via(entry, hop), (CLIENT, True) if trusted else (None, False)) + + def test_ipv6_entry_spelling_does_not_matter(self): + for entry in ("2001:db8::5", "2001:DB8::5", "2001:0db8:0000:0000:0000:0000:0000:0005", "2001:db8::5/128"): + with self.subTest(entry=entry): + self.assertEqual(trusted_via(entry, "2001:db8::5"), (CLIENT, True)) + self.assertEqual(trusted_via("2001:db8::5", "2001:db8::50"), (None, False)) + + def test_ipv6_prefix_respects_group_boundary(self): + cases = { + ("2001:db8:", "2001:db8::5"): True, + ("2001:DB8:", "2001:db8::5"): True, + ("2001:db8::", "2001:db8::5"): True, # trailing "::" stays a prefix, as in 4.0.0 + ("2001:db8", "2001:db8::5"): True, + ("2001:db8", "2001:db80::5"): False, + ("fe80:", "fe80::1%eth0"): True, + } + for (entry, hop), trusted in cases.items(): + with self.subTest(entry=entry, hop=hop): + self.assertEqual(trusted_via(entry, hop), (CLIENT, True) if trusted else (None, False)) + + def test_cidr_matchers(self): + cases = { + ("10.0.0.0/8", "10.9.9.9"): True, + ("10.0.0.5/24", "10.0.0.200"): True, # host bits allowed + ("10.0.0.0/8", "11.0.0.1"): False, + ("2001:db8::/32", "2001:db8:ffff::1"): True, + ("2001:db8::/32", "2001:db9::1"): False, + ("fe80::/10", "fe80::1%eth0"): True, + ("10.0.0.0/8", "2001:db8::1"): False, # cross-version never matches + } + for (entry, hop), trusted in cases.items(): + with self.subTest(entry=entry, hop=hop): + self.assertEqual(trusted_via(entry, hop), (CLIENT, True) if trusted else (None, False)) + + def test_embedded_ipv4_forms_match_as_ipv4(self): + # Hops are unwrapped to IPv4, so entries in either spelling must still match. + cases = { + ("10.0.0.0/8", "::ffff:10.0.0.2"): True, + ("10.0.0.0/8", "64:ff9b::a00:2"): True, + ("::ffff:10.0.0.0/104", "10.0.0.2"): True, + ("64:ff9b::a00:0/104", "10.0.0.2"): True, + ("::ffff:10.0.0.2", "10.0.0.2"): True, + ("64:ff9b::a00:2", "::ffff:10.0.0.2"): True, + ("::ffff:10.0.0.0/104", "11.0.0.2"): False, + } + for (entry, hop), trusted in cases.items(): + with self.subTest(entry=entry, hop=hop): + self.assertEqual(trusted_via(entry, hop), (CLIENT, True) if trusted else (None, False)) + + def test_zone_id_hop_matches_exact_entry(self): + self.assertEqual(trusted_via("fe80::1", "fe80::1%eth0"), (CLIENT, True)) + + +class TestCountAndList(unittest.TestCase): + """Both may be set and may differ (v3 API): the count is a hop-count + requirement (minimum, exact when strict); the list pins the position.""" + + META: ClassVar[dict[str, str]] = {"HTTP_X_FORWARDED_FOR": "6.6.6.6, 8.8.8.8, 9.9.9.9, 198.84.1.1"} + + def test_list_pins_position_count_is_minimum(self): + ipw = IpWare(proxy_count=2, proxy_list=["198.84."]) + self.assertEqual(ipw.get_client_ip(self.META), (ip("9.9.9.9"), True)) + + def test_count_above_hops_rejects(self): + ipw = IpWare(proxy_count=4, proxy_list=["198.84."]) + self.assertEqual(ipw.get_client_ip(self.META), (None, False)) + + def test_strict_needs_both_exact(self): + self.assertEqual(IpWare(proxy_count=3, proxy_list=["198.84."]).get_client_ip(self.META, True), (None, False)) + ipw = IpWare(proxy_count=1, proxy_list=["198.84."]) + meta = {"HTTP_X_FORWARDED_FOR": "8.8.8.8, 198.84.1.1"} + self.assertEqual(ipw.get_client_ip(meta, True), (CLIENT, True)) + + +class TestMetaHandling(unittest.TestCase): + def test_non_mapping_meta_raises(self): + for meta in (["REMOTE_ADDR"], "REMOTE_ADDR", 5): + with self.subTest(meta=meta), self.assertRaises(TypeError): + IpWare().get_client_ip(meta) + + def test_none_and_empty_meta(self): + self.assertEqual(IpWare().get_client_ip(None), (None, False)) + self.assertEqual(IpWare().get_client_ip({}), (None, False)) + + def test_non_string_keys_are_skipped(self): + meta = {1: "x", None: "y", ("a",): "z", "REMOTE_ADDR": "8.8.8.8"} + self.assertEqual(IpWare().get_client_ip(meta), (CLIENT, False)) + + def test_dash_spelling_beats_underscore_in_any_order(self): + pairs = [("x_forwarded_for", "6.6.6.6"), ("x-forwarded-for", "8.8.8.8")] + for order in (pairs, pairs[::-1]): + with self.subTest(order=order): + self.assertEqual(IpWare().get_client_ip(dict(order)), (CLIENT, False)) + + def test_underscore_spelling_cannot_forge_through_trusted_proxy(self): + ipw = IpWare(proxy_list=["10.0.0.0/8"]) + pairs = [("x_forwarded_for", "6.6.6.6, 10.0.0.2"), ("x-forwarded-for", "8.8.8.8, 10.0.0.2")] + for order in (pairs, pairs[::-1]): + with self.subTest(order=order): + self.assertEqual(ipw.get_client_ip(dict(order)), (CLIENT, True)) + + def test_conflicting_dash_spellings_are_ignored(self): + meta = {"X-Forwarded-For": "6.6.6.6", "x-forwarded-for": "8.8.8.8", "REMOTE_ADDR": "9.9.9.9"} + self.assertEqual(IpWare().get_client_ip(meta), (ip("9.9.9.9"), False)) + + def test_agreeing_spellings_resolve(self): + meta = {"X-Forwarded-For": "8.8.8.8", "x-forwarded-for": "8.8.8.8"} + self.assertEqual(IpWare().get_client_ip(meta), (CLIENT, False)) + + def test_unhashable_values_do_not_crash(self): + meta = {"x-forwarded-for": ["8.8.8.8"], "X-Forwarded-For": ["8.8.8.8"], "REMOTE_ADDR": "9.9.9.9"} + self.assertEqual(IpWare().get_client_ip(meta), (ip("9.9.9.9"), False)) + + def test_whitespace_only_value_is_absent(self): + meta = {"HTTP_X_FORWARDED_FOR": " \t\r\n", "REMOTE_ADDR": "8.8.8.8"} + self.assertEqual(IpWare().get_client_ip(meta), (CLIENT, False)) + + +class TestForwardedQuoting(unittest.TestCase): + def test_quoted_separators_do_not_split(self): + cases = { + 'for=10.0.0.1;ext="x,8.8.8.8"': ip("10.0.0.1"), # no smuggled hop + 'for=8.8.8.8;by="a,b"': CLIENT, + 'for=8.8.8.8;ext="a;for=6.6.6.6"': CLIENT, + 'ext="a\\"b,c";for=8.8.8.8': CLIENT, # escaped quote inside quoted-string + } + for raw, expected in cases.items(): + with self.subTest(raw=raw): + self.assertEqual(IpWare().get_client_ip({"HTTP_FORWARDED": raw}, strict=True), (expected, False)) + + def test_quoted_comma_cannot_fake_proxy_hop(self): + meta = {"HTTP_FORWARDED": 'for=8.8.4.4;by="_a,9.9.9.9"'} + self.assertEqual(IpWare(proxy_count=1).get_client_ip(meta), (None, False)) + + def test_unclosed_quote_swallows_the_rest(self): + meta = {"HTTP_FORWARDED": 'for=8.8.8.8, for="6.6.6.6, for=9.9.9.9'} + self.assertEqual(IpWare().get_client_ip(meta), (CLIENT, False)) + self.assertEqual(IpWare().get_client_ip(meta, strict=True), (None, False)) + + def test_split_unquoted_helper(self): + self.assertEqual(split_unquoted("a,b", ","), ["a", "b"]) + self.assertEqual(split_unquoted('a,"b,c",d', ","), ["a", '"b,c"', "d"]) + self.assertEqual(split_unquoted('"a\\",b",c', ","), ['"a\\",b"', "c"]) + self.assertEqual(split_unquoted("", ","), [""]) + + def test_split_proxy_chain_empty(self): + self.assertEqual(split_proxy_chain(""), []) + self.assertEqual(split_proxy_chain(None), []) + + +class TestAddressClasses(unittest.TestCase): + def test_extra_tiers(self): + cases = { + "fec0::1": TIER_PRIVATE, # deprecated site-local; Python calls it global + "64:ff9b:1::808:808": TIER_PRIVATE, # RFC 8215 local-use NAT64; Python calls it reserved + "0.1.2.3": TIER_REJECT, # "this network" + "0.255.255.255": TIER_REJECT, + } + for address, tier in cases.items(): + with self.subTest(address=address): + self.assertEqual(ip_tier(ip(address)), tier) + + def test_site_local_does_not_beat_public(self): + meta = {"HTTP_X_FORWARDED_FOR": "fec0::1, 8.8.8.8"} + self.assertEqual(IpWare().get_client_ip(meta), (CLIENT, False)) + + def test_this_network_never_returned(self): + self.assertEqual(IpWare().get_client_ip({"REMOTE_ADDR": "0.1.2.3"}), (None, False)) + + def test_local_use_nat64_is_a_private_fallback(self): + meta = {"HTTP_X_FORWARDED_FOR": "64:ff9b:1::808:808", "REMOTE_ADDR": "127.0.0.1"} + self.assertEqual(IpWare().get_client_ip(meta), (ip("64:ff9b:1::808:808"), False)) + + +class TestLargeInput(unittest.TestCase): + def test_long_header_is_handled(self): + raw = ", ".join(["10.0.0.1"] * 20000 + ["8.8.8.8"]) + self.assertEqual(IpWare().get_client_ip({"HTTP_X_FORWARDED_FOR": raw}), (CLIENT, False)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/tests_modern_matrix.py b/tests/tests_modern_matrix.py index 3b93951..1b4b9f6 100644 --- a/tests/tests_modern_matrix.py +++ b/tests/tests_modern_matrix.py @@ -57,10 +57,15 @@ (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"]), + (None, ["198.84.193.157"]), # exact IPv4 + (None, ["198.84."]), # IPv4 prefix + (None, ["198.84"]), # IPv4 prefix without trailing dot (octet boundary) + (None, ["10.0.0.0/8"]), # IPv4 CIDR + (None, ["2606:4700::6810:84E5"]), # exact IPv6, non-canonical case + (None, ["fd00::/8"]), # IPv6 CIDR + (None, ["198.84.193.157", "10.0.0.0/8"]), # two trusted hops + (1, ["198.84.193.157"]), # count and list agree + (2, ["198.84."]), # count and list differ: list pins, count is a minimum ) @@ -70,11 +75,19 @@ def _chains(max_len: int = 3): def _proxy_ok(address: str, pattern: str) -> bool: + """Spec for one proxy_list entry, written independently of the engine.""" 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) + try: + return ip == ipaddress.ip_address(pattern) # a complete IP is exact + except ValueError: + pass + # An IPv4 prefix names leading octets; a partial last octet is not allowed. + want = pattern.rstrip(".").split(".") + have = address.split(".") + return ip.version == 4 and have[: len(want)] == want def _oracle(raw_values, leftmost, proxy_count, proxy_list, strict):