api.py

@@ -278,21 +278,18 @@ def get_auth_schemes_for_user(swagger_user: dict) -> dict:
     allowed_tags = swagger_user.get("allowed_tags")
 
     if not allowed_tags:
-        # Admin complet
         return {
             "HTTPBearer": {
                 "type": "http",
                 "scheme": "bearer",
                 "bearerFormat": "JWT",
-                "description": "🎫 Authentification JWT pour utilisateurs (POST /auth/login). "
+                "description": "Authentification JWT pour utilisateurs (POST /auth/login)",
-                "Utilisez SOIT JWT SOIT API Key, pas les deux.",
             },
             "ApiKeyAuth": {
                 "type": "apiKey",
                 "in": "header",
                 "name": "X-API-Key",
-                "description": " Clé API pour intégrations externes (format: sdk_live_xxx). "
+                "description": "Clé API pour intégrations externes (format: sdk_live_xxx)",
-                "Utilisez SOIT JWT SOIT API Key, pas les deux.",
             },
         }
 
@@ -303,17 +300,24 @@ def get_auth_schemes_for_user(swagger_user: dict) -> dict:
             "type": "http",
             "scheme": "bearer",
             "bearerFormat": "JWT",
-            "description": "🎫 Authentification JWT pour utilisateurs (POST /auth/login). "
+            "description": "Authentification JWT pour utilisateurs (POST /auth/login)",
-            "Utilisez SOIT JWT SOIT API Key, pas les deux.",
         }
 
-    schemes["ApiKeyAuth"] = {
+    if "API Keys Management" in allowed_tags or len(allowed_tags) > 3:
-        "type": "apiKey",
+        schemes["ApiKeyAuth"] = {
-        "in": "header",
+            "type": "apiKey",
-        "name": "X-API-Key",
+            "in": "header",
-        "description": " Clé API pour intégrations externes (format: sdk_live_xxx). "
+            "name": "X-API-Key",
-        "Utilisez SOIT JWT SOIT API Key, pas les deux.",
+            "description": "Clé API pour intégrations externes (format: sdk_live_xxx)",
-    }
+        }
+
+    if not schemes:
+        schemes["HTTPBearer"] = {
+            "type": "http",
+            "scheme": "bearer",
+            "bearerFormat": "JWT",
+            "description": "Authentification requise",
+        }
 
     return schemes
 
@@ -348,10 +352,16 @@ def generate_filtered_openapi_schema(
 
     base_schema["components"]["securitySchemes"] = auth_schemes
 
-    base_schema["security"] = []
+    security_requirements = []
+    if "HTTPBearer" in auth_schemes:
+        security_requirements.append({"HTTPBearer": []})
+    if "ApiKeyAuth" in auth_schemes:
+        security_requirements.append({"ApiKeyAuth": []})
+
+    base_schema["security"] = security_requirements if security_requirements else []
 
     if not allowed_tags:
-        logger.info("⚙️ Schéma OpenAPI complet (admin)")
+        logger.info(" Schéma OpenAPI complet (admin)")
         return base_schema
 
     filtered_paths = {}
@@ -374,13 +384,6 @@ def generate_filtered_openapi_schema(
             operation_tags = operation.get("tags", [])
 
             if any(tag in allowed_tags for tag in operation_tags):
-                operation_security = []
-                if "HTTPBearer" in auth_schemes:
-                    operation_security.append({"HTTPBearer": []})
-                if "ApiKeyAuth" in auth_schemes:
-                    operation_security.append({"ApiKeyAuth": []})
-
-                operation["security"] = operation_security
                 filtered_operations[method] = operation
 
         if filtered_operations:
@@ -397,47 +400,16 @@ def generate_filtered_openapi_schema(
 
     if "components" in base_schema and "schemas" in base_schema["components"]:
         all_schemas = base_schema["components"]["schemas"]
-
+        filtered_schemas = get_schemas_for_tags(allowed_tags, all_schemas)
-        referenced_schemas = set()
-
-        def extract_schema_refs(obj):
-            if isinstance(obj, dict):
-                for key, value in obj.items():
-                    if key == "$ref" and isinstance(value, str):
-                        schema_name = value.split("/")[-1]
-                        referenced_schemas.add(schema_name)
-                    else:
-                        extract_schema_refs(value)
-            elif isinstance(obj, list):
-                for item in obj:
-                    extract_schema_refs(item)
-
-        extract_schema_refs(filtered_paths)
-
-        def add_dependencies(schema_name):
-            if schema_name not in all_schemas:
-                return
-            schema_def = all_schemas[schema_name]
-            extract_schema_refs(schema_def)
-
-        initial_schemas = referenced_schemas.copy()
-        for schema_name in initial_schemas:
-            add_dependencies(schema_name)
-
-        filtered_schemas = {}
-        for schema_name in referenced_schemas:
-            if schema_name in all_schemas:
-                filtered_schemas[schema_name] = all_schemas[schema_name]
-
         base_schema["components"]["schemas"] = filtered_schemas
 
         logger.info(
-            f"🔍 Schéma filtré: {len(filtered_paths)} paths, "
+            f" Schéma filtré: {len(filtered_paths)} paths, "
             f"{len(filtered_schemas)} schémas, tags: {allowed_tags}"
         )
     else:
         logger.info(
-            f"🔍 Schéma filtré: {len(filtered_paths)} paths, tags: {allowed_tags}"
+            f" Schéma filtré: {len(filtered_paths)} paths, tags: {allowed_tags}"
         )
 
     return base_schema
@@ -457,14 +429,10 @@ async def custom_openapi_endpoint(request: Request):
     username = swagger_user.get("username", "unknown")
     allowed_tags = swagger_user.get("allowed_tags")
 
-    logger.info(f"📖 OpenAPI demandé par: {username}, tags: {allowed_tags or 'ALL'}")
+    logger.info(f" OpenAPI demandé par: {username}, tags: {allowed_tags or 'ALL'}")
 
     schema = generate_filtered_openapi_schema(app, allowed_tags, swagger_user)
 
-    if request.url.scheme == "https":
-        if "servers" not in schema or not schema["servers"]:
-            schema["servers"] = [{"url": str(request.base_url).rstrip("/")}]
-
     return JSONResponse(content=schema)
 
 
@@ -488,9 +456,7 @@ async def custom_swagger_ui(request: Request):
             "displayRequestDuration": True,
             "filter": True,
             "tryItOutEnabled": True,
-            "docExpansion": "list",
+            "docExpansion": "list",  # Meilleure UX
-            # CORRECTIF : Ne pas pré-remplir les credentials
-            "preAuthorizeApiKey": False,
         },
     )
 

middleware/security.py

@@ -158,18 +158,15 @@ class ApiKeyMiddlewareHTTP(BaseHTTPMiddleware):
         auth_header = request.headers.get("Authorization")
         api_key_header = request.headers.get("X-API-Key")
 
-        # CORRECTIF : Nettoyer et valider la clé API
         if api_key_header:
-            api_key_header = api_key_header.strip()
+            logger.debug(f" API Key détectée pour {method} {path}")
-            # Si la clé est vide ou juste des espaces, la considérer comme absente
+            return await self._handle_api_key_auth(
-            if not api_key_header or api_key_header == "":
+                request, api_key_header, path, method, call_next
-                api_key_header = None
+            )
 
-        # Vérifier si c'est un token JWT Bearer
         if auth_header and auth_header.startswith("Bearer "):
-            token = auth_header.split(" ", 1)[1].strip()
+            token = auth_header.split(" ")[1]
 
-            # CORRECTIF : Vérifier si c'est une API Key envoyée par erreur dans Authorization
             if token.startswith("sdk_live_"):
                 logger.warning(
                     "⚠️ API Key envoyée dans Authorization au lieu de X-API-Key"
@@ -178,20 +175,11 @@ class ApiKeyMiddlewareHTTP(BaseHTTPMiddleware):
                     request, token, path, method, call_next
                 )
 
-            # C'est un JWT valide, déléguer à FastAPI
             logger.debug(f"🎫 JWT détecté pour {method} {path} → délégation à FastAPI")
             request.state.authenticated_via = "jwt"
             return await call_next(request)
 
-        # CORRECTIF : Si une clé API est présente ET non vide, la traiter
+        logger.debug(f"❓ Aucune auth pour {method} {path} → délégation à FastAPI")
-        if api_key_header:
-            logger.debug(f" API Key détectée pour {method} {path}")
-            return await self._handle_api_key_auth(
-                request, api_key_header, path, method, call_next
-            )
-
-        # Aucune authentification fournie, déléguer à FastAPI qui renverra 401
-        logger.debug(f"❌ Aucune auth pour {method} {path} → délégation à FastAPI")
         return await call_next(request)
 
     async def _handle_api_key_auth(
@@ -212,7 +200,7 @@ class ApiKeyMiddlewareHTTP(BaseHTTPMiddleware):
                 api_key_obj = await service.verify_api_key(api_key)
 
                 if not api_key_obj:
-                    logger.warning(f"🔒 Clé API invalide: {method} {path}")
+                    logger.warning(f" Clé API invalide: {method} {path}")
                     return JSONResponse(
                         status_code=status.HTTP_401_UNAUTHORIZED,
                         content={
@@ -262,7 +250,7 @@ class ApiKeyMiddlewareHTTP(BaseHTTPMiddleware):
                 request.state.api_key = api_key_obj
                 request.state.authenticated_via = "api_key"
 
-                logger.info(f"✅ ACCÈS AUTORISÉ: {api_key_obj.name} → {method} {path}")
+                logger.info(f" ACCÈS AUTORISÉ: {api_key_obj.name} → {method} {path}")
 
                 return await call_next(request)