91 lines
2.6 KiB
Python
91 lines
2.6 KiB
Python
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]:
|
|
parsed: list[ipaddress._BaseAddress] = []
|
|
for item in header_value.split(","):
|
|
ip = _parse_ip(item.strip())
|
|
if ip is not None:
|
|
parsed.append(ip)
|
|
return parsed
|
|
|
|
|
|
def _trusted_forwarded_ip(
|
|
header_value: str,
|
|
trusted_networks: Sequence[ipaddress._BaseNetwork],
|
|
) -> Optional[str]:
|
|
forwarded_ips = _forwarded_ips(header_value)
|
|
if not forwarded_ips:
|
|
return None
|
|
|
|
for ip in reversed(forwarded_ips):
|
|
if not any(ip in network for network in trusted_networks):
|
|
return str(ip)
|
|
return str(forwarded_ips[0])
|
|
|
|
|
|
def request_client_ip(
|
|
request: web.Request,
|
|
*,
|
|
trusted_proxies: Optional[Sequence[str] | str] = None,
|
|
) -> Optional[str]:
|
|
remote_ip = _parse_ip(request.remote or "")
|
|
forwarded_for = request.headers.get("X-Forwarded-For", "")
|
|
|
|
if remote_ip and forwarded_for:
|
|
trusted_networks = parse_ip_entries(trusted_proxies)
|
|
if any(remote_ip in network for network in trusted_networks):
|
|
forwarded_ip = _trusted_forwarded_ip(forwarded_for, trusted_networks)
|
|
if forwarded_ip:
|
|
return forwarded_ip
|
|
|
|
if remote_ip:
|
|
return str(remote_ip)
|
|
|
|
forwarded_ips = _forwarded_ips(forwarded_for)
|
|
return str(forwarded_ips[-1]) if forwarded_ips else 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)
|