109 lines
2.6 KiB
Python
109 lines
2.6 KiB
Python
"""
|
||
GeoIP dependency for FastAPI
|
||
"""
|
||
|
||
from functools import lru_cache
|
||
import ipaddress
|
||
from typing import Annotated
|
||
|
||
from app.config import settings
|
||
from app.helpers.geoip_helper import GeoIPHelper
|
||
|
||
from fastapi import Depends, Request
|
||
|
||
|
||
@lru_cache
|
||
def get_geoip_helper() -> GeoIPHelper:
|
||
"""
|
||
获取 GeoIP 帮助类实例
|
||
使用 lru_cache 确保单例模式
|
||
"""
|
||
return GeoIPHelper(
|
||
dest_dir=settings.geoip_dest_dir,
|
||
license_key=settings.maxmind_license_key,
|
||
editions=["City", "ASN"],
|
||
max_age_days=8,
|
||
timeout=60.0,
|
||
)
|
||
|
||
|
||
def get_client_ip(request: Request) -> str:
|
||
"""
|
||
获取客户端真实 IP 地址
|
||
支持 IPv4 和 IPv6，考虑代理、负载均衡器等情况
|
||
"""
|
||
headers = request.headers
|
||
|
||
# 1. Cloudflare 专用头部
|
||
cf_ip = headers.get("CF-Connecting-IP")
|
||
if cf_ip:
|
||
ip = cf_ip.strip()
|
||
if is_valid_ip(ip):
|
||
return ip
|
||
|
||
true_client_ip = headers.get("True-Client-IP")
|
||
if true_client_ip:
|
||
ip = true_client_ip.strip()
|
||
if is_valid_ip(ip):
|
||
return ip
|
||
|
||
# 2. 标准代理头部
|
||
forwarded_for = headers.get("X-Forwarded-For")
|
||
if forwarded_for:
|
||
# X-Forwarded-For 可能包含多个 IP，取第一个有效的
|
||
for ip_str in forwarded_for.split(","):
|
||
ip = ip_str.strip()
|
||
if is_valid_ip(ip) and not is_private_ip(ip):
|
||
return ip
|
||
|
||
real_ip = headers.get("X-Real-IP")
|
||
if real_ip:
|
||
ip = real_ip.strip()
|
||
if is_valid_ip(ip):
|
||
return ip
|
||
|
||
# 3. 回退到客户端 IP
|
||
client_ip = request.client.host if request.client else "127.0.0.1"
|
||
return client_ip if is_valid_ip(client_ip) else "127.0.0.1"
|
||
|
||
|
||
IPAddress = Annotated[str, Depends(get_client_ip)]
|
||
GeoIPService = Annotated[GeoIPHelper, Depends(get_geoip_helper)]
|
||
|
||
|
||
def is_valid_ip(ip_str: str) -> bool:
|
||
"""
|
||
验证 IP 地址是否有效（支持 IPv4 和 IPv6）
|
||
"""
|
||
try:
|
||
ipaddress.ip_address(ip_str)
|
||
return True
|
||
except ValueError:
|
||
return False
|
||
|
||
|
||
def is_private_ip(ip_str: str) -> bool:
|
||
"""
|
||
判断是否为私有 IP 地址
|
||
"""
|
||
try:
|
||
ip = ipaddress.ip_address(ip_str)
|
||
return ip.is_private
|
||
except ValueError:
|
||
return False
|
||
|
||
|
||
def normalize_ip(ip_str: str) -> str:
|
||
"""
|
||
标准化 IP 地址格式
|
||
对于 IPv6，转换为压缩格式
|
||
"""
|
||
try:
|
||
ip = ipaddress.ip_address(ip_str)
|
||
if isinstance(ip, ipaddress.IPv6Address):
|
||
return ip.compressed
|
||
else:
|
||
return str(ip)
|
||
except ValueError:
|
||
return ip_str
|