6
Self Samples
295
Cumulative
1
Lines Hit
1.01%
% of Total
6
Max Self
295
Max Total
Intensity:
Cold
Hot
Self Time
Total Time
Show All
Hot Only
Heat
Specialization
Line
Self
Total
Code
1
from __future__ import annotations
2
3
import functools
4
import ipaddress
5
6
from uvicorn._types import ASGI3Application, ASGIReceiveCallable, ASGISendCallable, Scope
7
8
9
class ProxyHeadersMiddleware:
10
"""Middleware for handling known proxy headers
11
12
This middleware can be used when a known proxy is fronting the application,
13
and is trusted to be properly setting the `X-Forwarded-Proto` and
14
`X-Forwarded-For` headers with the connecting client information.
15
16
Modifies the `client` and `scheme` information so that they reference
17
the connecting client, rather that the connecting proxy.
18
19
References:
20
- <https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers#Proxies>
21
- <https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/X-Forwarded-For>
22
"""
23
24
def __init__(self, app: ASGI3Application, trusted_hosts: list[str] | str = "127.0.0.1") -> None:
25
self.app = app
26
self.trusted_hosts = _TrustedHosts(trusted_hosts)
27
28
async def __call__(self, scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None:
29
if scope["type"] == "lifespan":
30
return await self.app(scope, receive, send)
31
32
client_addr = scope.get("client")
33
client_host = client_addr[0] if client_addr else None
34
35
if client_host in self.trusted_hosts:
36
x_forwarded_proto_value: bytes | None = None
37
x_forwarded_for_values: list[bytes] = []
38
for name, value in scope["headers"]:
39
if name == b"x-forwarded-proto":
40
x_forwarded_proto_value = value
41
elif name == b"x-forwarded-for":
42
x_forwarded_for_values.append(value)
43
44
if x_forwarded_proto_value is not None:
45
x_forwarded_proto = x_forwarded_proto_value.decode("latin1").strip()
46
47
if x_forwarded_proto in {"http", "https", "ws", "wss"}:
48
if scope["type"] == "websocket":
49
scope["scheme"] = x_forwarded_proto.replace("http", "ws")
50
else:
51
scope["scheme"] = x_forwarded_proto
52
53
if x_forwarded_for_values:
54
x_forwarded_for = b", ".join(x_forwarded_for_values).decode("latin1")
55
host, port = self.trusted_hosts.get_trusted_client_address(x_forwarded_for)
56
57
if host:
58
# If the x-forwarded-for header is empty then host is an empty string.
59
# Only set the client if we actually got something usable.
60
# See: https://github.com/Kludex/uvicorn/issues/1068
61
scope["client"] = (host, port)
62
63
6
295
return await self.app(scope, receive, send)
64
65
66
def _parse_raw_hosts(value: str) -> list[str]:
67
return [item.strip() for item in value.split(",")]
68
69
70
def _parse_host_port(value: str) -> tuple[str, int]:
71
"""Parse a forwarded host value into host and optional port.
72
73
Accepts bare IPs, IPv4 `host:port`, and bracketed IPv6 `[host]:port`.
74
Any unrecognized or malformed value is treated conservatively and returned
75
without a port so trust checks do not silently normalize arbitrary input.
76
"""
77
78
if value.startswith("["):
79
bracket_end = value.find("]")
80
if bracket_end == -1:
81
return value, 0
82
83
host = value[1:bracket_end]
84
remainder = value[bracket_end + 1 :]
85
if not remainder:
86
return host, 0
87
if not remainder.startswith(":"):
88
return value, 0
89
90
try:
91
return host, int(remainder[1:])
92
except ValueError:
93
return host, 0
94
95
if value.count(":") == 1:
96
host, port = value.rsplit(":", 1)
97
try:
98
return host, int(port)
99
except ValueError:
100
return value, 0
101
102
return value, 0
103
104
105
class _TrustedHosts:
106
"""Container for trusted hosts and networks"""
107
108
def __init__(self, trusted_hosts: list[str] | str) -> None:
109
self.always_trust: bool = trusted_hosts in ("*", ["*"])
110
111
self.trusted_literals: set[str] = set()
112
self.trusted_hosts: set[ipaddress.IPv4Address | ipaddress.IPv6Address] = set()
113
self.trusted_networks: set[ipaddress.IPv4Network | ipaddress.IPv6Network] = set()
114
115
# Notes:
116
# - We separate hosts from literals as there are many ways to write
117
# an IPv6 Address so we need to compare by object.
118
# - We don't convert IP Address to single host networks (e.g. /32 / 128) as
119
# it more efficient to do an address lookup in a set than check for
120
# membership in each network.
121
# - We still allow literals as it might be possible that we receive a
122
# something that isn't an IP Address e.g. a unix socket.
123
124
if not self.always_trust:
125
if isinstance(trusted_hosts, str):
126
trusted_hosts = _parse_raw_hosts(trusted_hosts)
127
128
for host in trusted_hosts:
129
# Note: because we always convert invalid IP types to literals it
130
# is not possible for the user to know they provided a malformed IP
131
# type - this may lead to unexpected / difficult to debug behaviour.
132
133
if "/" in host:
134
# Looks like a network
135
try:
136
self.trusted_networks.add(ipaddress.ip_network(host))
137
except ValueError:
138
# Was not a valid IP Network
139
self.trusted_literals.add(host)
140
else:
141
try:
142
self.trusted_hosts.add(ipaddress.ip_address(host))
143
except ValueError:
144
# Was not a valid IP Address
145
self.trusted_literals.add(host)
146
147
self._trusts = functools.lru_cache(maxsize=4096)(self._compute_trust)
148
149
def __contains__(self, host: str | None) -> bool:
150
if self.always_trust:
151
return True
152
153
if not host:
154
return False
155
156
# Don't cache hosts longer than a DNS name (253); they can't be trusted and would pin huge cache keys.
157
if len(host) > 253:
158
return self._compute_trust(host) # pragma: no cover
159
160
return self._trusts(host)
161
162
def _compute_trust(self, host: str) -> bool:
163
try:
164
ip = ipaddress.ip_address(host)
165
return ip in self.trusted_hosts or any(ip in net for net in self.trusted_networks)
166
except ValueError:
167
return host in self.trusted_literals
168
169
def get_trusted_client_address(self, x_forwarded_for: str) -> tuple[str, int]:
170
"""Extract the client address from x_forwarded_for header.
171
172
In general this is the first "untrusted" host in the forwarded for list.
173
"""
174
x_forwarded_for_hosts = _parse_raw_hosts(x_forwarded_for)
175
176
if self.always_trust:
177
return _parse_host_port(x_forwarded_for_hosts[0])
178
179
# Note: each proxy appends to the header list so check it in reverse order
180
for host_port in reversed(x_forwarded_for_hosts):
181
host, port = _parse_host_port(host_port)
182
if host not in self:
183
return host, port
184
185
# All hosts are trusted meaning that the client was also a trusted proxy
186
# See https://github.com/Kludex/uvicorn/issues/1068#issuecomment-855371576
187
return _parse_host_port(x_forwarded_for_hosts[0])