103 lines
2.7 KiB
Python
103 lines
2.7 KiB
Python
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)
|