import ipaddress import logging import os from typing import Optional from fastapi import Request logger = logging.getLogger("minimal-yt.network") def _parse_trusted_proxy_networks( value: str, ) -> tuple[ipaddress._BaseNetwork, ...]: networks: list[ipaddress._BaseNetwork] = [] for raw_value in (value or "").split(","): candidate = raw_value.strip() if not candidate: continue try: networks.append(ipaddress.ip_network(candidate, strict=False)) except ValueError: logger.warning( "Ignoring invalid TRUSTED_PROXY_CIDRS entry: %r", candidate, ) return tuple(networks) TRUSTED_PROXY_NETWORKS = _parse_trusted_proxy_networks( os.getenv("TRUSTED_PROXY_CIDRS", "") ) def _parse_ip(value: str) -> Optional[ipaddress._BaseAddress]: """ Parse a bare IP, [IPv6], [IPv6]:port, or IPv4:port. Hostnames are intentionally not accepted. """ value = str(value or "").strip() if not value: return None if value.startswith("["): closing_bracket = value.find("]") if closing_bracket == -1: return None value = value[1:closing_bracket] elif value.count(":") == 1: host, possible_port = value.rsplit(":", 1) if possible_port.isdigit(): value = host try: return ipaddress.ip_address(value) except ValueError: return None def _is_trusted_proxy(address: ipaddress._BaseAddress) -> bool: return any( address.version == network.version and address in network for network in TRUSTED_PROXY_NETWORKS ) def get_client_ip(request: Request) -> str: """ Return the effective client IP. Forwarded headers are trusted only when the direct TCP peer belongs to TRUSTED_PROXY_CIDRS. X-Forwarded-For is traversed right-to-left while skipping trusted proxy hops. """ peer_value = request.client.host if request.client else "" peer_address = _parse_ip(peer_value) if peer_address is None: return peer_value or "unknown" if not _is_trusted_proxy(peer_address): return str(peer_address) forwarded_for = request.headers.get("x-forwarded-for", "") if forwarded_for: addresses = [ address for raw_value in forwarded_for.split(",") if (address := _parse_ip(raw_value)) is not None ] for address in reversed(addresses): if not _is_trusted_proxy(address): return str(address) real_ip = _parse_ip(request.headers.get("x-real-ip", "")) if real_ip is not None and not _is_trusted_proxy(real_ip): return str(real_ip) return str(peer_address)