Add CIDR optionality to ProxyHeadersMiddleware

tests/middleware/test_proxy_headers.py

@@ -43,8 +43,13 @@ async def app(
         # trusted proxy list
         (["127.0.0.1", "10.0.0.1"], "Remote: https://1.2.3.4:0"),
         ("127.0.0.1, 10.0.0.1", "Remote: https://1.2.3.4:0"),
+        # trusted proxy list with CIDR
+        (["127.0.0.1", "10.0.0.1", "10.0.1.0/24"], "Remote: https://1.2.3.4:0"),
+        ("127.0.0.1, 10.0.0.1, 10.0.1.0/24", "Remote: https://1.2.3.4:0"),
         # request from untrusted proxy
         ("192.168.0.1", "Remote: http://127.0.0.1:123"),
+        # request from untrusted proxy with CIDR
+        ("192.168.0.0/24", "Remote: http://127.0.0.1:123"),
     ],
 )
 async def test_proxy_headers_trusted_hosts(trusted_hosts: list[str] | str, response_text: str) -> None:
@@ -75,6 +80,11 @@ async def test_proxy_headers_trusted_hosts(trusted_hosts: list[str] | str, respo
         ),
         # should set first untrusted as remote address
         (["192.168.0.2", "127.0.0.1"], "Remote: https://10.0.2.1:0"),
+        # works with CIDRs
+        (
+            ["127.0.0.1", "10.0.2.0/24", "192.168.0.2"],
+            "Remote: https://1.2.3.4:0",
+        ),
     ],
 )
 async def test_proxy_headers_multiple_proxies(trusted_hosts: list[str] | str, response_text: str) -> None:

uvicorn/middleware/proxy_headers.py

@@ -10,6 +10,7 @@
 """
 from __future__ import annotations
 
+import ipaddress
 from typing import Union, cast
 
 from uvicorn._types import ASGI3Application, ASGIReceiveCallable, ASGISendCallable, HTTPScope, Scope, WebSocketScope
@@ -23,28 +24,37 @@ def __init__(
     ) -> None:
         self.app = app
         if isinstance(trusted_hosts, str):
-            self.trusted_hosts = {item.strip() for item in trusted_hosts.split(",")}
+            trusted_hosts_set = {item.strip() for item in trusted_hosts.split(",")}
         else:
-            self.trusted_hosts = set(trusted_hosts)
-        self.always_trust = "*" in self.trusted_hosts
+            trusted_hosts_set = set(trusted_hosts)
+        self.always_trust = "*" in trusted_hosts_set
+        trusted_hosts_set.discard("*")
 
-    def get_trusted_client_host(self, x_forwarded_for_hosts: list[str]) -> str | None:
+        self.trusted_hosts = {ipaddress.ip_network(host) for host in trusted_hosts_set}
+
+    def get_trusted_client_host(self, x_forwarded_for_hosts: list[str]) -> str:
         if self.always_trust:
             return x_forwarded_for_hosts[0]
 
         for host in reversed(x_forwarded_for_hosts):
-            if host not in self.trusted_hosts:
+            if not self.check_trusted_host(host):
                 return host
 
-        return None
+        return ""
+
+    def check_trusted_host(self, host: str) -> bool:
+        for trusted_net in self.trusted_hosts:
+            if ipaddress.ip_address(host) in trusted_net:
+                return True
+        return False
 
     async def __call__(self, scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None:
         if scope["type"] in ("http", "websocket"):
             scope = cast(Union["HTTPScope", "WebSocketScope"], scope)
             client_addr: tuple[str, int] | None = scope.get("client")
             client_host = client_addr[0] if client_addr else None
 
-            if self.always_trust or client_host in self.trusted_hosts:
+            if self.always_trust or self.check_trusted_host(client_host):
                 headers = dict(scope["headers"])
 
                 if b"x-forwarded-proto" in headers:
@@ -64,6 +74,6 @@ async def __call__(self, scope: Scope, receive: ASGIReceiveCallable, send: ASGIS
                     x_forwarded_for_hosts = [item.strip() for item in x_forwarded_for.split(",")]
                     host = self.get_trusted_client_host(x_forwarded_for_hosts)
                     port = 0
-                    scope["client"] = (host, port)  # type: ignore[arg-type]
+                    scope["client"] = (host, port)
 
         return await self.app(scope, receive, send)