"""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/login/username", "/api/v1/auth/login/token", "/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_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_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}) # 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)