Merge remote-tracking branch 'origin/master' into proxy-headers

Co-authored-by: Marcelo Trylesinski <marcelotryle@gmail.com>

docs/deployment.md

@@ -93,10 +93,12 @@ Options:
                                   Enable/Disable default Server header.
   --date-header / --no-date-header
                                   Enable/Disable default Date header.
-  --forwarded-allow-ips TEXT      Comma separated list of IPs to trust with
-                                  proxy headers. Defaults to the
-                                  $FORWARDED_ALLOW_IPS environment variable if
-                                  available, or '127.0.0.1'.
+  --forwarded-allow-ips TEXT      Comma separated list of IP Addresses, IP
+                                  Networks, or literals (e.g. UNIX Socket
+                                  path) to trust with proxy headers. Defaults
+                                  to the $FORWARDED_ALLOW_IPS environment
+                                  variable if available, or '127.0.0.1'. The
+                                  literal '*' means trust everything.
   --root-path TEXT                Set the ASGI 'root_path' for applications
                                   submounted below a given URL path.
   --limit-concurrency INTEGER     Maximum number of concurrent connections or
@@ -279,12 +281,9 @@ Using Nginx as a proxy in front of your Uvicorn processes may not be necessary,
 
 In managed environments such as `Heroku`, you won't typically need to configure Nginx, as your server processes will already be running behind load balancing proxies.
 
-The recommended configuration for proxying from Nginx is to use a UNIX domain socket between Nginx and whatever the process manager that is being used to run Uvicorn.
-Note that when doing this you will need to run Uvicorn with `--forwarded-allow-ips='*'` to ensure that the domain socket is trusted as a source from which to proxy headers.
+The recommended configuration for proxying from Nginx is to use a UNIX domain socket between Nginx and whatever the process manager that is being used to run Uvicorn. If using Uvicorn directly you can bind it to a UNIX domain socket using `uvicorn --uds /path/to/socket.sock <...>`.
 
-When fronting the application with a proxy server you want to make sure that the proxy sets headers to ensure that the application can properly determine the client address of the incoming connection, and if the connection was over `http` or `https`.
-
-You should ensure that the `X-Forwarded-For` and `X-Forwarded-Proto` headers are set by the proxy, and that Uvicorn is run using the `--proxy-headers` setting. This ensures that the ASGI scope includes correct `client` and `scheme` information.
+When running your application behind one or more proxies you will want to make sure that each proxy sets appropriate headers to ensure that your application can properly determine the client address of the incoming connection, and if the connection was over `http` or `https`. For more information see [Proxies and Forwarded Headers][#proxies-and-forwarded-headers] below.
 
 Here's how a simple Nginx configuration might look. This example includes setting proxy headers, and using a UNIX domain socket to communicate with the application server.
 
@@ -362,3 +361,37 @@ $ gunicorn --keyfile=./key.pem --certfile=./cert.pem -k uvicorn.workers.UvicornW
 [nginx_websocket]: https://nginx.org/en/docs/http/websocket.html
 [letsencrypt]: https://letsencrypt.org/
 [mkcert]: https://github.com/FiloSottile/mkcert
+
+## Proxies and Forwarded Headers
+
+When running an application behind one or more proxies, certain information about the request is lost. To avoid this most proxies will add headers containing this information for downstream servers to read.
+
+Uvicorn currently supports the following headers:
+
+- `X-Forwarded-For` ([MDN Reference](https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/X-Forwarded-For))
+- `X-Forwarded-Proto`([MDN Reference](https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/X-Forwarded-Proto))
+
+Uvicorn can use these headers to correctly set the client and protocol in the request. However as anyone can set these headers you must configure which "clients" you will trust to have set them correctly.
+
+Uvicorn can be configured to trust IP Addresses (e.g. `127.0.0.1`), IP Networks (e.g. `10.100.0.0/16`), or Literals (e.g. `/path/to/socket.sock`). When running from CLI these are configured using `--forwarded-trust-ips`.
+
+!!! Warning: Only trust clients you can actually trust
+    Incorrectly trusting other clients can lead to malicious actors spoofing their apparent client address to your application.
+
+For more informations see [`ProxyHeadersMiddleware`](https://github.com/encode/uvicorn/blob/master/uvicorn/middleware/proxy_headers.py)
+### Client Port
+
+Currently if the `ProxyHeadersMiddleware` is able to retrieve a trusted client value then the client's port will be set to `0`. This is because port information is lost when using these headers.
+
+### UNIX Domain Sockets (UDS)
+
+Although it is common for UNIX Domain Sockets to be used for communicating between various HTTP servers, they can mess with some of the expected received values as they will be various non-address strings or missing values.
+
+For example:
+- when NGINX itself is running behind a UDS it will add the literal `unix:` as the client in the `X-Forwarded-For` header.
+- When Uvicorn is running behind a UDS the initial client will be `None`.
+
+
+### Trust Everything
+
+Rather than specifying what to trust, you can instruct Uvicorn to trust all clients using the literal `"*"`. You should only set this when you know you can trust all values within the forwarded headers (e.g. because your proxies remove the existing headers before setting their own).

docs/index.md

@@ -163,10 +163,12 @@ Options:
                                   Enable/Disable default Server header.
   --date-header / --no-date-header
                                   Enable/Disable default Date header.
-  --forwarded-allow-ips TEXT      Comma separated list of IPs to trust with
-                                  proxy headers. Defaults to the
-                                  $FORWARDED_ALLOW_IPS environment variable if
-                                  available, or '127.0.0.1'.
+  --forwarded-allow-ips TEXT      Comma separated list of IP Addresses, IP
+                                  Networks, or literals (e.g. UNIX Socket
+                                  path) to trust with proxy headers. Defaults
+                                  to the $FORWARDED_ALLOW_IPS environment
+                                  variable if available, or '127.0.0.1'. The
+                                  literal '*' means trust everything.
   --root-path TEXT                Set the ASGI 'root_path' for applications
                                   submounted below a given URL path.
   --limit-concurrency INTEGER     Maximum number of concurrent connections or

docs/settings.md

@@ -88,9 +88,9 @@ Note that WSGI mode always disables WebSocket support, as it is not supported by
 ## HTTP
 
 * `--root-path <str>` - Set the ASGI `root_path` for applications submounted below a given URL path.
-* `--proxy-headers` / `--no-proxy-headers` - Enable/Disable X-Forwarded-Proto, X-Forwarded-For, X-Forwarded-Port to populate remote address info. Defaults to enabled, but is restricted to only trusting
+* `--proxy-headers` / `--no-proxy-headers` - Enable/Disable X-Forwarded-Proto, X-Forwarded-For to populate remote address info. Defaults to enabled, but is restricted to only trusting
 connecting IPs in the `forwarded-allow-ips` configuration.
-* `--forwarded-allow-ips` <comma-separated-list> Comma separated list of IPs to trust with proxy headers. Defaults to the `$FORWARDED_ALLOW_IPS` environment variable if available, or '127.0.0.1'. A wildcard '*' means always trust.
+* `--forwarded-allow-ips` <comma-separated-list> Comma separated list of IP Addresses, IP Networks, or literals (e.g. UNIX Socket path) to trust with proxy headers. Defaults to the `$FORWARDED_ALLOW_IPS` environment variable if available, or '127.0.0.1'. The literal `'*'` means trust everything.
 * `--server-header` / `--no-server-header` - Enable/Disable default `Server` header.
 * `--date-header` / `--no-date-header` - Enable/Disable default `Date` header.
 

tests/middleware/test_proxy_headers.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 from typing import TYPE_CHECKING
 
 import httpx
+import httpx._transports.asgi
 import pytest
 import websockets.client
 
@@ -10,7 +11,7 @@ from tests.response import Response
 from tests.utils import run_server
 from uvicorn._types import ASGIReceiveCallable, ASGISendCallable, Scope
 from uvicorn.config import Config
-from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
+from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware, _TrustedHosts
 
 if TYPE_CHECKING:
     from uvicorn.protocols.http.h11_impl import H11Protocol
@@ -19,99 +20,436 @@ if TYPE_CHECKING:
     from uvicorn.protocols.websockets.wsproto_impl import WSProtocol
 
 
-async def app(
+X_FORWARDED_FOR = "X-Forwarded-For"
+X_FORWARDED_PROTO = "X-Forwarded-Proto"
+
+
+async def default_app(
     scope: Scope,
     receive: ASGIReceiveCallable,
     send: ASGISendCallable,
 ) -> None:
     scheme = scope["scheme"]  # type: ignore
-    host, port = scope["client"]  # type: ignore
-    addr = "%s://%s:%d" % (scheme, host, port)
-    response = Response("Remote: " + addr, media_type="text/plain")
+    if (client := scope["client"]) is None:  # type: ignore
+        client_addr = "NONE"
+    else:
+        host, port = client
+        client_addr = f"{host}:{port}"
+
+    response = Response(f"{scheme}://{client_addr}", media_type="text/plain")
     await response(scope, receive, send)
 
 
+def make_httpx_client(
+    trusted_hosts: str | list[str],
+    client: tuple[str, int] = ("127.0.0.1", 123),
+) -> httpx.AsyncClient:
+    """Create async client for use in test cases
+
+    Args:
+        trusted_hosts: trusted_hosts for proxy middleware
+        client: transport client to use
+    """
+
+    app = ProxyHeadersMiddleware(default_app, trusted_hosts)
+    transport = httpx.ASGITransport(app=app, client=client)  # type: ignore
+    return httpx.AsyncClient(transport=transport, base_url="http://testserver")
+
+
+# Note: we vary the format here to also test some of the functionality
+# of the _TrustedHosts.__init__ method.
+_TRUSTED_NOTHING: list[str] = []
+_TRUSTED_EVERYTHING = "*"
+_TRUSTED_IPv4_ADDRESSES = "127.0.0.1, 10.0.0.1"
+_TRUSTED_IPv4_NETWORKS = ["127.0.0.0/8", "10.0.0.0/8"]
+_TRUSTED_IPv6_ADDRESSES = [
+    "2001:db8::",
+    "2001:0db8:0001:0000:0000:0ab9:C0A8:0102",
+    "2001:db8:3333:4444:5555:6666:1.2.3.4",  # This is a dual address
+    "::11.22.33.44",  # This is a dual address
+]
+_TRUSTED_IPv6_NETWORKS = "2001:db8:abcd:0012::0/64"
+_TRUSTED_LITERALS = "some-literal , unix:///foo/bar  ,  /foo/bar"
+
+
+@pytest.mark.parametrize(
+    ("init_hosts", "test_host", "expected"),
+    [
+        ## Never Trust trust
+        ## -----------------------------
+        # Test IPv4 Addresses
+        (_TRUSTED_NOTHING, "127.0.0.0", False),
+        (_TRUSTED_NOTHING, "127.0.0.1", False),
+        (_TRUSTED_NOTHING, "127.1.1.1", False),
+        (_TRUSTED_NOTHING, "127.255.255.255", False),
+        (_TRUSTED_NOTHING, "10.0.0.0", False),
+        (_TRUSTED_NOTHING, "10.0.0.1", False),
+        (_TRUSTED_NOTHING, "10.1.1.1", False),
+        (_TRUSTED_NOTHING, "10.255.255.255", False),
+        (_TRUSTED_NOTHING, "192.168.0.0", False),
+        (_TRUSTED_NOTHING, "192.168.0.1", False),
+        (_TRUSTED_NOTHING, "1.1.1.1", False),
+        # Test IPv6 Addresses
+        (_TRUSTED_NOTHING, "2001:db8::", False),
+        (_TRUSTED_NOTHING, "2001:db8:abcd:0012::0", False),
+        (_TRUSTED_NOTHING, "2001:db8:abcd:0012::1:1", False),
+        (_TRUSTED_NOTHING, "::", False),
+        (_TRUSTED_NOTHING, "::1", False),
+        (
+            _TRUSTED_NOTHING,
+            "2001:db8:3333:4444:5555:6666:102:304",
+            False,
+        ),  # aka 2001:db8:3333:4444:5555:6666:1.2.3.4
+        (_TRUSTED_NOTHING, "::b16:212c", False),  # aka ::11.22.33.44
+        (_TRUSTED_NOTHING, "a:b:c:d::", False),
+        (_TRUSTED_NOTHING, "::a:b:c:d", False),
+        # Test Literals
+        (_TRUSTED_NOTHING, "some-literal", False),
+        (_TRUSTED_NOTHING, "unix:///foo/bar", False),
+        (_TRUSTED_NOTHING, "/foo/bar", False),
+        (_TRUSTED_NOTHING, "*", False),
+        (_TRUSTED_NOTHING, "another-literal", False),
+        (_TRUSTED_NOTHING, "unix:///another/path", False),
+        (_TRUSTED_NOTHING, "/another/path", False),
+        (_TRUSTED_NOTHING, "", False),
+        ## Always trust
+        ## -----------------------------
+        # Test IPv4 Addresses
+        (_TRUSTED_EVERYTHING, "127.0.0.0", True),
+        (_TRUSTED_EVERYTHING, "127.0.0.1", True),
+        (_TRUSTED_EVERYTHING, "127.1.1.1", True),
+        (_TRUSTED_EVERYTHING, "127.255.255.255", True),
+        (_TRUSTED_EVERYTHING, "10.0.0.0", True),
+        (_TRUSTED_EVERYTHING, "10.0.0.1", True),
+        (_TRUSTED_EVERYTHING, "10.1.1.1", True),
+        (_TRUSTED_EVERYTHING, "10.255.255.255", True),
+        (_TRUSTED_EVERYTHING, "192.168.0.0", True),
+        (_TRUSTED_EVERYTHING, "192.168.0.1", True),
+        (_TRUSTED_EVERYTHING, "1.1.1.1", True),
+        # Test IPv6 Addresses
+        (_TRUSTED_EVERYTHING, "2001:db8::", True),
+        (_TRUSTED_EVERYTHING, "2001:db8:abcd:0012::0", True),
+        (_TRUSTED_EVERYTHING, "2001:db8:abcd:0012::1:1", True),
+        (_TRUSTED_EVERYTHING, "::", True),
+        (_TRUSTED_EVERYTHING, "::1", True),
+        (
+            _TRUSTED_EVERYTHING,
+            "2001:db8:3333:4444:5555:6666:102:304",
+            True,
+        ),  # aka 2001:db8:3333:4444:5555:6666:1.2.3.4
+        (_TRUSTED_EVERYTHING, "::b16:212c", True),  # aka ::11.22.33.44
+        (_TRUSTED_EVERYTHING, "a:b:c:d::", True),
+        (_TRUSTED_EVERYTHING, "::a:b:c:d", True),
+        # Test Literals
+        (_TRUSTED_EVERYTHING, "some-literal", True),
+        (_TRUSTED_EVERYTHING, "unix:///foo/bar", True),
+        (_TRUSTED_EVERYTHING, "/foo/bar", True),
+        (_TRUSTED_EVERYTHING, "*", True),
+        (_TRUSTED_EVERYTHING, "another-literal", True),
+        (_TRUSTED_EVERYTHING, "unix:///another/path", True),
+        (_TRUSTED_EVERYTHING, "/another/path", True),
+        (_TRUSTED_EVERYTHING, "", True),
+        ## Trust IPv4 Addresses
+        ## -----------------------------
+        # Test IPv4 Addresses
+        (_TRUSTED_IPv4_ADDRESSES, "127.0.0.0", False),
+        (_TRUSTED_IPv4_ADDRESSES, "127.0.0.1", True),
+        (_TRUSTED_IPv4_ADDRESSES, "127.1.1.1", False),
+        (_TRUSTED_IPv4_ADDRESSES, "127.255.255.255", False),
+        (_TRUSTED_IPv4_ADDRESSES, "10.0.0.0", False),
+        (_TRUSTED_IPv4_ADDRESSES, "10.0.0.1", True),
+        (_TRUSTED_IPv4_ADDRESSES, "10.1.1.1", False),
+        (_TRUSTED_IPv4_ADDRESSES, "10.255.255.255", False),
+        (_TRUSTED_IPv4_ADDRESSES, "192.168.0.0", False),
+        (_TRUSTED_IPv4_ADDRESSES, "192.168.0.1", False),
+        (_TRUSTED_IPv4_ADDRESSES, "1.1.1.1", False),
+        # Test IPv6 Addresses
+        (_TRUSTED_IPv4_ADDRESSES, "2001:db8::", False),
+        (_TRUSTED_IPv4_ADDRESSES, "2001:db8:abcd:0012::0", False),
+        (_TRUSTED_IPv4_ADDRESSES, "2001:db8:abcd:0012::1:1", False),
+        (_TRUSTED_IPv4_ADDRESSES, "::", False),
+        (_TRUSTED_IPv4_ADDRESSES, "::1", False),
+        (
+            _TRUSTED_IPv4_ADDRESSES,
+            "2001:db8:3333:4444:5555:6666:102:304",
+            False,
+        ),  # aka 2001:db8:3333:4444:5555:6666:1.2.3.4
+        (_TRUSTED_IPv4_ADDRESSES, "::b16:212c", False),  # aka ::11.22.33.44
+        (_TRUSTED_IPv4_ADDRESSES, "a:b:c:d::", False),
+        (_TRUSTED_IPv4_ADDRESSES, "::a:b:c:d", False),
+        # Test Literals
+        (_TRUSTED_IPv4_ADDRESSES, "some-literal", False),
+        (_TRUSTED_IPv4_ADDRESSES, "unix:///foo/bar", False),
+        (_TRUSTED_IPv4_ADDRESSES, "*", False),
+        (_TRUSTED_IPv4_ADDRESSES, "/foo/bar", False),
+        (_TRUSTED_IPv4_ADDRESSES, "another-literal", False),
+        (_TRUSTED_IPv4_ADDRESSES, "unix:///another/path", False),
+        (_TRUSTED_IPv4_ADDRESSES, "/another/path", False),
+        (_TRUSTED_IPv4_ADDRESSES, "", False),
+        ## Trust IPv6 Addresses
+        ## -----------------------------
+        # Test IPv4 Addresses
+        (_TRUSTED_IPv6_ADDRESSES, "127.0.0.0", False),
+        (_TRUSTED_IPv6_ADDRESSES, "127.0.0.1", False),
+        (_TRUSTED_IPv6_ADDRESSES, "127.1.1.1", False),
+        (_TRUSTED_IPv6_ADDRESSES, "127.255.255.255", False),
+        (_TRUSTED_IPv6_ADDRESSES, "10.0.0.0", False),
+        (_TRUSTED_IPv6_ADDRESSES, "10.0.0.1", False),
+        (_TRUSTED_IPv6_ADDRESSES, "10.1.1.1", False),
+        (_TRUSTED_IPv6_ADDRESSES, "10.255.255.255", False),
+        (_TRUSTED_IPv6_ADDRESSES, "192.168.0.0", False),
+        (_TRUSTED_IPv6_ADDRESSES, "192.168.0.1", False),
+        (_TRUSTED_IPv6_ADDRESSES, "1.1.1.1", False),
+        # Test IPv6 Addresses
+        (_TRUSTED_IPv6_ADDRESSES, "2001:db8::", True),
+        (_TRUSTED_IPv6_ADDRESSES, "2001:db8:abcd:0012::0", False),
+        (_TRUSTED_IPv6_ADDRESSES, "2001:db8:abcd:0012::1:1", False),
+        (_TRUSTED_IPv6_ADDRESSES, "::", False),
+        (_TRUSTED_IPv6_ADDRESSES, "::1", False),
+        (
+            _TRUSTED_IPv6_ADDRESSES,
+            "2001:db8:3333:4444:5555:6666:102:304",
+            True,
+        ),  # aka 2001:db8:3333:4444:5555:6666:1.2.3.4
+        (_TRUSTED_IPv6_ADDRESSES, "::b16:212c", True),  # aka ::11.22.33.44
+        (_TRUSTED_IPv6_ADDRESSES, "a:b:c:d::", False),
+        (_TRUSTED_IPv6_ADDRESSES, "::a:b:c:d", False),
+        # Test Literals
+        (_TRUSTED_IPv6_ADDRESSES, "some-literal", False),
+        (_TRUSTED_IPv6_ADDRESSES, "unix:///foo/bar", False),
+        (_TRUSTED_IPv6_ADDRESSES, "*", False),
+        (_TRUSTED_IPv6_ADDRESSES, "/foo/bar", False),
+        (_TRUSTED_IPv6_ADDRESSES, "another-literal", False),
+        (_TRUSTED_IPv6_ADDRESSES, "unix:///another/path", False),
+        (_TRUSTED_IPv6_ADDRESSES, "/another/path", False),
+        (_TRUSTED_IPv6_ADDRESSES, "", False),
+        ## Trust IPv4 Networks
+        ## -----------------------------
+        # Test IPv4 Addresses
+        (_TRUSTED_IPv4_NETWORKS, "127.0.0.0", True),
+        (_TRUSTED_IPv4_NETWORKS, "127.0.0.1", True),
+        (_TRUSTED_IPv4_NETWORKS, "127.1.1.1", True),
+        (_TRUSTED_IPv4_NETWORKS, "127.255.255.255", True),
+        (_TRUSTED_IPv4_NETWORKS, "10.0.0.0", True),
+        (_TRUSTED_IPv4_NETWORKS, "10.0.0.1", True),
+        (_TRUSTED_IPv4_NETWORKS, "10.1.1.1", True),
+        (_TRUSTED_IPv4_NETWORKS, "10.255.255.255", True),
+        (_TRUSTED_IPv4_NETWORKS, "192.168.0.0", False),
+        (_TRUSTED_IPv4_NETWORKS, "192.168.0.1", False),
+        (_TRUSTED_IPv4_NETWORKS, "1.1.1.1", False),
+        # Test IPv6 Addresses
+        (_TRUSTED_IPv4_NETWORKS, "2001:db8::", False),
+        (_TRUSTED_IPv4_NETWORKS, "2001:db8:abcd:0012::0", False),
+        (_TRUSTED_IPv4_NETWORKS, "2001:db8:abcd:0012::1:1", False),
+        (_TRUSTED_IPv4_NETWORKS, "::", False),
+        (_TRUSTED_IPv4_NETWORKS, "::1", False),
+        (
+            _TRUSTED_IPv4_NETWORKS,
+            "2001:db8:3333:4444:5555:6666:102:304",
+            False,
+        ),  # aka 2001:db8:3333:4444:5555:6666:1.2.3.4
+        (_TRUSTED_IPv4_NETWORKS, "::b16:212c", False),  # aka ::11.22.33.44
+        (_TRUSTED_IPv4_NETWORKS, "a:b:c:d::", False),
+        (_TRUSTED_IPv4_NETWORKS, "::a:b:c:d", False),
+        # Test Literals
+        (_TRUSTED_IPv4_NETWORKS, "some-literal", False),
+        (_TRUSTED_IPv4_NETWORKS, "unix:///foo/bar", False),
+        (_TRUSTED_IPv4_NETWORKS, "*", False),
+        (_TRUSTED_IPv4_NETWORKS, "/foo/bar", False),
+        (_TRUSTED_IPv4_NETWORKS, "another-literal", False),
+        (_TRUSTED_IPv4_NETWORKS, "unix:///another/path", False),
+        (_TRUSTED_IPv4_NETWORKS, "/another/path", False),
+        (_TRUSTED_IPv4_NETWORKS, "", False),
+        ## Trust IPv6 Networks
+        ## -----------------------------
+        # Test IPv4 Addresses
+        (_TRUSTED_IPv6_NETWORKS, "127.0.0.0", False),
+        (_TRUSTED_IPv6_NETWORKS, "127.0.0.1", False),
+        (_TRUSTED_IPv6_NETWORKS, "127.1.1.1", False),
+        (_TRUSTED_IPv6_NETWORKS, "127.255.255.255", False),
+        (_TRUSTED_IPv6_NETWORKS, "10.0.0.0", False),
+        (_TRUSTED_IPv6_NETWORKS, "10.0.0.1", False),
+        (_TRUSTED_IPv6_NETWORKS, "10.1.1.1", False),
+        (_TRUSTED_IPv6_NETWORKS, "10.255.255.255", False),
+        (_TRUSTED_IPv6_NETWORKS, "192.168.0.0", False),
+        (_TRUSTED_IPv6_NETWORKS, "192.168.0.1", False),
+        (_TRUSTED_IPv6_NETWORKS, "1.1.1.1", False),
+        # Test IPv6 Addresses
+        (_TRUSTED_IPv6_NETWORKS, "2001:db8::", False),
+        (_TRUSTED_IPv6_NETWORKS, "2001:db8:abcd:0012::0", True),
+        (_TRUSTED_IPv6_NETWORKS, "2001:db8:abcd:0012::1:1", True),
+        (_TRUSTED_IPv6_NETWORKS, "::", False),
+        (_TRUSTED_IPv6_NETWORKS, "::1", False),
+        (
+            _TRUSTED_IPv6_NETWORKS,
+            "2001:db8:3333:4444:5555:6666:102:304",
+            False,
+        ),  # aka 2001:db8:3333:4444:5555:6666:1.2.3.4
+        (_TRUSTED_IPv6_NETWORKS, "::b16:212c", False),  # aka ::11.22.33.44
+        (_TRUSTED_IPv6_NETWORKS, "a:b:c:d::", False),
+        (_TRUSTED_IPv6_NETWORKS, "::a:b:c:d", False),
+        # Test Literals
+        (_TRUSTED_IPv6_NETWORKS, "some-literal", False),
+        (_TRUSTED_IPv6_NETWORKS, "unix:///foo/bar", False),
+        (_TRUSTED_IPv6_NETWORKS, "*", False),
+        (_TRUSTED_IPv6_NETWORKS, "/foo/bar", False),
+        (_TRUSTED_IPv6_NETWORKS, "another-literal", False),
+        (_TRUSTED_IPv6_NETWORKS, "unix:///another/path", False),
+        (_TRUSTED_IPv6_NETWORKS, "/another/path", False),
+        (_TRUSTED_IPv6_NETWORKS, "", False),
+        ## Trust Literals
+        ## -----------------------------
+        # Test IPv4 Addresses
+        (_TRUSTED_LITERALS, "127.0.0.0", False),
+        (_TRUSTED_LITERALS, "127.0.0.1", False),
+        (_TRUSTED_LITERALS, "127.1.1.1", False),
+        (_TRUSTED_LITERALS, "127.255.255.255", False),
+        (_TRUSTED_LITERALS, "10.0.0.0", False),
+        (_TRUSTED_LITERALS, "10.0.0.1", False),
+        (_TRUSTED_LITERALS, "10.1.1.1", False),
+        (_TRUSTED_LITERALS, "10.255.255.255", False),
+        (_TRUSTED_LITERALS, "192.168.0.0", False),
+        (_TRUSTED_LITERALS, "192.168.0.1", False),
+        (_TRUSTED_LITERALS, "1.1.1.1", False),
+        # Test IPv6 Addresses
+        (_TRUSTED_LITERALS, "2001:db8::", False),
+        (_TRUSTED_LITERALS, "2001:db8:abcd:0012::0", False),
+        (_TRUSTED_LITERALS, "2001:db8:abcd:0012::1:1", False),
+        (_TRUSTED_LITERALS, "::", False),
+        (_TRUSTED_LITERALS, "::1", False),
+        (
+            _TRUSTED_LITERALS,
+            "2001:db8:3333:4444:5555:6666:102:304",
+            False,
+        ),  # aka 2001:db8:3333:4444:5555:6666:1.2.3.4
+        (_TRUSTED_LITERALS, "::b16:212c", False),  # aka ::11.22.33.44
+        (_TRUSTED_LITERALS, "a:b:c:d::", False),
+        (_TRUSTED_LITERALS, "::a:b:c:d", False),
+        # Test Literals
+        (_TRUSTED_LITERALS, "some-literal", True),
+        (_TRUSTED_LITERALS, "unix:///foo/bar", True),
+        (_TRUSTED_LITERALS, "*", False),
+        (_TRUSTED_LITERALS, "/foo/bar", True),
+        (_TRUSTED_LITERALS, "another-literal", False),
+        (_TRUSTED_LITERALS, "unix:///another/path", False),
+        (_TRUSTED_LITERALS, "/another/path", False),
+        (_TRUSTED_LITERALS, "", False),
+    ],
+)
+def test_forwarded_hosts(init_hosts: str | list[str], test_host: str, expected: bool) -> None:
+    trusted_hosts = _TrustedHosts(init_hosts)
+    assert (test_host in trusted_hosts) is expected
+
+
 @pytest.mark.anyio
 @pytest.mark.parametrize(
-    ("trusted_hosts", "response_text"),
+    ("trusted_hosts", "expected"),
     [
         # always trust
-        ("*", "Remote: https://1.2.3.4:0"),
+        ("*", "https://1.2.3.4:0"),
         # trusted proxy
-        ("127.0.0.1", "Remote: https://1.2.3.4:0"),
-        (["127.0.0.1"], "Remote: https://1.2.3.4:0"),
+        ("127.0.0.1", "https://1.2.3.4:0"),
+        (["127.0.0.1"], "https://1.2.3.4:0"),
         # 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"),
+        (["127.0.0.1", "10.0.0.1"], "https://1.2.3.4:0"),
+        ("127.0.0.1, 10.0.0.1", "https://1.2.3.4:0"),
+        # trusted proxy network
+        # https://github.com/encode/uvicorn/issues/1068#issuecomment-1004813267
+        ("127.0.0.0/24, 10.0.0.1", "https://1.2.3.4:0"),
         # request from untrusted proxy
-        ("192.168.0.1", "Remote: http://127.0.0.1:123"),
+        ("192.168.0.1", "http://127.0.0.1:123"),
+        # request from untrusted proxy network
+        ("192.168.0.0/16", "http://127.0.0.1:123"),
+        # request from client running on proxy server itself
+        # https://github.com/encode/uvicorn/issues/1068#issuecomment-855371576
+        (["127.0.0.1", "1.2.3.4"], "https://1.2.3.4:0"),
     ],
 )
-async def test_proxy_headers_trusted_hosts(trusted_hosts: list[str] | str, response_text: str) -> None:
-    app_with_middleware = ProxyHeadersMiddleware(app, trusted_hosts=trusted_hosts)
-    transport = httpx.ASGITransport(app=app_with_middleware)  # type: ignore
-    async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
-        headers = {"X-Forwarded-Proto": "https", "X-Forwarded-For": "1.2.3.4"}
-        response = await client.get("/", headers=headers)
-
-    assert response.status_code == 200
-    assert response.text == response_text
-
-
-@pytest.mark.anyio
-@pytest.mark.parametrize(
-    ("trusted_hosts", "response_text"),
-    [
-        # always trust
-        ("*", "Remote: https://1.2.3.4:0"),
-        # all proxies are trusted
-        (
-            ["127.0.0.1", "10.0.2.1", "192.168.0.2"],
-            "Remote: https://1.2.3.4:0",
-        ),
-        # order doesn't matter
-        (
-            ["10.0.2.1", "192.168.0.2", "127.0.0.1"],
-            "Remote: https://1.2.3.4:0",
-        ),
-        # should set first untrusted as remote address
-        (["192.168.0.2", "127.0.0.1"], "Remote: https://10.0.2.1:0"),
-    ],
-)
-async def test_proxy_headers_multiple_proxies(trusted_hosts: list[str] | str, response_text: str) -> None:
-    app_with_middleware = ProxyHeadersMiddleware(app, trusted_hosts=trusted_hosts)
-    transport = httpx.ASGITransport(app=app_with_middleware)  # type: ignore
-    async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
+async def test_proxy_headers_trusted_hosts(trusted_hosts: str | list[str], expected: str) -> None:
+    async with make_httpx_client(trusted_hosts) as client:
         headers = {
-            "X-Forwarded-Proto": "https",
-            "X-Forwarded-For": "1.2.3.4, 10.0.2.1, 192.168.0.2",
+            X_FORWARDED_FOR: "1.2.3.4",
+            X_FORWARDED_PROTO: "https",
         }
         response = await client.get("/", headers=headers)
-
     assert response.status_code == 200
-    assert response.text == response_text
+    assert response.text == expected
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    ("forwarded_for", "forwarded_proto", "expected"),
+    [
+        ("", "", "http://127.0.0.1:123"),
+        ("", None, "http://127.0.0.1:123"),
+        ("", "asdf", "http://127.0.0.1:123"),
+        (" , ", "https", "https://127.0.0.1:123"),
+        (", , ", "https", "https://127.0.0.1:123"),
+        (" , 10.0.0.1", "https", "https://127.0.0.1:123"),
+        ("9.9.9.9 , , , 10.0.0.1", "https", "https://127.0.0.1:123"),
+        (", , 9.9.9.9", "https", "https://9.9.9.9:0"),
+        (", , 9.9.9.9, , ", "https", "https://127.0.0.1:123"),
+    ],
+)
+async def test_proxy_headers_trusted_hosts_malformed(
+    forwarded_for: str,
+    forwarded_proto: str | None,
+    expected: str,
+) -> None:
+    async with make_httpx_client("127.0.0.1, 10.0.0.0/8") as client:
+        headers = {X_FORWARDED_FOR: forwarded_for}
+        if forwarded_proto is not None:
+            headers[X_FORWARDED_PROTO] = forwarded_proto
+        response = await client.get("/", headers=headers)
+    assert response.status_code == 200
+    assert response.text == expected
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    ("trusted_hosts", "expected"),
+    [
+        # always trust
+        ("*", "https://1.2.3.4:0"),
+        # all proxies are trusted
+        (["127.0.0.1", "10.0.2.1", "192.168.0.2"], "https://1.2.3.4:0"),
+        # order doesn't matter
+        (["10.0.2.1", "192.168.0.2", "127.0.0.1"], "https://1.2.3.4:0"),
+        # should set first untrusted as remote address
+        (["192.168.0.2", "127.0.0.1"], "https://10.0.2.1:0"),
+        # Mixed literals and networks
+        (["127.0.0.1", "10.0.0.0/8", "192.168.0.2"], "https://1.2.3.4:0"),
+    ],
+)
+async def test_proxy_headers_multiple_proxies(trusted_hosts: str | list[str], expected: str) -> None:
+    async with make_httpx_client(trusted_hosts) as client:
+        headers = {
+            X_FORWARDED_FOR: "1.2.3.4, 10.0.2.1, 192.168.0.2",
+            X_FORWARDED_PROTO: "https",
+        }
+        response = await client.get("/", headers=headers)
+    assert response.status_code == 200
+    assert response.text == expected
 
 
 @pytest.mark.anyio
 async def test_proxy_headers_invalid_x_forwarded_for() -> None:
-    app_with_middleware = ProxyHeadersMiddleware(app, trusted_hosts="*")
-    transport = httpx.ASGITransport(app=app_with_middleware)  # type: ignore
-    async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
+    async with make_httpx_client("*") as client:
         headers = httpx.Headers(
             {
-                "X-Forwarded-Proto": "https",
-                "X-Forwarded-For": "1.2.3.4, \xf0\xfd\xfd\xfd",
+                X_FORWARDED_FOR: "1.2.3.4, \xf0\xfd\xfd\xfd, unix:, ::1",
+                X_FORWARDED_PROTO: "https",
             },
             encoding="latin-1",
         )
         response = await client.get("/", headers=headers)
     assert response.status_code == 200
-    assert response.text == "Remote: https://1.2.3.4:0"
+    assert response.text == "https://1.2.3.4:0"
 
 
 @pytest.mark.anyio
 @pytest.mark.parametrize(
-    "x_forwarded_proto,addr",
+    "forwarded_proto,expected",
     [
         ("http", "ws://1.2.3.4:0"),
         ("https", "wss://1.2.3.4:0"),
@@ -120,8 +458,8 @@ async def test_proxy_headers_invalid_x_forwarded_for() -> None:
     ],
 )
 async def test_proxy_headers_websocket_x_forwarded_proto(
-    x_forwarded_proto: str,
-    addr: str,
+    forwarded_proto: str,
+    expected: str,
     ws_protocol_cls: type[WSProtocol | WebSocketProtocol],
     http_protocol_cls: type[H11Protocol | HttpToolsProtocol],
     unused_tcp_port: int,
@@ -131,9 +469,8 @@ async def test_proxy_headers_websocket_x_forwarded_proto(
         scheme = scope["scheme"]
         assert scope["client"] is not None
         host, port = scope["client"]
-        addr = "%s://%s:%d" % (scheme, host, port)
         await send({"type": "websocket.accept"})
-        await send({"type": "websocket.send", "text": addr})
+        await send({"type": "websocket.send", "text": f"{scheme}://{host}:{port}"})
 
     app_with_middleware = ProxyHeadersMiddleware(websocket_app, trusted_hosts="*")
     config = Config(
@@ -146,7 +483,24 @@ async def test_proxy_headers_websocket_x_forwarded_proto(
 
     async with run_server(config):
         url = f"ws://127.0.0.1:{unused_tcp_port}"
-        headers = {"X-Forwarded-Proto": x_forwarded_proto, "X-Forwarded-For": "1.2.3.4"}
+        headers = {
+            X_FORWARDED_FOR: "1.2.3.4",
+            X_FORWARDED_PROTO: forwarded_proto,
+        }
         async with websockets.client.connect(url, extra_headers=headers) as websocket:
             data = await websocket.recv()
-            assert data == addr
+            assert data == expected
+
+
+@pytest.mark.anyio
+async def test_proxy_headers_empty_x_forwarded_for() -> None:
+    # fallback to the default behavior if x-forwarded-for is an empty list
+    # https://github.com/encode/uvicorn/issues/1068#issuecomment-855371576
+    async with make_httpx_client("*") as client:
+        headers = {
+            X_FORWARDED_FOR: "",
+            X_FORWARDED_PROTO: "https",
+        }
+        response = await client.get("/", headers=headers)
+    assert response.status_code == 200
+    assert response.text == "https://127.0.0.1:123"

uvicorn/main.py

@@ -240,8 +240,10 @@ def print_version(ctx: click.Context, param: click.Parameter, value: bool) -> No
     "--forwarded-allow-ips",
     type=str,
     default=None,
-    help="Comma separated list of IPs to trust with proxy headers. Defaults to"
-    " the $FORWARDED_ALLOW_IPS environment variable if available, or '127.0.0.1'.",
+    help="Comma separated list of IP Addresses, IP Networks, or literals "
+    "(e.g. UNIX Socket path) to trust with proxy headers. Defaults to the "
+    "$FORWARDED_ALLOW_IPS environment variable if available, or '127.0.0.1'. "
+    "The literal '*' means trust everything.",
 )
 @click.option(
     "--root-path",

uvicorn/middleware/proxy_headers.py

@@ -1,70 +1,150 @@
-"""
-This middleware can be used when a known proxy is fronting the application,
-and is trusted to be properly setting the `X-Forwarded-Proto` and
-`X-Forwarded-For` headers with the connecting client information.
-
-Modifies the `client` and `scheme` information so that they reference
-the connecting client, rather that the connecting proxy.
-
-https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers#Proxies
-"""
-
 from __future__ import annotations
 
+import ipaddress
 from typing import Union, cast
 
 from uvicorn._types import ASGI3Application, ASGIReceiveCallable, ASGISendCallable, HTTPScope, Scope, WebSocketScope
 
 
+def _parse_raw_hosts(value: str) -> list[str]:
+    return [item.strip() for item in value.split(",")]
+
+
+class _TrustedHosts:
+    """Container for trusted hosts and networks"""
+
+    def __init__(
+        self,
+        trusted_hosts: list[str] | str,
+    ) -> None:
+        self.always_trust: bool = trusted_hosts == "*"
+
+        self.trusted_literals: set[str] = set()
+        self.trusted_hosts: set[ipaddress._BaseAddress] = set()
+        self.trusted_networks: set[ipaddress._BaseNetwork] = set()
+
+        # Notes:
+        # - We seperate hosts from literals as there are many ways to write
+        #   an IPv6 Address so we need to compare by object.
+        # - We don't convert IP Address to single host networks (e.g. /32 / 128) as
+        #   it more efficient to do an address lookup in a set than check for
+        #   membership in each network.
+        # - We still allow literals as it might be possible that we receive a
+        #   something that isn't an IP Address e.g. a unix socket.
+
+        if not self.always_trust:
+            if isinstance(trusted_hosts, str):
+                trusted_hosts = _parse_raw_hosts(trusted_hosts)
+
+            for host in trusted_hosts:
+                # Note: because we always convert invalid IP types to literals it
+                # is not possible for the user to know they provided a malformed IP
+                # type - this may lead to unexpected / difficult to debug behaviour.
+
+                if "/" in host:
+                    # Looks like a network
+                    try:
+                        self.trusted_networks.add(ipaddress.ip_network(host))
+                    except ValueError:
+                        # Was not a valid IP Network
+                        self.trusted_literals.add(host)
+                else:
+                    try:
+                        self.trusted_hosts.add(ipaddress.ip_address(host))
+                    except ValueError:
+                        # Was not a valid IP Adress
+                        self.trusted_literals.add(host)
+        return
+
+    def __contains__(self, item: str | None) -> bool:
+        if self.always_trust:
+            return True
+
+        if not item:
+            return False
+
+        try:
+            ip = ipaddress.ip_address(item)
+            if ip in self.trusted_hosts:
+                return True
+            return any(ip in net for net in self.trusted_networks)
+
+        except ValueError:
+            return item in self.trusted_literals
+
+    def get_trusted_client_host(self, x_forwarded_for: str) -> str:
+        """Extract the client host from x_forwarded_for header
+
+        In general this is the first "untrusted" host in the forwarded for list.
+        """
+        x_forwarded_for_hosts = _parse_raw_hosts(x_forwarded_for)
+
+        if self.always_trust:
+            return x_forwarded_for_hosts[0]
+
+        # Note: each proxy appends to the header list so check it in reverse order
+        for host in reversed(x_forwarded_for_hosts):
+            if host not in self:
+                return host
+
+        # All hosts are trusted meaning that the client was also a trusted proxy
+        # See https://github.com/encode/uvicorn/issues/1068#issuecomment-855371576
+        return x_forwarded_for_hosts[0]
+
+
 class ProxyHeadersMiddleware:
+    """Middleware for handling known proxy headers
+
+    This middleware can be used when a known proxy is fronting the application,
+    and is trusted to be properly setting the `X-Forwarded-Proto` and
+    `X-Forwarded-For` headers with the connecting client information.
+
+    Modifies the `client` and `scheme` information so that they reference
+    the connecting client, rather that the connecting proxy.
+
+    References:
+    - <https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers#Proxies>
+    - <https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/X-Forwarded-For>
+    """
+
     def __init__(
         self,
         app: ASGI3Application,
         trusted_hosts: list[str] | str = "127.0.0.1",
     ) -> None:
         self.app = app
-        if isinstance(trusted_hosts, str):
-            self.trusted_hosts = {item.strip() for item in trusted_hosts.split(",")}
-        else:
-            self.trusted_hosts = set(trusted_hosts)
-        self.always_trust = "*" in self.trusted_hosts
-
-    def get_trusted_client_host(self, x_forwarded_for_hosts: list[str]) -> str | None:
-        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:
-                return host
-
-        return None  # pragma: full coverage
+        self.trusted_hosts = _TrustedHosts(trusted_hosts)
 
     async def __call__(self, scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None:
         if scope["type"] in ("http", "websocket"):
-            scope = cast(Union["HTTPScope", "WebSocketScope"], scope)
+            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 client_host in self.trusted_hosts:
                 headers = dict(scope["headers"])
 
                 if b"x-forwarded-proto" in headers:
-                    # Determine if the incoming request was http or https based on
-                    # the X-Forwarded-Proto header.
                     x_forwarded_proto = headers[b"x-forwarded-proto"].decode("latin1").strip()
-                    if scope["type"] == "websocket":
-                        scope["scheme"] = x_forwarded_proto.replace("http", "ws")
-                    else:
-                        scope["scheme"] = x_forwarded_proto
+
+                    if x_forwarded_proto in {"http", "https", "ws", "wss"}:
+                        if scope["type"] == "websocket":
+                            scope["scheme"] = x_forwarded_proto.replace("http", "ws")
+                        else:
+                            scope["scheme"] = x_forwarded_proto
 
                 if b"x-forwarded-for" in headers:
-                    # Determine the client address from the last trusted IP in the
-                    # X-Forwarded-For header. We've lost the connecting client's port
-                    # information by now, so only include the host.
                     x_forwarded_for = headers[b"x-forwarded-for"].decode("latin1")
-                    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]
+                    host = self.trusted_hosts.get_trusted_client_host(x_forwarded_for)
+
+                    if host:
+                        # If the x-forwarded-for header is empty then host is an empty string.
+                        # Only set the client if we actually got something usable.
+                        # See: https://github.com/encode/uvicorn/issues/1068
+
+                        # We've lost the connecting client's port information by now,
+                        # so only include the host.
+                        port = 0
+                        scope["client"] = (host, port)
 
         return await self.app(scope, receive, send)