import ipaddress from django.conf import settings FORWARDED_META_KEYS = ("HTTP_X_FORWARDED_FOR", "HTTP_X_FORWARDED_PROTO", "HTTP_X_FORWARDED_HOST", "HTTP_X_REAL_IP") def _parse_ip(value): try: return ipaddress.ip_address(value.strip()) except ValueError: return None def is_trusted_proxy(ip): return ip is not None and any(ip in net for net in settings.TRUSTED_PROXIES) def resolve_client_ip(peer, meta): """Return the real client IP for a request that arrived from a trusted proxy. With CLIENT_IP_HEADER set (e.g. CF-Connecting-IP) that header wins. Otherwise X-Forwarded-For is walked from the right, skipping trusted proxies; the first untrusted address is the client. Anything left of it is client-controlled. """ if settings.CLIENT_IP_HEADER: key = "HTTP_" + settings.CLIENT_IP_HEADER.upper().replace("-", "_") ip = _parse_ip(meta.get(key, "")) if ip is not None: return ip hops = [h for h in meta.get("HTTP_X_FORWARDED_FOR", "").split(",") if h.strip()] for hop in reversed(hops): ip = _parse_ip(hop) if ip is None: break if not is_trusted_proxy(ip): return ip return peer class ProxyMiddleware: """Honour X-Forwarded-* only when the TCP peer is a trusted proxy. Requests from anyone else get those headers stripped, so is_secure() and REMOTE_ADDR cannot be spoofed. request.META["REMOTE_ADDR"] holds the real client IP afterwards; the original peer is kept in "PROXY_PEER_ADDR". """ def __init__(self, get_response): self.get_response = get_response def __call__(self, request): meta = request.META peer = _parse_ip(meta.get("REMOTE_ADDR", "")) if is_trusted_proxy(peer): client = resolve_client_ip(peer, meta) meta["PROXY_PEER_ADDR"] = str(peer) meta["REMOTE_ADDR"] = str(client) else: for key in FORWARDED_META_KEYS: meta.pop(key, None) if settings.CLIENT_IP_HEADER: meta.pop("HTTP_" + settings.CLIENT_IP_HEADER.upper().replace("-", "_"), None) return self.get_response(request) class SecurityHeadersMiddleware: """Adds the headers Django does not set itself (CSP, Permissions-Policy).""" def __init__(self, get_response): self.get_response = get_response def __call__(self, request): response = self.get_response(request) response.headers.setdefault("Content-Security-Policy", settings.CONTENT_SECURITY_POLICY) response.headers.setdefault("Permissions-Policy", settings.PERMISSIONS_POLICY) response.headers.setdefault("Cross-Origin-Resource-Policy", "same-origin") user = getattr(request, "user", None) if user is not None and user.is_authenticated: # Pages with private content must not end up in shared caches. response.headers.setdefault("Cache-Control", "private, no-store") return response