core/engine/src/middleware/auth.py - main - alexefan136/flowstack

core/engine/src/middleware/auth.py
226 строк8 KB
Alexander Efanov
upd fix
31 июл 2026, 19:17
31 июл 2026, 19:17d146d86
"""Auth middleware для извлечения user context из Gateway headers.

⚠️ Security: Engine должен быть доступен ТОЛЬКО через Gateway (который
проверяет JWT и устанавливает X-User-ID/X-Workspace-ID). При прямом доступе
заголовки можно подделать. Для защиты настройте network policy или shared secret.

Использует pure ASGI middleware (не BaseHTTPMiddleware) — корректно работает
с SSE streaming (не буферизует response).
"""

from __future__ import annotations

import json
import os
import uuid
from dataclasses import dataclass
from typing import Any

import structlog
from fastapi import HTTPException, Request, status

logger = structlog.get_logger()

# Default user для dev/optional режимов
DEFAULT_USER_ID = uuid.UUID("00000000-0000-0000-0000-000000000000")
DEFAULT_WORKSPACE_ID = "default"


@dataclass
class UserContext:
    """Контекст текущего пользователя."""

    user_id: uuid.UUID
    workspace_id: str


# Пути, которые НЕ требуют авторизации
PUBLIC_PATHS = {
    "/health",
    "/ready",
    "/info",
    "/docs",
    "/redoc",
    "/openapi.json",
    "/metrics",
}


class AuthMiddleware:
    """
    Pure ASGI middleware для извлечения user context из заголовков Gateway.

    Pure ASGI (вместо BaseHTTPMiddleware) — не буферизует response,
    что критично для SSE streaming endpoints.

    Args:
        app: ASGI приложение
        auth_mode: Режим авторизации ("required", "optional", "disabled")
    """

    def __init__(self, app: Any, auth_mode: str = "optional") -> None:
        self.app = app
        self.auth_mode = auth_mode
        logger.info("auth.middleware_initialized", auth_mode=auth_mode)

    async def __call__(self, scope: Any, receive: Any, send: Any) -> None:
        """ASGI entry point."""
        # Обрабатываем только HTTP запросы
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        path = scope.get("path", "")

        # Пропускаем публичные endpoints
        if self._is_public(path):
            await self.app(scope, receive, send)
            return

        # Если авторизация отключена — используем default context
        if self.auth_mode == "disabled":
            self._set_context(scope, DEFAULT_USER_ID, DEFAULT_WORKSPACE_ID)
            logger.debug("auth.disabled_mode", path=path)
            await self.app(scope, receive, send)
            return

        # Извлекаем заголовки от Gateway
        headers = self._get_headers(scope)
        user_id_str = headers.get("x-user-id")
        workspace_id = headers.get("x-workspace-id")

        # Для API endpoints
        if path.startswith("/api/"):
            # Optional mode: fallback на default если нет заголовков
            if self.auth_mode == "optional" and (not user_id_str or not workspace_id):
                self._set_context(scope, DEFAULT_USER_ID, workspace_id or DEFAULT_WORKSPACE_ID)
                logger.debug(
                    "auth.optional_mode_fallback",
                    path=path,
                    workspace_id=workspace_id or DEFAULT_WORKSPACE_ID,
                )
                await self.app(scope, receive, send)
                return

            # Required mode: требуем заголовки
            if not user_id_str:
                logger.warning("auth.missing_user_id", path=path)
                await self._send_error(send, 401, "Authentication required (missing X-User-ID)")
                return

            if not workspace_id:
                logger.warning("auth.missing_workspace_id", path=path)
                await self._send_error(
                    send, 401, "Workspace context required (missing X-Workspace-ID)"
                )
                return

            try:
                user_id = uuid.UUID(user_id_str)
            except ValueError:
                logger.warning("auth.invalid_user_id", user_id=user_id_str)
                await self._send_error(send, 400, "Invalid X-User-ID format")
                return

            # Сохраняем контекст в scope state
            self._set_context(scope, user_id, workspace_id)
            logger.debug(
                "auth.authenticated",
                user_id=str(user_id),
                workspace_id=workspace_id,
                path=path,
            )

        await self.app(scope, receive, send)

    # ========================================================================
    # HELPERS
    # ========================================================================

    @staticmethod
    def _is_public(path: str) -> bool:
        """Проверить, является ли путь публичным."""
        return (
            path in PUBLIC_PATHS
            or path.startswith("/docs")
            or path.startswith("/redoc")
            or path.startswith("/openapi")
        )

    @staticmethod
    def _get_headers(scope: Any) -> dict[str, str]:
        """Извлечь заголовки из ASGI scope (lowercase keys)."""
        return {
            key.decode("latin-1").lower(): value.decode("latin-1")
            for key, value in scope.get("headers", [])
        }

    @staticmethod
    def _set_context(scope: Any, user_id: uuid.UUID, workspace_id: str) -> None:
        """Сохранить UserContext в scope state (доступен как request.state)."""
        scope.setdefault("state", {})
        scope["state"]["user_context"] = UserContext(user_id=user_id, workspace_id=workspace_id)

    @staticmethod
    async def _send_error(send: Any, status_code: int, detail: str) -> None:
        """Отправить JSON error response (pure ASGI)."""
        body = json.dumps({"detail": detail}).encode("utf-8")
        await send(
            {
                "type": "http.response.start",
                "status": status_code,
                "headers": [
                    (b"content-type", b"application/json"),
                    (b"content-length", str(len(body)).encode("latin-1")),
                ],
            }
        )
        await send(
            {
                "type": "http.response.body",
                "body": body,
            }
        )


# ============================================================================
# DEPENDENCIES
# ============================================================================


def get_user_context(request: Request) -> UserContext | None:
    """Dependency для получения user context (может быть None)."""
    return getattr(request.state, "user_context", None)


def require_user_context(request: Request) -> UserContext:
    """Dependency для обязательного user context."""
    ctx = get_user_context(request)
    if ctx is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Authentication required",
        )
    return ctx


# ============================================================================
# FACTORY
# ============================================================================


def create_auth_middleware(app: Any, auth_mode: str | None = None) -> AuthMiddleware:
    """
    Factory функция для создания AuthMiddleware.

    Args:
        app: ASGI приложение
        auth_mode: Режим авторизации (если None — читается из ENV)

    Returns:
        Настроенный middleware.
    """
    if auth_mode is None:
        auth_mode = os.getenv("FLOWSTACK_AUTH_MODE", "optional").lower()

    return AuthMiddleware(app, auth_mode=auth_mode)