Rewrite the ``remote_ip`` and ``protocol`` fields.
(self, headers: httputil.HTTPHeaders)
| 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 |
| 356 | proto_header = headers.get( |
| 357 | "X-Scheme", headers.get("X-Forwarded-Proto", self.protocol) |
| 358 | ) |
| 359 | if proto_header: |
| 360 | # use only the last proto entry if there is more than one |
| 361 | # TODO: support trusting multiple layers of proxied protocol |
| 362 | proto_header = proto_header.split(",")[-1].strip() |
| 363 | if proto_header in ("http", "https"): |
| 364 | self.protocol = proto_header |
| 365 | |
| 366 | def _unapply_xheaders(self) -> None: |
| 367 | """Undo changes from `_apply_xheaders`. |
no test coverage detected