from __future__ import annotations import ipaddress from typing import Optional, Sequence from aiohttp import web def parse_ip_entries(raw_values: Optional[Sequence[str] | str]) -> list[ipaddress._BaseNetwork]: if raw_values is None: return [] if isinstance(raw_values, str): values = [item.strip() for item in raw_values.split(",")] else: values = [str(item).strip() for item in raw_values] parsed: list[ipaddress._BaseNetwork] = [] for value in values: if not value: continue try: parsed.append(ipaddress.ip_network(value, strict=False)) except ValueError: continue return parsed def _parse_ip(value: Optional[str]) -> Optional[ipaddress._BaseAddress]: if not value: return None try: return ipaddress.ip_address(value.strip()) except ValueError: return None def _forwarded_ips(header_value: str) -> list[ipaddress._BaseAddress]: candidates = [item.strip() for item in header_value.split(",") if item.strip()] parsed: list[ipaddress._BaseAddress] = [] for candidate in candidates: parsed_ip = _parse_ip(candidate) if parsed_ip is not None: parsed.append(parsed_ip) return parsed def _forwarded_client_ip( forwarded_ips: Sequence[ipaddress._BaseAddress], trusted_networks: Sequence[ipaddress._BaseNetwork], ) -> Optional[str]: for forwarded_ip in reversed(forwarded_ips): if not any(forwarded_ip in network for network in trusted_networks): return str(forwarded_ip) if forwarded_ips: return str(forwarded_ips[0]) return None def request_client_ip( request: web.Request, *, trusted_proxies: Optional[Sequence[str] | str] = None, ) -> Optional[str]: remote_ip = _parse_ip(request.remote or "") forwarded_ips = _forwarded_ips(request.headers.get("X-Forwarded-For", "")) if remote_ip and forwarded_ips: trusted_networks = parse_ip_entries(trusted_proxies) if any(remote_ip in network for network in trusted_networks): forwarded_ip = _forwarded_client_ip(forwarded_ips, trusted_networks) if forwarded_ip: return forwarded_ip if remote_ip: return str(remote_ip) if forwarded_ips: return str(forwarded_ips[-1]) return None def ip_in_allowlist( ip_value: Optional[str], allowed_entries: Optional[Sequence[str] | str] ) -> bool: parsed_ip = _parse_ip(ip_value) if parsed_ip is None: return False allowed_networks = parse_ip_entries(allowed_entries) return any(parsed_ip in network for network in allowed_networks)