106 lines
3.6 KiB
Python
106 lines
3.6 KiB
Python
"""API Key 认证依赖
|
||
|
||
用于保护写操作（POST/PUT/DELETE）。
|
||
- 来自可信内网（Tailscale 100.64.0.0/10、本机回环）的请求直接放行；
|
||
- 其余来源：API_KEY 已配置则校验 X-API-Key，未配置则拒绝（防止外部裸奔）。
|
||
"""
|
||
|
||
import hmac
|
||
import ipaddress
|
||
from typing import Optional
|
||
|
||
from fastapi import Header, HTTPException, Request, status
|
||
|
||
from app.core.config import settings
|
||
|
||
# 可信来源网段：Tailscale 使用 CGNAT 100.64.0.0/10 分配内网 IP（如 100.89.x.x）；
|
||
# 回环地址覆盖 tailscale serve 代理转发与本地开发场景。
|
||
TRUSTED_NETWORKS = [
|
||
ipaddress.ip_network("100.64.0.0/10"),
|
||
ipaddress.ip_network("127.0.0.0/8"),
|
||
ipaddress.ip_network("::1/128"),
|
||
]
|
||
|
||
|
||
def _key_matches(provided: Optional[str], expected: str) -> bool:
|
||
"""常量时间比较密钥，避免时序旁路泄露密钥长度/前缀信息"""
|
||
if not provided:
|
||
return False
|
||
return hmac.compare_digest(provided.encode(), expected.encode())
|
||
|
||
|
||
def _client_ip(request: Request) -> str:
|
||
"""获取客户端真实 IP
|
||
|
||
tailscale serve 转发到本机时 request.client 为回环地址，真实来源在
|
||
X-Forwarded-For 头中；仅信任来自回环的 XFF，防止外部直连时伪造
|
||
X-Forwarded-For 绕过鉴权。
|
||
"""
|
||
host = request.client.host if request.client else ""
|
||
if host in ("127.0.0.1", "::1"):
|
||
xff = request.headers.get("x-forwarded-for")
|
||
if xff:
|
||
return xff.split(",")[0].strip()
|
||
return host
|
||
|
||
|
||
def _is_trusted(ip: str) -> bool:
|
||
"""判断来源 IP 是否属于可信内网（Tailscale 网段 / 本机回环）"""
|
||
if not ip:
|
||
return False
|
||
try:
|
||
addr = ipaddress.ip_address(ip)
|
||
except ValueError:
|
||
return False
|
||
return any(addr in net for net in TRUSTED_NETWORKS)
|
||
|
||
|
||
def _is_funnel_request(request: Request) -> bool:
|
||
"""判断是否 tailscale funnel 公网流量
|
||
|
||
tailscale serve 转发时会把客户端伪造的 Tailscale-* 头清空后按真实
|
||
情况设置：tailnet 内无此头，funnel 公网流量为 "?1"（不可伪造）。
|
||
"""
|
||
return request.headers.get("tailscale-funnel-request") == "?1"
|
||
|
||
|
||
async def _require_key(x_api_key: Optional[str]) -> None:
|
||
"""按外部来源校验 API Key（可信内网已在调用方排除）"""
|
||
if not settings.API_KEY:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="外部访问需要 API Key，请先在 .env 配置 API_KEY 并设置到前端",
|
||
)
|
||
if not _key_matches(x_api_key, settings.API_KEY):
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="API Key 无效或缺失",
|
||
)
|
||
|
||
|
||
async def require_api_key(
|
||
request: Request,
|
||
x_api_key: Optional[str] = Header(default=None, alias="X-API-Key"),
|
||
) -> None:
|
||
"""校验 API Key：Tailscale 内网/本机放行，外部来源必须携带有效 Key"""
|
||
# funnel 公网流量：即使来源 IP 恰在 tailnet 网段也强制校验（双保险）
|
||
if _is_funnel_request(request):
|
||
await _require_key(x_api_key)
|
||
return
|
||
if _is_trusted(_client_ip(request)):
|
||
return
|
||
await _require_key(x_api_key)
|
||
|
||
|
||
async def require_agent_key(
|
||
x_agent_key: Optional[str] = Header(default=None, alias="X-Agent-Key"),
|
||
) -> None:
|
||
"""校验 Agent 上报 Key（可选启用）"""
|
||
if not settings.AGENT_KEY:
|
||
return
|
||
if not _key_matches(x_agent_key, settings.AGENT_KEY):
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="Agent Key 无效或缺失",
|
||
)
|