Codex Release Bot
69 lines
2.1 KiB
Python
69 lines
2.1 KiB
Python
"""Resolve real client IP behind reverse proxy (1Panel / Nginx / Docker)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ipaddress
|
|
|
|
from starlette.requests import Request
|
|
|
|
# Upstream hop is a private/loopback/docker address → trust forwarded headers.
|
|
_TRUSTED_PROXY_NETWORKS: tuple[ipaddress.IPv4Network | ipaddress.IPv6Network, ...] = (
|
|
ipaddress.ip_network("127.0.0.0/8"),
|
|
ipaddress.ip_network("10.0.0.0/8"),
|
|
ipaddress.ip_network("172.16.0.0/12"),
|
|
ipaddress.ip_network("192.168.0.0/16"),
|
|
ipaddress.ip_network("::1/128"),
|
|
ipaddress.ip_network("fc00::/7"),
|
|
)
|
|
|
|
|
|
def _parse_ip(value: str) -> str | None:
|
|
text = (value or "").strip()
|
|
if not text:
|
|
return None
|
|
# X-Real-IP may rarely contain comma-separated values
|
|
candidate = text.split(",")[0].strip()
|
|
try:
|
|
ipaddress.ip_address(candidate)
|
|
except ValueError:
|
|
return None
|
|
return candidate
|
|
|
|
|
|
def _is_trusted_proxy_hop(ip: str) -> bool:
|
|
try:
|
|
addr = ipaddress.ip_address(ip)
|
|
except ValueError:
|
|
return False
|
|
return any(addr in net for net in _TRUSTED_PROXY_NETWORKS)
|
|
|
|
|
|
def _ip_from_forwarded_headers(request: Request) -> str | None:
|
|
"""X-Real-IP first; X-Forwarded-For last (Nginx appends $remote_addr)."""
|
|
x_real = _parse_ip(request.headers.get("X-Real-IP", ""))
|
|
if x_real:
|
|
return x_real
|
|
|
|
xff = (request.headers.get("X-Forwarded-For") or "").strip()
|
|
if not xff:
|
|
return None
|
|
parts = [p.strip() for p in xff.split(",") if p.strip()]
|
|
if not parts:
|
|
return None
|
|
return _parse_ip(parts[-1])
|
|
|
|
|
|
def get_client_ip(request: Request) -> str:
|
|
"""Return client IP for auth, audit, and rate limits.
|
|
|
|
When the TCP peer is a trusted reverse proxy (Docker bridge, loopback, LAN),
|
|
use X-Real-IP / X-Forwarded-For. Direct public connections ignore those
|
|
headers to prevent spoofing.
|
|
"""
|
|
direct = request.client.host if request.client else ""
|
|
if direct and _is_trusted_proxy_hop(direct):
|
|
forwarded = _ip_from_forwarded_headers(request)
|
|
if forwarded:
|
|
return forwarded
|
|
return direct or "unknown"
|