|
@ -1,6 +1,7 @@ |
|
|
from fastapi import APIRouter, Request, Response, HTTPException |
|
|
|
|
|
import ipaddress |
|
|
import ipaddress |
|
|
from typing import Dict, Any |
|
|
from typing import Any |
|
|
|
|
|
|
|
|
|
|
|
from fastapi import APIRouter, HTTPException, Request, Response |
|
|
|
|
|
|
|
|
from app.api.deps import CurrentUser |
|
|
from app.api.deps import CurrentUser |
|
|
|
|
|
|
|
@ -57,11 +58,11 @@ async def get_secure_ip(request: Request, current_user: CurrentUser): |
|
|
return { |
|
|
return { |
|
|
"ip": client_ip, |
|
|
"ip": client_ip, |
|
|
"user_id": str(current_user.id), |
|
|
"user_id": str(current_user.id), |
|
|
"user_email": current_user.email |
|
|
"user_email": current_user.email, |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@router.get("/analyze/{ip_address}", response_model=Dict[str, Any]) |
|
|
@router.get("/analyze/{ip_address}", response_model=dict[str, Any]) |
|
|
async def analyze_ip(ip_address: str): |
|
|
async def analyze_ip(ip_address: str): |
|
|
""" |
|
|
""" |
|
|
分析指定的IP地址 |
|
|
分析指定的IP地址 |
|
@ -84,26 +85,26 @@ async def analyze_ip(ip_address: str): |
|
|
# IPv6特有属性 |
|
|
# IPv6特有属性 |
|
|
if ip.version == 6: |
|
|
if ip.version == 6: |
|
|
ipv6 = ipaddress.IPv6Address(ip_address) |
|
|
ipv6 = ipaddress.IPv6Address(ip_address) |
|
|
result.update({ |
|
|
result.update( |
|
|
|
|
|
{ |
|
|
"is_site_local": ipv6.is_site_local, |
|
|
"is_site_local": ipv6.is_site_local, |
|
|
"ipv4_mapped": ipv6.ipv4_mapped is not None, |
|
|
"ipv4_mapped": ipv6.ipv4_mapped is not None, |
|
|
"teredo": ipv6.teredo is not None, |
|
|
"teredo": ipv6.teredo is not None, |
|
|
"sixtofour": ipv6.sixtofour is not None |
|
|
"sixtofour": ipv6.sixtofour is not None, |
|
|
}) |
|
|
} |
|
|
|
|
|
) |
|
|
|
|
|
|
|
|
# 尝试获取反向DNS查询名称 |
|
|
# 尝试获取反向DNS查询名称 |
|
|
try: |
|
|
try: |
|
|
import socket |
|
|
import socket |
|
|
|
|
|
|
|
|
result["reverse_pointer"] = socket.gethostbyaddr(ip_address)[0] |
|
|
result["reverse_pointer"] = socket.gethostbyaddr(ip_address)[0] |
|
|
except (socket.herror, socket.gaierror): |
|
|
except (socket.herror, socket.gaierror): |
|
|
result["reverse_pointer"] = None |
|
|
result["reverse_pointer"] = None |
|
|
|
|
|
|
|
|
return result |
|
|
return result |
|
|
except ValueError: |
|
|
except ValueError: |
|
|
raise HTTPException( |
|
|
raise HTTPException(status_code=400, detail=f"无效的IP地址: {ip_address}") |
|
|
status_code=400, |
|
|
|
|
|
detail=f"无效的IP地址: {ip_address}" |
|
|
|
|
|
) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _extract_client_ip(request: Request) -> str: |
|
|
def _extract_client_ip(request: Request) -> str: |
|
@ -118,7 +119,7 @@ def _extract_client_ip(request: Request) -> str: |
|
|
"X-Real-IP", |
|
|
"X-Real-IP", |
|
|
"CF-Connecting-IP", # Cloudflare |
|
|
"CF-Connecting-IP", # Cloudflare |
|
|
"True-Client-IP", # Akamai/Cloudflare |
|
|
"True-Client-IP", # Akamai/Cloudflare |
|
|
"X-Client-IP" |
|
|
"X-Client-IP", |
|
|
]: |
|
|
]: |
|
|
if header_value := request.headers.get(header): |
|
|
if header_value := request.headers.get(header): |
|
|
# 如果是X-Forwarded-For,可能包含多个IP,取第一个 |
|
|
# 如果是X-Forwarded-For,可能包含多个IP,取第一个 |
|
|