From 2fbb658126c10ea489388edb1a6d1d0d4ae11b2f Mon Sep 17 00:00:00 2001
From: gouki <gouki.xiao@gmail.com>
Date: Sat, 15 Aug 2026 22:24:47 +0000
Subject: [PATCH] =?UTF-8?q?fix(security):=20funnel=E5=85=AC=E7=BD=91?=
 =?UTF-8?q?=E6=B5=81=E9=87=8F=E5=BC=BA=E5=88=B6=E6=A0=A1=E9=AA=8CAPI=5FKEY?=
 =?UTF-8?q?=E5=8F=8C=E4=BF=9D=E9=99=A9(Tailscale-Funnel-Request=E5=A4=B4,?=
 =?UTF-8?q?=E4=B8=8D=E5=8F=AF=E4=BC=AA=E9=80=A0)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/core/security.py | 32 +++++++++++++++++++++++++-------
 1 file changed, 25 insertions(+), 7 deletions(-)

diff --git a/app/core/security.py b/app/core/security.py
index 2630190..8d294cc 100644
--- a/app/core/security.py
+++ b/app/core/security.py
@@ -55,13 +55,17 @@ def _is_trusted(ip: str) -> bool:
     return any(addr in net for net in TRUSTED_NETWORKS)
 
 
-async def require_api_key(
-    request: Request,
-    x_api_key: Optional[str] = Header(default=None, alias="X-API-Key"),
-) -> None:
-    """校验 API Key：Tailscale 内网/本机放行，外部来源必须携带有效 Key"""
-    if _is_trusted(_client_ip(request)):
-        return
+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,
@@ -74,6 +78,20 @@ async def require_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: