340 lines
15 KiB
Python
340 lines
15 KiB
Python
"""Global authentication middleware — fail-closed safety net.
|
|
|
|
Rejects unauthenticated requests to non-public paths with 401. When a
|
|
request passes the cookie check, resolves the JWT payload to a real
|
|
``User`` object and stamps it into both ``request.state.user`` and the
|
|
``deerflow.runtime.user_context`` contextvar so that repository-layer
|
|
owner filtering works automatically via the sentinel pattern.
|
|
|
|
Fine-grained permission checks remain in authz.py decorators.
|
|
"""
|
|
|
|
import hashlib
|
|
import hmac
|
|
import re
|
|
from collections.abc import Callable
|
|
|
|
from fastapi import HTTPException, Request, Response
|
|
from starlette.middleware.base import BaseHTTPMiddleware
|
|
from starlette.responses import JSONResponse
|
|
from starlette.types import ASGIApp
|
|
|
|
from app.gateway.auth.disabled_mode import get_assumed_user_for_disabled_auth, is_auth_disabled
|
|
from app.gateway.auth.errors import AuthErrorCode, AuthErrorResponse
|
|
from app.gateway.authz import _ALL_PERMISSIONS, AuthContext
|
|
from app.gateway.internal_auth import INTERNAL_AUTH_HEADER_NAME, get_internal_user, is_valid_internal_auth_token
|
|
from app.gateway.proxy_path import app_relative_path
|
|
from app.gateway.weknora_embed import (
|
|
WEKNORA_EMBED_COOKIE,
|
|
has_valid_weknora_embed_cookie,
|
|
is_weknora_embed_proxy_path,
|
|
)
|
|
from deerflow.runtime.user_context import reset_current_user, set_current_user
|
|
|
|
# Health probes are deliberately public, including the LangGraph-compatible
|
|
# gateway alias used by some deployments. Keep these as exact paths: health
|
|
# checks must bypass authentication, but similarly named application routes
|
|
# must not inherit that exemption.
|
|
PUBLIC_HEALTH_PATHS: frozenset[str] = frozenset(
|
|
{
|
|
"/health",
|
|
"/api/health",
|
|
"/api/langgraph/health",
|
|
}
|
|
)
|
|
|
|
# Path prefixes that never require authentication.
|
|
#
|
|
# ``/api/public/`` is a namespace reserved for intentionally world-readable,
|
|
# read-only endpoints (e.g. public conversation share snapshots). Only mount
|
|
# safe read-only routes under it — everything there bypasses auth entirely.
|
|
_PUBLIC_PATH_PREFIXES: tuple[str, ...] = (
|
|
"/docs",
|
|
"/redoc",
|
|
"/openapi.json",
|
|
"/api/public/",
|
|
# 开放问答接口:无需登录即可指定模型 + 智能体做问答(见 routers/open_chat.py)。
|
|
# 以 "default" 用户桶运行,只暴露受控的问答能力。
|
|
"/api/open/",
|
|
)
|
|
|
|
# Exact auth paths that are public (login/register/status check).
|
|
# /api/v1/auth/me, /api/v1/auth/change-password etc. are NOT public.
|
|
_PUBLIC_EXACT_PATHS: frozenset[str] = frozenset(
|
|
{
|
|
*PUBLIC_HEALTH_PATHS,
|
|
"/api/v1/auth/login/local",
|
|
"/api/v1/auth/options",
|
|
"/api/v1/auth/login/username",
|
|
"/api/v1/auth/login/token",
|
|
"/api/v1/auth/service/exchange",
|
|
"/api/v1/auth/register",
|
|
"/api/v1/auth/logout",
|
|
"/api/v1/auth/setup-status",
|
|
"/api/v1/auth/initialize",
|
|
# Anonymous vector search over published local Wiki indexes.
|
|
"/api/knowledge/vector-search",
|
|
# Stateless Markdown -> DOCX conversion; required by public report pages.
|
|
"/api/writing/export/docx",
|
|
# 并行多智能体面板专用的「绕过常规登录」取 token 端点（内网部署限制了标准登录时用）。
|
|
"/api/parallel-agents/auth/token",
|
|
}
|
|
)
|
|
|
|
|
|
def _is_public(path: str) -> bool:
|
|
stripped = path.rstrip("/")
|
|
if stripped in _PUBLIC_EXACT_PATHS:
|
|
return True
|
|
return any(path.startswith(prefix) for prefix in _PUBLIC_PATH_PREFIXES)
|
|
|
|
|
|
def _is_external_llmwiki(path: str) -> bool:
|
|
return path.startswith("/api/external/llmwiki/")
|
|
|
|
|
|
def _authorize_external_llmwiki(request: Request) -> JSONResponse | None:
|
|
"""Authenticate the isolated service-to-service namespace before routing."""
|
|
|
|
from deerflow.config import get_app_config
|
|
|
|
config = get_app_config().llmwiki.local_wiki_index.external_api
|
|
supplied = request.headers.get("X-API-Key", "")
|
|
valid = bool(config.enabled and supplied) and any(hmac.compare_digest(supplied, candidate) for candidate in (config.api_key, config.previous_api_key) if candidate)
|
|
if not valid:
|
|
return JSONResponse(status_code=401, content={"detail": "Invalid API key"})
|
|
request.state.external_api_key_fingerprint = hashlib.sha256(supplied.encode("utf-8")).hexdigest()[:8]
|
|
return None
|
|
|
|
|
|
def _is_task_deeplink_external_write(request: Request, path: str) -> bool:
|
|
"""Whether this is the credential-free integration write path."""
|
|
|
|
return request.method.upper() == "PUT" and path.startswith("/api/task-deeplink-details/")
|
|
|
|
|
|
# 力量态势的任务详情由既有定时技能直接读写。该类技能无法及时统一
|
|
# 迁移到新登录鉴权，产品侧决定暂时将「当前详情」读写接口作为开放的
|
|
# 内网集成接口。严格限制为一个 task id 段与 detail 终点，不能因为
|
|
# 此豁免而放开 versions、其他力量态势接口或整个 /api/force-situation 前缀。
|
|
_FORCE_SITUATION_DETAIL_RE = re.compile(
|
|
r"^/api/force-situation/tasks/[^/]+/(?:detail|versions)$"
|
|
)
|
|
|
|
|
|
def _is_force_situation_detail_external_request(method: str, path: str) -> bool:
|
|
"""Whether this is the credential-free force-situation detail read/write API."""
|
|
|
|
normalized = path.rstrip("/")
|
|
if not _FORCE_SITUATION_DETAIL_RE.fullmatch(normalized):
|
|
return False
|
|
return method.upper() == "GET" or (
|
|
method.upper() == "PUT" and normalized.endswith("/detail")
|
|
)
|
|
|
|
|
|
def _authorize_task_deeplink_external_write(request: Request) -> JSONResponse | None:
|
|
"""Mark the external b5-1 task-detail write; no credential is required.
|
|
|
|
The embedding system writes task snapshots server-to-server before the
|
|
user's browser is involved, so by product decision this PUT is an open
|
|
endpoint: any caller may store a snapshot without a login session or an
|
|
integration key. A supplied ``X-Integration-Key`` header is accepted but
|
|
only recorded (fingerprint) for audit; it is never validated. Reads still
|
|
follow the normal login path.
|
|
"""
|
|
|
|
request.state.task_deeplink_external_writer = True
|
|
supplied = request.headers.get("X-Integration-Key", "")
|
|
if supplied:
|
|
request.state.task_deeplink_external_key_fingerprint = hashlib.sha256(supplied.encode("utf-8")).hexdigest()[:8]
|
|
return None
|
|
|
|
|
|
def _is_weknora_embed_proxy_request(request: Request) -> bool:
|
|
"""Let the WeKnora embed proxy verify its own signed iframe cookie.
|
|
|
|
The embedded WeKnora frontend loads root-relative routes (``/platform``,
|
|
``/assets``, ``/locales`` and ``/api/v1/*``). Those requests do not carry
|
|
DeerFlow's Authorization header, but the iframe bootstrap has already set a
|
|
short-lived signed cookie that the proxy router validates against the
|
|
requested knowledge base.
|
|
"""
|
|
|
|
path = app_relative_path(request)
|
|
if not is_weknora_embed_proxy_path(path):
|
|
return False
|
|
return has_valid_weknora_embed_cookie(request.cookies.get(WEKNORA_EMBED_COOKIE))
|
|
|
|
|
|
# 「发起问答」的 run 创建接口(会真正调用大模型):
|
|
# POST /api/threads/{id}/runs[/stream|/wait]
|
|
# POST /api/runs/{stream|wait}
|
|
# （含 nginx 未改写到的 /api/langgraph/... 前缀,双保险)
|
|
# 只匹配这些「提交问答」的 POST,不碰 GET 列表 / join / cancel / feedback 等,
|
|
# 这样登录白名单开启后,未放行用户(凭旧 token)无法再用接口问答,但普通浏览不受影响。
|
|
_QA_RUN_RE = re.compile(r"^/api/(?:langgraph/)?(?:threads/[^/]+/)?runs(?:/stream|/wait)?$")
|
|
|
|
|
|
def _is_qa_run_request(method: str, path: str) -> bool:
|
|
"""True for an authenticated「发起问答」run-creation request (see ``_QA_RUN_RE``)."""
|
|
return method.upper() == "POST" and bool(_QA_RUN_RE.match(path.rstrip("/")))
|
|
|
|
|
|
class AuthMiddleware(BaseHTTPMiddleware):
|
|
"""Strict auth gate: reject requests without a valid session.
|
|
|
|
Two-stage check for non-public paths:
|
|
|
|
1. Cookie presence — return 401 NOT_AUTHENTICATED if missing
|
|
2. JWT validation via ``get_optional_user_from_request`` — return 401
|
|
TOKEN_INVALID if the token is absent, malformed, expired, or the
|
|
signed user does not exist / is stale
|
|
|
|
On success, stamps ``request.state.user`` and the
|
|
``deerflow.runtime.user_context`` contextvar so that repository-layer
|
|
owner filters work downstream without every route needing a
|
|
``@require_auth`` decorator. Routes that need per-resource
|
|
authorization (e.g. "user A cannot read user B's thread by guessing
|
|
the URL") should additionally use ``@require_permission(...,
|
|
owner_check=True)`` for explicit enforcement — but authentication
|
|
itself is fully handled here.
|
|
"""
|
|
|
|
def __init__(self, app: ASGIApp) -> None:
|
|
super().__init__(app)
|
|
|
|
async def dispatch(self, request: Request, call_next: Callable) -> Response:
|
|
# Match the application-relative path so a reverse-proxy prefix
|
|
# (e.g. /xxx/api/public/...) does not defeat the public whitelist.
|
|
relative_path = app_relative_path(request)
|
|
if _is_external_llmwiki(relative_path):
|
|
rejected = _authorize_external_llmwiki(request)
|
|
return rejected if rejected is not None else await call_next(request)
|
|
if _is_task_deeplink_external_write(request, relative_path):
|
|
rejected = _authorize_task_deeplink_external_write(request)
|
|
if rejected is not None:
|
|
return rejected
|
|
if getattr(request.state, "task_deeplink_external_writer", False):
|
|
return await call_next(request)
|
|
if _is_force_situation_detail_external_request(request.method, relative_path):
|
|
return await call_next(request)
|
|
if _is_public(relative_path):
|
|
return await call_next(request)
|
|
if _is_weknora_embed_proxy_request(request):
|
|
return await call_next(request)
|
|
|
|
if is_auth_disabled():
|
|
user = await get_assumed_user_for_disabled_auth()
|
|
request.state.user = user
|
|
request.state.auth = AuthContext(user=user, permissions=_ALL_PERMISSIONS)
|
|
token = set_current_user(user)
|
|
try:
|
|
return await call_next(request)
|
|
finally:
|
|
reset_current_user(token)
|
|
|
|
internal_user = None
|
|
if is_valid_internal_auth_token(request.headers.get(INTERNAL_AUTH_HEADER_NAME)):
|
|
internal_user = get_internal_user()
|
|
|
|
# Non-public path: require session cookie
|
|
from app.gateway.deps import get_request_access_token
|
|
|
|
if internal_user is None and not get_request_access_token(request):
|
|
return JSONResponse(
|
|
status_code=401,
|
|
content={
|
|
"detail": AuthErrorResponse(
|
|
code=AuthErrorCode.NOT_AUTHENTICATED,
|
|
message="Authentication required",
|
|
).model_dump()
|
|
},
|
|
)
|
|
|
|
# Strict JWT validation: reject junk/expired tokens with 401
|
|
# right here instead of silently passing through. This closes
|
|
# the "junk cookie bypass" gap (AUTH_TEST_PLAN test 7.5.8):
|
|
# without this, non-isolation routes like /api/models would
|
|
# accept any cookie-shaped string as authentication.
|
|
#
|
|
# We call the *strict* resolver so that fine-grained error
|
|
# codes (token_expired, token_invalid, user_not_found, …)
|
|
# propagate from AuthErrorCode, not get flattened into one
|
|
# generic code. BaseHTTPMiddleware doesn't let HTTPException
|
|
# bubble up, so we catch and render it as JSONResponse here.
|
|
from app.gateway.deps import get_current_user_from_request
|
|
|
|
if internal_user is not None:
|
|
user = internal_user
|
|
else:
|
|
try:
|
|
user = await get_current_user_from_request(request)
|
|
except HTTPException as exc:
|
|
return JSONResponse(status_code=exc.status_code, content={"detail": exc.detail})
|
|
|
|
if internal_user is None:
|
|
from app.gateway.auth import TokenError, decode_token
|
|
|
|
payload = decode_token(get_request_access_token(request) or "")
|
|
if not isinstance(payload, TokenError) and payload.auth_method == "service":
|
|
method = request.method.upper()
|
|
path = app_relative_path(request).rstrip("/")
|
|
allowed = _is_service_scope_request(
|
|
payload.scope,
|
|
method,
|
|
path,
|
|
all_requested=request.query_params.get("all", "").lower() in {"true", "1", "yes", "on"},
|
|
)
|
|
if not allowed:
|
|
return JSONResponse(status_code=403, content={"detail": "Service token is not permitted for this endpoint"})
|
|
|
|
# Stamp both request.state.user (for the contextvar pattern)
|
|
# and request.state.auth (so @require_permission's "auth is
|
|
# None" branch short-circuits instead of running the entire
|
|
# JWT-decode + DB-lookup pipeline a second time per request).
|
|
request.state.user = user
|
|
request.state.auth = AuthContext(user=user, permissions=_ALL_PERMISSIONS)
|
|
|
|
# 登录白名单:开关开启时,未放行用户(非 admin)不仅被拦在登录外,连「接口
|
|
# 问答」也封掉——已拿到 token 的未放行用户用旧会话直接调问答接口同样被拒。
|
|
# 仅作用于「发起问答」的 run 接口;内部/渠道调用(internal_user)不受限。
|
|
if internal_user is None and _is_qa_run_request(request.method, app_relative_path(request)):
|
|
from app.gateway.routers.auth import enforce_user_approved
|
|
|
|
try:
|
|
enforce_user_approved(user)
|
|
except HTTPException as exc:
|
|
return JSONResponse(status_code=exc.status_code, content={"detail": exc.detail})
|
|
|
|
token = set_current_user(user)
|
|
try:
|
|
return await call_next(request)
|
|
finally:
|
|
reset_current_user(token)
|
|
|
|
|
|
_LLMWIKI_SYNC_COLLECTION = "/api/llmwiki/knowledge-bases"
|
|
_LLMWIKI_SYNC_ITEM_RE = re.compile(r"^/api/llmwiki/knowledge-bases/[^/]+$")
|
|
_LLMWIKI_SYNC_DOCUMENT_RE = re.compile(r"^/api/llmwiki/knowledge-bases/[^/]+/documents(?:/[^/]+)?$")
|
|
_LLMWIKI_SYNC_MANUAL_RE = re.compile(r"^/api/llmwiki/knowledge-bases/[^/]+/documents/manual$")
|
|
|
|
|
|
def _is_llmwiki_sync_request(method: str, path: str) -> bool:
|
|
if path == _LLMWIKI_SYNC_COLLECTION:
|
|
return method in {"GET", "POST"}
|
|
if _LLMWIKI_SYNC_ITEM_RE.fullmatch(path):
|
|
return method in {"GET", "PATCH", "DELETE"}
|
|
if _LLMWIKI_SYNC_MANUAL_RE.fullmatch(path):
|
|
return method == "POST"
|
|
if _LLMWIKI_SYNC_DOCUMENT_RE.fullmatch(path):
|
|
return method in {"GET", "POST", "DELETE"}
|
|
return False
|
|
|
|
|
|
def _is_service_scope_request(scopes: list[str], method: str, path: str, *, all_requested: bool = False) -> bool:
|
|
return (
|
|
("llmwiki:sync" in scopes and _is_llmwiki_sync_request(method, path))
|
|
or ("recommended-questions:read" in scopes and method == "GET" and path == "/api/recommended-questions" and not all_requested)
|
|
)
|