vps-manager/app/core/security.py

"""API Key 认证依赖

用于保护写操作（POST/PUT/DELETE）与 Agent 上报。
- 来自可信内网（Tailscale 100.64.0.0/10、本机回环）的请求直接放行；
- 其余来源：密钥已配置则校验对应请求头，未配置则拒绝（防止外部裸奔）。
"""

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]) -> None:
    """按外部来源校验 Agent Key（可信内网已在调用方排除）"""
    if not settings.AGENT_KEY:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=(
                "Agent 上报来自外部网络，需先在 .env 配置 AGENT_KEY"
                "并写入各 VPS 的 /etc/vps-agent.env（VPS_AGENT_KEY）"
            ),
        )
    if not _key_matches(x_agent_key, settings.AGENT_KEY):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Agent Key 无效或缺失",
        )


async def require_agent_key(
    request: Request,
    x_agent_key: Optional[str] = Header(default=None, alias="X-Agent-Key"),
) -> None:
    """校验 Agent 上报 Key：tailnet/本机放行，外部来源必须携带有效 Key

    与 require_api_key 用同一套可信来源判定。此前未配 AGENT_KEY 时对任意来源全放行，
    而服务默认监听 0.0.0.0——公网上任何人都能伪造监控指标与安全检查结果上报。
    改成「外部来源必须有 key」后，走 tailnet IP 上报的 Agent 不受影响。
    """
    if _is_funnel_request(request):
        await _require_agent_key(x_agent_key)
        return
    if _is_trusted(_client_ip(request)):
        return
    await _require_agent_key(x_agent_key)