backend/src/main.py - master - Vibek/Agent_GG

backend/src/main.py
330 строк13 KB
Vladimir
P0 безопасность API, auth и прод-конфиг
26 май 2026, 19:02
26 май 2026, 19:02ca07c54
from __future__ import annotations

from contextlib import asynccontextmanager
from hmac import compare_digest
import logging
import os
import secrets
from typing import Any
from uuid import uuid4

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.staticfiles import StaticFiles
from sqlalchemy import text

from src.api.v1.auth import router as auth_router
from src.api.v1.chat import router as chat_router
from src.api.v1.db_viewer import router as db_viewer_router
from src.api.v1.listings import router as listings_router
from src.core.config import get_settings
from src.db import SessionLocal, create_db_and_tables
from src.metrics import ensure_metrics_views, probe_metrics_database
from src.modules.auth import InvalidAccessTokenError, verify_access_token
from src.modules.chat.agent_gateway import initialize_agent_runtime, is_agent_runtime_initialized

from agent.evaluator.vectorization.readiness import evaluate_document_store_readiness


settings = get_settings()
_PARSE_PROGRESS_PATH_PREFIX = f"{settings.api_prefix}/chat/parse-progress/"
_EVAL_PROGRESS_PATH_PREFIX = f"{settings.api_prefix}/chat/eval-progress/"
_runtime_tunnel_access_token: str | None = None
_TUNNEL_SKIP_PATHS = frozenset(
    {
        "/health",
        "/docs",
        "/openapi.json",
        "/favicon.ico",
    }
)
_NOISY_LOG_PATH_PREFIXES = (
    _PARSE_PROGRESS_PATH_PREFIX,
    _EVAL_PROGRESS_PATH_PREFIX,
    "/health",
)
_STATIC_EXTENSIONS = (
    ".css",
    ".js",
    ".map",
    ".ico",
    ".png",
    ".jpg",
    ".jpeg",
    ".svg",
    ".webp",
    ".woff",
    ".woff2",
    ".ttf",
)
_AUTH_EXEMPT_API_PATHS = {
    f"{settings.api_prefix}/auth/login",
    f"{settings.api_prefix}/auth/register",
}


def _should_skip_request_log(path: str) -> bool:
    if not path:
        return False
    lowered = path.lower()
    if lowered.startswith(_NOISY_LOG_PATH_PREFIXES):
        return True
    return lowered.endswith(_STATIC_EXTENSIONS)


def _is_local_metrics_enabled() -> bool:
    return str(os.getenv("GG_LOCAL_METRICS_ENABLED", "1")).strip().lower() not in {"0", "false", "off", "no"}


class _ExcludeParseProgressAccessFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        try:
            message = record.getMessage()
            lowered = message.lower()
            if any(prefix in lowered for prefix in _NOISY_LOG_PATH_PREFIXES):
                return False
            if any(ext in lowered for ext in _STATIC_EXTENSIONS):
                return False
            return True
        except Exception:
            return True

if not logging.getLogger().handlers:
    logging.basicConfig(
        level=getattr(logging, os.getenv("GG_LOG_LEVEL", "INFO").upper(), logging.INFO),
        format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
    )

uvicorn_access_logger = logging.getLogger("uvicorn.access")
if not any(isinstance(existing_filter, _ExcludeParseProgressAccessFilter) for existing_filter in uvicorn_access_logger.filters):
    uvicorn_access_logger.addFilter(_ExcludeParseProgressAccessFilter())


def create_app() -> FastAPI:
    @asynccontextmanager
    async def lifespan(_app: FastAPI):
        global _runtime_tunnel_access_token
        logger = logging.getLogger(__name__)
        logger.info("Application startup: initializing runtime dependencies")

        if settings.tunnel_mode_enabled:
            if settings.tunnel_access_token:
                _runtime_tunnel_access_token = settings.tunnel_access_token
                logger.info("Tunnel mode enabled: using GG_TUNNEL_ACCESS_TOKEN from environment")
            else:
                _runtime_tunnel_access_token = secrets.token_urlsafe(32)
                logger.warning(
                    "Tunnel mode enabled: GG_TUNNEL_ACCESS_TOKEN is empty, generated runtime token=%s",
                    _runtime_tunnel_access_token,
                )
        else:
            _runtime_tunnel_access_token = None

        # Fail-fast на старте: интеграция с agent должна быть готова до первого запроса.
        initialize_agent_runtime()

        # Минимальная readiness-проверка БД на старте.
        session = SessionLocal()
        try:
            session.execute(text("SELECT 1"))
        except Exception as exc:
            logger.exception("Application startup failed: database probe error: %s", exc)
            raise RuntimeError("Database is not available during startup") from exc
        finally:
            session.close()

        # Runtime-safe migration path for MVP additions (e.g. users table for auth).
        create_db_and_tables()

        if _is_local_metrics_enabled():
            try:
                probe_metrics_database()
                ensure_metrics_views()
            except Exception as exc:
                if settings.metrics_database_required:
                    logger.exception("Application startup failed: metrics database probe error: %s", exc)
                    raise RuntimeError("Metrics database is not available during startup") from exc
                logger.warning(
                    "Metrics database is unavailable, continuing without metrics DB: %s",
                    exc,
                )
        else:
            logger.info("Local metrics are disabled (GG_LOCAL_METRICS_ENABLED=0); metrics DB probe skipped")

        logger.info("Application startup completed")
        try:
            yield
        finally:
            _runtime_tunnel_access_token = None
            logger.info("Application shutdown completed")

    app = FastAPI(
        title=settings.app_name,
        version=settings.app_version,
        docs_url="/docs",
        redoc_url=None,
        lifespan=lifespan,
    )

    cors_raw = (os.getenv("GG_CORS_ORIGINS") or "").strip()
    if cors_raw:
        from fastapi.middleware.cors import CORSMiddleware

        cors_origins = [o.strip() for o in cors_raw.split(",") if o.strip()]
        if cors_origins:
            app.add_middleware(
                CORSMiddleware,
                allow_origins=cors_origins,
                allow_credentials=True,
                allow_methods=["*"],
                allow_headers=["*"],
            )

    @app.middleware("http")
    async def require_tunnel_token(request: Request, call_next):
        if not settings.tunnel_mode_enabled:
            return await call_next(request)
        request_path = request.url.path
        if request_path in _TUNNEL_SKIP_PATHS:
            return await call_next(request)
        if request_path.startswith("/docs/") or request_path.startswith("/openapi"):
            return await call_next(request)
        presented_token = (
            request.headers.get("X-Tunnel-Token")
            or request.headers.get("x-tunnel-token")
            or ""
        ).strip()
        expected_token = str(settings.tunnel_access_token or "")
        if not expected_token:
            expected_token = str(_runtime_tunnel_access_token or "")
        if not presented_token or not compare_digest(presented_token, expected_token):
            return JSONResponse(status_code=401, content={"detail": "Unauthorized tunnel token"})
        return await call_next(request)

    @app.middleware("http")
    async def require_bearer_token(request: Request, call_next):
        path = request.url.path
        if not path.startswith(settings.api_prefix) or path in _AUTH_EXEMPT_API_PATHS:
            return await call_next(request)
        auth_header = str(request.headers.get("Authorization") or "")
        scheme, _, token = auth_header.partition(" ")
        if scheme.lower() != "bearer" or not token.strip():
            return JSONResponse(status_code=401, content={"detail": "Missing bearer token"})
        try:
            payload = verify_access_token(token.strip(), secret=settings.auth_jwt_secret)
            request.state.auth_user = payload
        except InvalidAccessTokenError:
            return JSONResponse(status_code=401, content={"detail": "Invalid or expired token"})
        return await call_next(request)

    @app.middleware("http")
    async def add_request_id(request: Request, call_next):
        request_id = request.headers.get("X-Request-ID", str(uuid4()))
        request.state.request_id = request_id
        logger = logging.getLogger(__name__)
        skip_request_log = _should_skip_request_log(request.url.path)
        if not skip_request_log:
            logger.info("[request_id=%s] %s %s started", request_id, request.method, request.url.path)
        response = await call_next(request)
        response.headers["X-Request-ID"] = request_id
        if not skip_request_log:
            logger.info(
                "[request_id=%s] %s %s finished status=%s",
                request_id,
                request.method,
                request.url.path,
                response.status_code,
            )
        return response

    @app.exception_handler(Exception)
    async def unhandled_exception_handler(request: Request, exc: Exception) -> JSONResponse:
        request_id = getattr(request.state, "request_id", str(uuid4()))
        logging.getLogger(__name__).exception(
            "[request_id=%s] Unhandled exception on %s %s: %s",
            request_id,
            request.method,
            request.url.path,
            exc,
        )
        response = JSONResponse(status_code=500, content={"detail": "Internal server error"})
        response.headers["X-Request-ID"] = request_id
        return response

    @app.get("/health", tags=["system"])
    async def healthcheck() -> dict[str, Any]:
        agent_ready = is_agent_runtime_initialized()

        db_ready = False
        db_error: str | None = None
        metrics_db_ready = False
        metrics_db_error: str | None = None
        evaluator_document_store_ready = False
        evaluator_document_store_error: str | None = None
        evaluator_document_store_details: dict[str, Any] = {}
        session = SessionLocal()
        try:
            session.execute(text("SELECT 1"))
            db_ready = True
        except Exception as exc:
            db_error = str(exc)
            logging.getLogger(__name__).warning("Healthcheck DB probe failed: %s", exc)
        finally:
            session.close()

        if _is_local_metrics_enabled():
            try:
                probe_metrics_database()
                metrics_db_ready = True
            except Exception as exc:
                metrics_db_error = str(exc)
                logging.getLogger(__name__).warning("Healthcheck metrics DB probe failed: %s", exc)
        else:
            metrics_db_ready = False

        if not _is_local_metrics_enabled():
            status = "ok" if (agent_ready and db_ready) else "degraded"
            metrics_component_status = "disabled"
        elif settings.metrics_database_required:
            status = "ok" if (agent_ready and db_ready and metrics_db_ready) else "degraded"
            metrics_component_status = "ok" if metrics_db_ready else "failed"
        else:
            status = "ok" if (agent_ready and db_ready) else "degraded"
            metrics_component_status = "ok" if metrics_db_ready else "optional_unavailable"
        payload: dict[str, Any] = {
            "status": status,
            "components": {
                "agent_runtime": "ok" if agent_ready else "failed",
                "database": "ok" if db_ready else "failed",
                "metrics_database": metrics_component_status,
            },
        }
        details: dict[str, Any] = {}
        if db_error:
            details["database_error"] = db_error
        if metrics_db_error:
            details["metrics_database_error"] = metrics_db_error
        if evaluator_document_store_error:
            details["evaluator_document_store_error"] = evaluator_document_store_error
        if evaluator_document_store_details:
            details.update(evaluator_document_store_details)
        if details:
            payload["details"] = details
        return payload

    @app.get(f"{settings.api_prefix}/config", tags=["system"])
    async def get_config() -> dict[str, str]:
        return {
            "app_name": settings.app_name,
            "app_version": settings.app_version,
            "api_prefix": settings.api_prefix,
            "db_viewer_enabled": "1" if settings.db_viewer_enabled else "0",
        }

    app.include_router(auth_router, prefix=settings.api_prefix)
    app.include_router(chat_router, prefix=settings.api_prefix)
    app.include_router(listings_router, prefix=settings.api_prefix)
    if settings.db_viewer_enabled:
        app.include_router(db_viewer_router, prefix=settings.api_prefix)
    app.mount("/", StaticFiles(directory=settings.frontend_dir, html=True), name="frontend")
    return app