from __future__ import annotations import functools import ipaddress from uvicorn._types import ASGI3Application, ASGIReceiveCallable, ASGISendCallable, Scope 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: - - """ def __init__(self, app: ASGI3Application, trusted_hosts: list[str] | str = "127.0.0.1") -> None: self.app = app self.trusted_hosts = _TrustedHosts(trusted_hosts) async def __call__(self, scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None: if scope["type"] == "lifespan": return await self.app(scope, receive, send) client_addr = scope.get("client") client_host = client_addr[0] if client_addr else None if client_host in self.trusted_hosts: x_forwarded_proto_value: bytes | None = None x_forwarded_for_values: list[bytes] = [] for name, value in scope["headers"]: if name == b"x-forwarded-proto": x_forwarded_proto_value = value elif name == b"x-forwarded-for": x_forwarded_for_values.append(value) if x_forwarded_proto_value is not None: x_forwarded_proto = x_forwarded_proto_value.decode("latin1").strip() 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 x_forwarded_for_values: x_forwarded_for = b", ".join(x_forwarded_for_values).decode("latin1") host, port = self.trusted_hosts.get_trusted_client_address(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/Kludex/uvicorn/issues/1068 scope["client"] = (host, port) return await self.app(scope, receive, send) def _parse_raw_hosts(value: str) -> list[str]: return [item.strip() for item in value.split(",")] def _parse_host_port(value: str) -> tuple[str, int]: """Parse a forwarded host value into host and optional port. Accepts bare IPs, IPv4 `host:port`, and bracketed IPv6 `[host]:port`. Any unrecognized or malformed value is treated conservatively and returned without a port so trust checks do not silently normalize arbitrary input. """ if value.startswith("["): bracket_end = value.find("]") if bracket_end == -1: return value, 0 host = value[1:bracket_end] remainder = value[bracket_end + 1 :] if not remainder: return host, 0 if not remainder.startswith(":"): return value, 0 try: return host, int(remainder[1:]) except ValueError: return host, 0 if value.count(":") == 1: host, port = value.rsplit(":", 1) try: return host, int(port) except ValueError: return value, 0 return value, 0 class _TrustedHosts: """Container for trusted hosts and networks""" def __init__(self, trusted_hosts: list[str] | str) -> None: self.always_trust: bool = trusted_hosts in ("*", ["*"]) self.trusted_literals: set[str] = set() self.trusted_hosts: set[ipaddress.IPv4Address | ipaddress.IPv6Address] = set() self.trusted_networks: set[ipaddress.IPv4Network | ipaddress.IPv6Network] = set() # Notes: # - We separate 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 Address self.trusted_literals.add(host) self._trusts = functools.lru_cache(maxsize=4096)(self._compute_trust) def __contains__(self, host: str | None) -> bool: if self.always_trust: return True if not host: return False # Don't cache hosts longer than a DNS name (253); they can't be trusted and would pin huge cache keys. if len(host) > 253: return self._compute_trust(host) # pragma: no cover return self._trusts(host) def _compute_trust(self, host: str) -> bool: try: ip = ipaddress.ip_address(host) return ip in self.trusted_hosts or any(ip in net for net in self.trusted_networks) except ValueError: return host in self.trusted_literals def get_trusted_client_address(self, x_forwarded_for: str) -> tuple[str, int]: """Extract the client address 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 _parse_host_port(x_forwarded_for_hosts[0]) # Note: each proxy appends to the header list so check it in reverse order for host_port in reversed(x_forwarded_for_hosts): host, port = _parse_host_port(host_port) if host not in self: return host, port # All hosts are trusted meaning that the client was also a trusted proxy # See https://github.com/Kludex/uvicorn/issues/1068#issuecomment-855371576 return _parse_host_port(x_forwarded_for_hosts[0])