# SPDX-License-Identifier: AGPL-3.0-or-later """Implementation of a middleware to determine the real IP of an HTTP request (:py:obj:`flask.request.remote_addr`) behind a proxy chain.""" # pylint: disable=too-many-branches import typing as t from collections import abc from ipaddress import IPv4Address, IPv6Address, ip_address, ip_network, IPv4Network, IPv6Network from werkzeug.http import parse_list_header from . import config from ._helpers import log_error_only_once, logger if t.TYPE_CHECKING: from _typeshed.wsgi import StartResponse from _typeshed.wsgi import WSGIApplication from _typeshed.wsgi import WSGIEnvironment @t.final class ProxyFix: """A middleware like the ProxyFix_ class, where the ``x_for`` argument is replaced by a method that determines the number of trusted proxies via the ``botdetection.trusted_proxies`` setting. .. sidebar:: :py:obj:`flask.Request.remote_addr` SearXNG uses Werkzeug's ProxyFix_ (with it default ``x_for=1``). The remote IP (:py:obj:`flask.Request.remote_addr`) of the request is taken from (first match): - X-Forwarded-For_: If the header is set, the first untrusted IP that comes before the IPs that are still part of the ``botdetection.trusted_proxies`` is used. - `X-Real-IP `__: If X-Forwarded-For_ is not set, `X-Real-IP` is used (``botdetection.trusted_proxies`` is ignored). If none of the header is set, the REMOTE_ADDR_ from the WSGI layer is used. If (for whatever reasons) none IP can be determined, an error message is displayed and ``100::`` is used instead (:rfc:`6666`). .. _ProxyFix: https://werkzeug.palletsprojects.com/middleware/proxy_fix/ .. _X-Forwarded-For: https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/X-Forwarded-For .. _REMOTE_ADDR: https://wsgi.readthedocs.io/en/latest/proposals-2.0.html#making-some-keys-required """ def __init__(self, wsgi_app: "WSGIApplication") -> None: self.wsgi_app = wsgi_app def trusted_proxies(self) -> list[IPv4Network | IPv6Network]: cfg = config.get_global_cfg() proxy_list: list[str] = cfg.get("botdetection.trusted_proxies", default=[]) return [ip_network(net, strict=False) for net in proxy_list] def trusted_remote_addr( self, x_forwarded_for: list[IPv4Address | IPv6Address], trusted_proxies: list[IPv4Network | IPv6Network], ) -> str: # always rtl for addr in reversed(x_forwarded_for): trust: bool = False for net in trusted_proxies: if addr.version == net.version and addr in net: logger.debug("trust proxy %s (member of %s)", addr, net) trust = True break # client address if not trust: return addr.compressed # fallback to first address return x_forwarded_for[0].compressed def __call__(self, environ: "WSGIEnvironment", start_response: "StartResponse") -> abc.Iterable[bytes]: # pylint: disable=too-many-statements trusted_proxies = self.trusted_proxies() # We do not rely on the REMOTE_ADDR from the WSGI environment / the # variable is first removed from the WSGI environment and explicitly set # in this function! orig_remote_addr: str | None = environ.pop("REMOTE_ADDR") # Validate the IPs involved in this game and delete all invalid ones # from the WSGI environment. if orig_remote_addr: try: addr = ip_address(orig_remote_addr) if addr.version == 6 and addr.ipv4_mapped: addr = addr.ipv4_mapped orig_remote_addr = addr.compressed except ValueError as exc: logger.error("REMOTE_ADDR: %s / discard REMOTE_ADDR from WSGI environment", exc) orig_remote_addr = None x_real_ip: str | None = environ.get("HTTP_X_REAL_IP") if x_real_ip: try: addr = ip_address(x_real_ip) if addr.version == 6 and addr.ipv4_mapped: addr = addr.ipv4_mapped x_real_ip = addr.compressed except ValueError as exc: logger.error("X-Real-IP: %s / discard HTTP_X_REAL_IP from WSGI environment", exc) environ.pop("HTTP_X_REAL_IP") x_real_ip = None x_forwarded_for: list[IPv4Address | IPv6Address] = [] if environ.get("HTTP_X_FORWARDED_FOR"): for x_for_ip in parse_list_header(str(environ.get("HTTP_X_FORWARDED_FOR"))): try: addr = ip_address(x_for_ip) except ValueError as exc: logger.error("X-Forwarded-For: %s / discard HTTP_X_FORWARDED_FOR from WSGI environment", exc) environ.pop("HTTP_X_FORWARDED_FOR") x_forwarded_for = [] break if addr.version == 6 and addr.ipv4_mapped: addr = addr.ipv4_mapped x_forwarded_for.append(addr) # log questionable WSGI environments if not x_forwarded_for and not x_real_ip: log_error_only_once("X-Forwarded-For nor X-Real-IP header is set!") if x_forwarded_for and not trusted_proxies: log_error_only_once("missing botdetection.trusted_proxies config") # without trusted_proxies, this variable is useless for determining # the real IP x_forwarded_for = [] # securing the WSGI environment variables that are adjusted environ.update({"botdetection.trusted_proxies.orig": {"REMOTE_ADDR": orig_remote_addr}}) # determine *the real IP* if x_forwarded_for: environ["REMOTE_ADDR"] = self.trusted_remote_addr(x_forwarded_for, trusted_proxies) elif x_real_ip: environ["REMOTE_ADDR"] = x_real_ip elif orig_remote_addr: environ["REMOTE_ADDR"] = orig_remote_addr else: logger.error("No remote IP could be determined, use black-hole address: 100::") environ["REMOTE_ADDR"] = "100::" try: _ = ip_address(environ["REMOTE_ADDR"]) except ValueError as exc: logger.error("REMOTE_ADDR: %s, use black-hole address: 100::", exc) environ["REMOTE_ADDR"] = "100::" logger.debug("final REMOTE_ADDR is: %s", environ["REMOTE_ADDR"]) return self.wsgi_app(environ, start_response)