xsl
5b2b73f5d5
保留5号历史三档实现，新增中下庭高度定位、扶正裁切、矩形叠加与30案例对比页。 worker仅允许本机和配置白名单访问；网关不增加IP限制。空图片URL字段改写限定为52号，保持旧接口兼容。同步接口、裁切与部署文档。 验证：113项测试通过；1项历史数值基线失败，在未修改的a3fddbb版本复现。浏览器验证5/52及30案例对比页通过。
78 lines
3.2 KiB
Python
78 lines
3.2 KiB
Python
"""Direct-peer IP access policy for the GPU worker's dynamic endpoints."""
|
|
import ipaddress
|
|
import json
|
|
from pathlib import Path
|
|
import socket
|
|
|
|
CONFIG_PATH = Path(__file__).resolve().parent / "ip_access.json"
|
|
|
|
|
|
def _address(value):
|
|
address = ipaddress.ip_address(value)
|
|
return address.ipv4_mapped if isinstance(address, ipaddress.IPv6Address) and address.ipv4_mapped else address
|
|
|
|
|
|
def local_addresses():
|
|
addresses = {_address("127.0.0.1"), _address("::1")}
|
|
try:
|
|
addresses.update(_address(item[4][0].split("%", 1)[0])
|
|
for item in socket.getaddrinfo(socket.gethostname(), None))
|
|
except (OSError, ValueError):
|
|
pass
|
|
# Include all assigned Linux interface addresses, not entire private ranges.
|
|
try:
|
|
import fcntl
|
|
import struct
|
|
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
|
|
for _, name in socket.if_nameindex():
|
|
try:
|
|
data = fcntl.ioctl(sock.fileno(), 0x8915, struct.pack("256s", name.encode()[:15]))
|
|
addresses.add(_address(socket.inet_ntoa(data[20:24])))
|
|
except OSError:
|
|
continue
|
|
path = Path("/proc/net/if_inet6")
|
|
if path.exists():
|
|
addresses.update(ipaddress.IPv6Address(line.split()[0]) for line in path.read_text().splitlines())
|
|
except (ImportError, OSError, ValueError):
|
|
pass
|
|
return addresses
|
|
|
|
|
|
def _networks(values):
|
|
if not isinstance(values, list) or not all(isinstance(value, str) for value in values):
|
|
raise ValueError("IP配置必须是字符串列表")
|
|
return [ipaddress.ip_network(value, strict=False) for value in values]
|
|
|
|
|
|
def _matches(address, networks):
|
|
return any(address.version == network.version and address in network for network in networks)
|
|
|
|
|
|
def check_access(request):
|
|
"""Return (allowed, reason), re-reading the whitelist for each request."""
|
|
try:
|
|
config = json.loads(CONFIG_PATH.read_text()) if CONFIG_PATH.exists() else {}
|
|
allowed = _networks(config.get("allowed_ips", []))
|
|
if request.client is None or request.client.port == 0:
|
|
# Uvicorn proxy middleware has discarded the socket peer at port 0.
|
|
# Reject rather than treating an unverified forwarded IP as local.
|
|
return False, "无法确认直连来源，请使用 --no-proxy-headers 启动服务"
|
|
peer = _address(request.client.host)
|
|
# Authorize the gateway's socket IP, never its end-user forwarding headers.
|
|
if peer.is_unspecified:
|
|
return False, "来源地址无效"
|
|
return (True, "") if peer.is_loopback or peer in local_addresses() or _matches(peer, allowed) else (False, "来源IP不在接口白名单")
|
|
except (OSError, ValueError, TypeError, AttributeError):
|
|
return False, "IP白名单配置或来源地址无效"
|
|
|
|
|
|
def rejection(request):
|
|
from fastapi.responses import JSONResponse
|
|
if request.url.path == "/static" or request.url.path.startswith("/static/"):
|
|
return None
|
|
allowed, reason = check_access(request)
|
|
if allowed:
|
|
return None
|
|
return JSONResponse(status_code=403, content={"code": 1010, "message": reason,
|
|
"request_id": "ip-access", "data": None})
|