| 296 | |
| 297 | |
| 298 | class _HTTPRequestContext(object): |
| 299 | def __init__( |
| 300 | self, |
| 301 | stream: iostream.IOStream, |
| 302 | address: Tuple, |
| 303 | protocol: Optional[str], |
| 304 | trusted_downstream: Optional[List[str]] = None, |
| 305 | ) -> None: |
| 306 | self.address = address |
| 307 | # Save the socket's address family now so we know how to |
| 308 | # interpret self.address even after the stream is closed |
| 309 | # and its socket attribute replaced with None. |
| 310 | if stream.socket is not None: |
| 311 | self.address_family = stream.socket.family |
| 312 | else: |
| 313 | self.address_family = None |
| 314 | # In HTTPServerRequest we want an IP, not a full socket address. |
| 315 | if ( |
| 316 | self.address_family in (socket.AF_INET, socket.AF_INET6) |
| 317 | and address is not None |
| 318 | ): |
| 319 | self.remote_ip = address[0] |
| 320 | else: |
| 321 | # Unix (or other) socket; fake the remote address. |
| 322 | self.remote_ip = "0.0.0.0" |
| 323 | if protocol: |
| 324 | self.protocol = protocol |
| 325 | elif isinstance(stream, iostream.SSLIOStream): |
| 326 | self.protocol = "https" |
| 327 | else: |
| 328 | self.protocol = "http" |
| 329 | self._orig_remote_ip = self.remote_ip |
| 330 | self._orig_protocol = self.protocol |
| 331 | self.trusted_downstream = set(trusted_downstream or []) |
| 332 | |
| 333 | def __str__(self) -> str: |
| 334 | if self.address_family in (socket.AF_INET, socket.AF_INET6): |
| 335 | return self.remote_ip |
| 336 | elif isinstance(self.address, bytes): |
| 337 | # Python 3 with the -bb option warns about str(bytes), |
| 338 | # so convert it explicitly. |
| 339 | # Unix socket addresses are str on mac but bytes on linux. |
| 340 | return native_str(self.address) |
| 341 | else: |
| 342 | return str(self.address) |
| 343 | |
| 344 | def _apply_xheaders(self, headers: httputil.HTTPHeaders) -> None: |
| 345 | """Rewrite the ``remote_ip`` and ``protocol`` fields.""" |
| 346 | # Squid uses X-Forwarded-For, others use X-Real-Ip |
| 347 | ip = headers.get("X-Forwarded-For", self.remote_ip) |
| 348 | # Skip trusted downstream hosts in X-Forwarded-For list |
| 349 | for ip in (cand.strip() for cand in reversed(ip.split(","))): |
| 350 | if ip not in self.trusted_downstream: |
| 351 | break |
| 352 | ip = headers.get("X-Real-Ip", ip) |
| 353 | if netutil.is_valid_ip(ip): |
| 354 | self.remote_ip = ip |
| 355 | # AWS uses X-Forwarded-Proto |