fix: IP校验改用endpoint级HTTPException替代Pydantic validator

model_validator/field_validator在自定义RequestValidationError handler下
无法正确序列化为JSON,导致500。改用endpoint级_validate_ip_list()辅助函数
抛HTTPException(400),避免Pydantic→FastAPI序列化链路问题。
This commit is contained in:
Your Name
2026-05-30 22:48:41 +08:00
parent d19f5af807
commit 64503c0260
+14 -23
View File
@@ -14,7 +14,7 @@ import socket
import bcrypt
from fastapi import APIRouter, Depends, HTTPException, Request
from pydantic import BaseModel, Field, field_validator
from pydantic import BaseModel, Field
from server.api.dependencies import get_db
from server.api.auth_jwt import get_current_admin
@@ -93,6 +93,16 @@ async def _verify_reauth(db: AsyncSession, admin: Admin, password: str) -> None:
raise HTTPException(status_code=400, detail="当前密码错误")
def _validate_ip_list(ips: list[str]) -> None:
"""Validate IP/CIDR/domain list. Raises 400 on invalid entry."""
for raw in ips:
ip = raw.strip()
if not ip:
continue
if not (_IP_RE.match(ip) or _DOMAIN_RE.match(ip)):
raise HTTPException(status_code=400, detail=f"无效的 IP/CIDR/域名: {ip}")
# ── System Settings ──
@router.get("/", response_model=list)
@@ -841,17 +851,6 @@ _DOMAIN_RE = re.compile(r"^[a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?(\.[a-zA-
class IpAllowlistRequest(BaseModel):
ips: list[str] = Field(..., min_length=1, max_length=100, description="IP or hostname list")
@field_validator("ips", mode="before")
@classmethod
def validate_ips(cls, v: list[str]) -> list[str]:
for raw in v:
ip = raw.strip()
if not ip:
continue
if not (_IP_RE.match(ip) or _DOMAIN_RE.match(ip)):
raise ValueError(f"无效的 IP/CIDR/域名: {ip}")
return v
class SubscriptionParseRequest(BaseModel):
url: str = Field(..., min_length=1, description="Proxy subscription URL to fetch and parse")
@@ -939,17 +938,6 @@ class IpAllowlistSaveRequest(BaseModel):
subscription_url: Optional[str] = None # None = don't change; "" = clear
enabled: Optional[bool] = None # None = don't change
@field_validator("manual_ips", mode="before")
@classmethod
def validate_manual_ips(cls, v: list[str]) -> list[str]:
for raw in v:
ip = raw.strip()
if not ip:
continue
if not (_IP_RE.match(ip) or _DOMAIN_RE.match(ip)):
raise ValueError(f"无效的 IP/CIDR/域名: {ip}")
return v
@router.post("/ip-allowlist", response_model=dict)
async def set_ip_allowlist(
@@ -960,6 +948,8 @@ async def set_ip_allowlist(
):
"""Save manual IPs, subscription URL, and/or enabled toggle."""
from server.config import settings as _settings
if payload.manual_ips:
_validate_ip_list(payload.manual_ips)
repo = SettingRepositoryImpl(db)
@@ -1046,6 +1036,7 @@ async def add_manual_ips(
):
"""Append IPs to the manual list and trigger allowlist rebuild."""
from server.config import settings as _settings
_validate_ip_list(payload.ips)
existing = [ip.strip() for ip in (_settings.LOGIN_MANUAL_IPS or "").split(",") if ip.strip()]
new_ips = [ip.strip() for ip in payload.ips if ip.strip()]