""" Nom du script : main.py Chemin : /gitlab-bridge/app/main.py Description : Point d'entrée FastAPI du bridge GitLab entre OpenWebUI et GitLab. Options éventuelles : Démarrage via uvicorn. Exemples d'utilisation : `uvicorn app.main:app --reload --host 0.0.0.0 --port 8080`. Prérequis : Python 3.11+, FastAPI, GitLab accessible, base MariaDB disponible. Auteur : Sylvain SCATTOLINI Date de création / modification : 2026-03-25 Version : 1.1 """ from __future__ import annotations from pathlib import Path import json import logging import time import uuid from typing import Any from dotenv import load_dotenv from fastapi import FastAPI, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse, Response BASE_DIR = Path(__file__).resolve().parents[1] load_dotenv(BASE_DIR / ".env") from app.api.v1.health import router as health_router from app.api.v1.issues import router as issues_router from app.api.v1.projects import router as projects_router from app.api.v1.groups import router as groups_router from app.api.v1.milestones import router as milestones_router from app.core.exceptions import BridgeError from app.core.logging import configure_logging from app.db.base import Base from app.db.session import engine configure_logging() logger = logging.getLogger(__name__) SENSITIVE_LOG_KEYS = { "password", "token", "api_token", "private_token", "authorization", "secret", "access_token", "refresh_token", } MAX_LOG_BODY_LENGTH = 4000 def _truncate_log_value(value: str, limit: int = MAX_LOG_BODY_LENGTH) -> str: if len(value) <= limit: return value return value[:limit] + f"... [truncated {len(value) - limit} chars]" def _sanitize_for_log(value: Any) -> Any: if isinstance(value, dict): sanitized: dict[str, Any] = {} for key, item in value.items(): if str(key).strip().lower() in SENSITIVE_LOG_KEYS: sanitized[str(key)] = "***REDACTED***" else: sanitized[str(key)] = _sanitize_for_log(item) return sanitized if isinstance(value, list): return [_sanitize_for_log(item) for item in value] if isinstance(value, tuple): return [_sanitize_for_log(item) for item in value] if isinstance(value, str): return _truncate_log_value(value) return value def _decode_body_for_log(body: bytes | None, content_type: str | None) -> Any: if not body: return None raw_content_type = (content_type or "").lower() if "application/json" in raw_content_type: try: return _sanitize_for_log(json.loads(body.decode("utf-8"))) except Exception: pass if ( "text/" in raw_content_type or "json" in raw_content_type or "xml" in raw_content_type or "x-www-form-urlencoded" in raw_content_type ): return _truncate_log_value(body.decode("utf-8", errors="replace")) return f"<{len(body)} bytes binary>" app = FastAPI( title="GitLab Bridge", version="1.1", openapi_version="3.0.3", ) from fastapi.openapi.utils import get_openapi def custom_openapi(): if app.openapi_schema: return app.openapi_schema openapi_schema = get_openapi( title=app.title, version=app.version, openapi_version="3.0.3", description=app.description, routes=app.routes, ) app.openapi_schema = openapi_schema return app.openapi_schema app.openapi = custom_openapi from app.core.config import settings app.add_middleware( CORSMiddleware, allow_origins=settings.cors_origins, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) app.include_router(groups_router) app.include_router(health_router) app.include_router(projects_router) app.include_router(issues_router) app.include_router(milestones_router) @app.middleware("http") async def log_http_requests(request: Request, call_next): request_id = uuid.uuid4().hex[:12] started_at = time.perf_counter() request_body = await request.body() request_extra = { "event": "http_request", "request_id": request_id, "method": request.method, "path": request.url.path, "query": str(request.url.query or ""), "client_ip": request.client.host if request.client else None, "user_agent": request.headers.get("user-agent"), "request_content_type": request.headers.get("content-type"), "request_body": _decode_body_for_log(request_body, request.headers.get("content-type")), } logger.info("HTTP request received", extra=request_extra) try: response = await call_next(request) except Exception: duration_ms = round((time.perf_counter() - started_at) * 1000, 2) logger.exception( "HTTP request failed", extra={ **request_extra, "event": "http_error", "duration_ms": duration_ms, }, ) raise response_body_bytes = b"" if hasattr(response, "body") and response.body is not None: response_body_bytes = response.body elif hasattr(response, "body_iterator") and response.body_iterator is not None: chunks = [chunk async for chunk in response.body_iterator] response_body_bytes = b"".join(chunks) response = Response( content=response_body_bytes, status_code=response.status_code, headers=dict(response.headers), media_type=response.media_type, background=response.background, ) response.headers["X-Request-ID"] = request_id duration_ms = round((time.perf_counter() - started_at) * 1000, 2) logger.info( "HTTP response sent", extra={ **request_extra, "event": "http_response", "status_code": response.status_code, "duration_ms": duration_ms, "response_content_type": response.headers.get("content-type"), "response_body": _decode_body_for_log( response_body_bytes, response.headers.get("content-type"), ), }, ) return response @app.on_event("startup") def startup() -> None: """Initialisation minimale de l'application.""" Base.metadata.create_all(bind=engine) logger.info("GitLab Bridge démarré.") @app.exception_handler(BridgeError) def handle_bridge_error(_: Request, exc: BridgeError) -> JSONResponse: """Transforme les erreurs métier en réponses JSON homogènes.""" return JSONResponse( status_code=400 if exc.code not in {"access_denied"} else 403, content={ "success": False, "error": exc.code, "message": exc.message, }, ) @app.exception_handler(Exception) def handle_unexpected_error(_: Request, exc: Exception) -> JSONResponse: """Gère les erreurs inattendues sans divulguer de détails sensibles au client.""" logger.exception("Erreur inattendue.", exc_info=exc) return JSONResponse( status_code=500, content={ "success": False, "error": "internal_error", "message": "Une erreur interne est survenue.", }, )