feat(session): accept CIDR ranges in trusted_proxies

CHANGELOG.md

@@ -31,6 +31,12 @@ Note that 0.3.3 was never released; 0.3.4 follows 0.3.2.
   honouring `trusted_proxies`. Pass it to `create_session()`: behind a trusted
   proxy, `request.client.host` is the proxy's address, so a session bound to it
   was rejected on its next request. ([#87](https://github.com/allen0099/FastAPI-CacheX/issues/87))
+- `SessionConfig.trusted_proxies` accepts CIDR ranges (`10.0.0.0/8`,
+  `2001:db8::/32`) as well as single addresses, for load balancers that connect
+  from a subnet. It applies to both the peer check and the `X-Forwarded-For`
+  walk. IPv4-mapped IPv6 peers match IPv4 entries, and non-IP entries such as
+  `testclient` still match exactly. An entry containing `/` that is not a valid
+  range now fails config validation. ([#73](https://github.com/allen0099/FastAPI-CacheX/issues/73))
 
 ### Fixed
 

docs/SESSION.md

@@ -533,14 +533,27 @@ config = SessionConfig(
 )
 ```
 
+Entries can be single addresses or CIDR ranges, for load balancers that connect from a subnet:
+
+```python
+config = SessionConfig(
+    secret_key="...",
+    ip_binding=True,
+    trusted_proxies=["10.0.0.0/8", "2001:db8::/32"],
+)
+```
+
 The client address is then the **rightmost `X-Forwarded-For` entry that is not listed in
 `trusted_proxies`**: proxies append to the header, so the leftmost entry is whatever the caller
 chose to send and cannot be trusted. If every entry in the chain is a trusted proxy, the direct
 peer address is used. `X-Real-IP` is written by the proxy itself and has no chain to walk, so it is
 used only when `X-Forwarded-For` yields no usable value.
 
 > [!NOTE]
-> `trusted_proxies` currently uses **exact string matching**; CIDR ranges are not supported.
+> An IPv4 peer reported in IPv4-mapped form (`::ffff:10.0.0.8`, as dual-stack sockets do) matches
+> IPv4 entries. Entries that are not IP addresses (for example TestClient's `testclient`) match
+> only an identical peer string, and an entry containing `/` that is not a valid CIDR range is
+> rejected when the config is created.
 
 The middleware applies this logic when it checks a binding, but `create_session()` binds whatever
 `ip_address` you pass it. Behind a trusted proxy, `request.client.host` is the proxy's address,

fastapi_cachex/session/config.py

@@ -1,5 +1,7 @@
 """Session configuration settings."""
 
+import ipaddress
+from functools import lru_cache
 from typing import Literal
 
 from pydantic import BaseModel
@@ -10,6 +12,32 @@
 
 SameSitePolicy = Literal["lax", "strict", "none"]
 
+IPNetwork = ipaddress.IPv4Network | ipaddress.IPv6Network
+IPAddress = ipaddress.IPv4Address | ipaddress.IPv6Address
+
+
+@lru_cache(maxsize=256)
+def _parse_network(entry: str) -> IPNetwork | None:
+    """Parse a trusted-proxy entry as an IP network, or None if it is not one."""
+    try:
+        return ipaddress.ip_network(entry, strict=False)
+    except ValueError:
+        return None
+
+
+@lru_cache(maxsize=1024)
+def _parse_address(value: str) -> IPAddress | None:
+    """Parse a peer or forwarded address as an IP address, or None if it is not one."""
+    try:
+        address = ipaddress.ip_address(value)
+    except ValueError:
+        return None
+    # A dual-stack socket reports IPv4 peers as ::ffff:a.b.c.d.
+    if isinstance(address, ipaddress.IPv6Address) and address.ipv4_mapped:
+        return address.ipv4_mapped
+    return address
+
+
 # Signing algorithms accepted for JWT session tokens. `none` is deliberately
 # absent: an unsigned token would make every session forgeable.
 JWT_ALGORITHMS = frozenset(
@@ -117,8 +145,8 @@ class SessionConfig(BaseModel):
         "the direct peer address, since anyone can send them. When the peer is "
         "trusted, the client address is the rightmost X-Forwarded-For entry "
         "that is not itself listed here: proxies append, so the leftmost entry "
-        "is whatever the caller chose to send. Matching is by exact string; "
-        "CIDR ranges are not supported.",
+        "is whatever the caller chose to send. Entries may be IP addresses, "
+        "CIDR ranges (e.g. 10.0.0.0/8) or other strings, which match exactly.",
     )
     user_agent_binding: bool = Field(
         default=False,
@@ -161,6 +189,41 @@ class SessionConfig(BaseModel):
         "Domain attribute",
     )
 
+    @field_validator("trusted_proxies")
+    @classmethod
+    def _check_trusted_proxies(cls, value: list[str]) -> list[str]:
+        """Reject entries written as a CIDR range that do not parse as one."""
+        for entry in value:
+            if "/" in entry and _parse_network(entry) is None:
+                msg = f"trusted_proxies entry is not a valid CIDR range: {entry!r}"
+                raise ValueError(msg)
+        return value
+
+    def is_trusted_proxy(self, address: str) -> bool:
+        """Report whether `address` matches an entry of `trusted_proxies`.
+
+        An IP address matches an entry that is the same address or a CIDR range
+        containing it (IPv4-mapped IPv6 addresses count as their IPv4 form).
+        Anything else, such as TestClient's ``testclient`` peer, matches only
+        an identical entry.
+
+        Args:
+            address: Peer or forwarded address to check
+
+        Returns:
+            True if the address is a trusted proxy
+        """
+        if address in self.trusted_proxies:
+            return True
+        parsed = _parse_address(address)
+        if parsed is None:
+            return False
+        for entry in self.trusted_proxies:
+            network = _parse_network(entry)
+            if network is not None and parsed in network:
+                return True
+        return False
+
     @field_validator("jwt_algorithm")
     @classmethod
     def _check_jwt_algorithm(cls, value: str) -> str:

fastapi_cachex/session/middleware.py

@@ -59,12 +59,12 @@ def get_client_ip(connection: HTTPConnection, config: SessionConfig) -> str | No
     """
     peer = connection.client.host if connection.client else None
 
-    if peer is not None and peer in config.trusted_proxies:
+    if peer is not None and config.is_trusted_proxy(peer):
         forwarded_for = connection.headers.get("x-forwarded-for")
         if forwarded_for:
             for entry in reversed(forwarded_for.split(",")):
                 candidate = entry.strip()
-                if candidate and candidate not in config.trusted_proxies:
+                if candidate and not config.is_trusted_proxy(candidate):
                     logger.debug("Client IP from X-Forwarded-For: %s", candidate)
                     return candidate
 

tests/session/test_client_ip.py

@@ -4,6 +4,7 @@
 from fastapi import FastAPI
 from fastapi import Request
 from fastapi.testclient import TestClient
+from pydantic import ValidationError
 from starlette.requests import HTTPConnection
 
 from fastapi_cachex.backends.memory import MemoryBackend
@@ -106,3 +107,62 @@ def test_peer_address_binding_fails_behind_trusted_proxy(proxied_app: FastAPI):
 
     me = client.get("/me", headers={**forwarded, "X-Session-Token": token})
     assert me.json() == {"authenticated": False}
+
+
+@pytest.mark.parametrize(
+    ("entry", "peer"),
+    [
+        ("10.0.0.0/8", "10.20.30.40"),
+        ("10.0.0.8", "10.0.0.8"),
+        ("10.0.0.8/24", "10.0.0.200"),  # host bits are ignored
+        ("2001:db8::/32", "2001:db8:1::5"),
+        ("2001:db8::1", "2001:DB8::1"),
+        ("10.0.0.0/8", "::ffff:10.1.2.3"),  # dual-stack socket
+        ("testclient", "testclient"),
+    ],
+)
+def test_trusted_proxy_matches(entry: str, peer: str):
+    config = SessionConfig(secret_key="a" * 32, trusted_proxies=[entry])
+
+    assert config.is_trusted_proxy(peer)
+
+
+@pytest.mark.parametrize(
+    ("entry", "peer"),
+    [
+        ("10.0.0.0/8", "11.0.0.1"),
+        ("10.0.0.8", "10.0.0.9"),
+        ("2001:db8::/32", "2001:db9::1"),
+        ("10.0.0.0/8", "2001:db8::1"),
+        ("10.0.0.0/8", "testclient"),
+        ("testclient", "10.0.0.1"),
+    ],
+)
+def test_trusted_proxy_rejects(entry: str, peer: str):
+    config = SessionConfig(secret_key="a" * 32, trusted_proxies=[entry])
+
+    assert not config.is_trusted_proxy(peer)
+
+
+@pytest.mark.parametrize("entry", ["10.0.0.0/33", "10.0.0.0/x", "proxy/8"])
+def test_malformed_cidr_entry_is_rejected(entry: str):
+    with pytest.raises(ValidationError, match="not a valid CIDR range"):
+        SessionConfig(secret_key="a" * 32, trusted_proxies=[entry])
+
+
+def test_forwarded_chain_skips_every_hop_inside_a_trusted_range():
+    config = SessionConfig(secret_key="a" * 32, trusted_proxies=["10.0.0.0/8"])
+    connection = _connection(
+        "10.0.0.9", {"X-Forwarded-For": "198.51.100.66, 203.0.113.5, 10.1.1.1"}
+    )
+
+    assert get_client_ip(connection, config) == "203.0.113.5"
+
+
+def test_model_copy_update_uses_the_new_ranges():
+    """model_copy(update=...) skips validation; matching must not be stale."""
+    config = SessionConfig(secret_key="a" * 32, trusted_proxies=["10.0.0.0/8"])
+    copied = config.model_copy(update={"trusted_proxies": ["192.168.0.0/16"]})
+
+    assert copied.is_trusted_proxy("192.168.1.1")
+    assert not copied.is_trusted_proxy("10.0.0.1")