Trust reverse proxy client IP headers

karldns/security.py

@@ -15,6 +15,19 @@ from .settings import DdnsSettings
 DASHBOARD_UNLOCK_COOKIE = "ddns_dashboard_unlock"
 DASHBOARD_UNLOCK_DEFAULT_SECONDS = 24 * 60 * 60
 DASHBOARD_UNLOCK_STAY_SIGNED_IN_SECONDS = 30 * 24 * 60 * 60
+PRIVATE_PROXY_NETWORKS = tuple(
+    ip_network(network)
+    for network in (
+        "10.0.0.0/8",
+        "172.16.0.0/12",
+        "192.168.0.0/16",
+        "127.0.0.0/8",
+        "169.254.0.0/16",
+        "::1/128",
+        "fc00::/7",
+        "fe80::/10",
+    )
+)
 
 
 def credentials_from_basic_auth(header: str) -> tuple[str | None, str | None]:
@@ -146,11 +159,10 @@ def require_admin_csrf(settings: DdnsSettings, supplied: str) -> None:
 
 def client_ip(request: Request, settings: DdnsSettings | None = None) -> str:
     peer = request.client.host if request.client else "unknown"
-    if settings and _trusted_proxy(peer, settings.trusted_proxy_ips):
-        for header_name in ("cf-connecting-ip", "x-forwarded-for", "x-real-ip"):
-            candidate = request.headers.get(header_name, "")
-            if candidate:
-                return candidate.split(",", 1)[0].strip()
+    if _should_trust_proxy_headers(peer, settings):
+        forwarded_ip = _client_ip_from_headers(request)
+        if forwarded_ip:
+            return forwarded_ip
     return peer
 
 
@@ -181,3 +193,64 @@ def _trusted_proxy(peer: str, trusted: set[str]) -> bool:
         except ValueError:
             continue
     return False
+
+
+def _should_trust_proxy_headers(peer: str, settings: DdnsSettings | None) -> bool:
+    if settings and _trusted_proxy(peer, settings.trusted_proxy_ips):
+        return True
+    if settings and settings.trust_private_proxy_headers:
+        return _private_proxy_peer(peer)
+    return False
+
+
+def _private_proxy_peer(peer: str) -> bool:
+    try:
+        peer_ip = ip_address(peer)
+    except ValueError:
+        return False
+    return any(peer_ip in network for network in PRIVATE_PROXY_NETWORKS)
+
+
+def _client_ip_from_headers(request: Request) -> str | None:
+    for header_name in ("cf-connecting-ip", "true-client-ip", "x-real-ip"):
+        candidate = _valid_header_ip(request.headers.get(header_name, ""))
+        if candidate:
+            return candidate
+    for candidate in request.headers.get("x-forwarded-for", "").split(","):
+        parsed = _valid_header_ip(candidate)
+        if parsed:
+            return parsed
+    return _client_ip_from_forwarded(request.headers.get("forwarded", ""))
+
+
+def _client_ip_from_forwarded(value: str) -> str | None:
+    for forwarded_pair in value.split(";"):
+        key, separator, candidate = forwarded_pair.strip().partition("=")
+        if separator and key.lower() == "for":
+            parsed = _valid_header_ip(candidate)
+            if parsed:
+                return parsed
+    return None
+
+
+def _valid_header_ip(value: str) -> str | None:
+    cleaned = _strip_header_ip(value)
+    if not cleaned:
+        return None
+    try:
+        return str(ip_address(cleaned))
+    except ValueError:
+        return None
+
+
+def _strip_header_ip(value: str) -> str:
+    cleaned = value.strip().strip('"').strip("'")
+    if not cleaned or cleaned.lower() == "unknown":
+        return ""
+    if cleaned.startswith("[") and "]" in cleaned:
+        return cleaned[1 : cleaned.index("]")]
+    if cleaned.count(":") == 1:
+        host, port = cleaned.rsplit(":", 1)
+        if port.isdigit():
+            return host
+    return cleaned