fanilo/Sage100-vps
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(security): improve auth handling and openapi schema generation

Browse source
This commit is contained in:
Fanilo-Nantenaina 2026-01-21 13:50:44 +03:00
parent a6a623d1ab
commit 574d82f3c4
2 changed files with 23 additions and 21 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

23
api.py
View file

@ -343,15 +343,7 @@ def generate_filtered_openapi_schema(
base_schema["components"]["securitySchemes"] = auth_schemes base_schema["components"]["securitySchemes"] = auth_schemes
security_requirements = [] base_schema["security"] = []
if "HTTPBearer" in auth_schemes and "ApiKeyAuth" in auth_schemes:
security_requirements = [{"HTTPBearer": []}, {"ApiKeyAuth": []}]
elif "HTTPBearer" in auth_schemes:
security_requirements = [{"HTTPBearer": []}]
elif "ApiKeyAuth" in auth_schemes:
security_requirements = [{"ApiKeyAuth": []}]
base_schema["security"] = security_requirements
if not allowed_tags: if not allowed_tags:
logger.info("⚙️ Schéma OpenAPI complet (admin)") logger.info("⚙️ Schéma OpenAPI complet (admin)")
@ -377,7 +369,13 @@ def generate_filtered_openapi_schema(
operation_tags = operation.get("tags", []) operation_tags = operation.get("tags", [])
if any(tag in allowed_tags for tag in operation_tags): if any(tag in allowed_tags for tag in operation_tags):
operation["security"] = security_requirements 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 filtered_operations[method] = operation
if filtered_operations: if filtered_operations:
@ -398,7 +396,6 @@ def generate_filtered_openapi_schema(
referenced_schemas = set() referenced_schemas = set()
def extract_schema_refs(obj): def extract_schema_refs(obj):
"""Extrait récursivement tous les $ref depuis un objet"""
if isinstance(obj, dict): if isinstance(obj, dict):
for key, value in obj.items(): for key, value in obj.items():
if key == "$ref" and isinstance(value, str): if key == "$ref" and isinstance(value, str):
@ -430,12 +427,12 @@ def generate_filtered_openapi_schema(
base_schema["components"]["schemas"] = filtered_schemas base_schema["components"]["schemas"] = filtered_schemas
logger.info( 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}" f"{len(filtered_schemas)} schémas, tags: {allowed_tags}"
) )
else: else:
logger.info( 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 return base_schema

21
middleware/security.py
View file

@ -159,13 +159,12 @@ class ApiKeyMiddlewareHTTP(BaseHTTPMiddleware):
api_key_header = request.headers.get("X-API-Key") api_key_header = request.headers.get("X-API-Key")
if api_key_header: if api_key_header:
logger.debug(f" API Key détectée pour {method} {path}") api_key_header = api_key_header.strip()
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
)
if auth_header and auth_header.startswith("Bearer "): if auth_header and auth_header.startswith("Bearer "):
token = auth_header.split(" ")[1] token = auth_header.split(" ")[1].strip()
if token.startswith("sdk_live_"): if token.startswith("sdk_live_"):
logger.warning( logger.warning(
@ -179,7 +178,13 @@ class ApiKeyMiddlewareHTTP(BaseHTTPMiddleware):
request.state.authenticated_via = "jwt" request.state.authenticated_via = "jwt"
return await call_next(request) return await call_next(request)
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
)
logger.debug(f"❌ Aucune auth pour {method} {path} → délégation à FastAPI")
return await call_next(request) return await call_next(request)
async def _handle_api_key_auth( async def _handle_api_key_auth(
@ -200,7 +205,7 @@ class ApiKeyMiddlewareHTTP(BaseHTTPMiddleware):
api_key_obj = await service.verify_api_key(api_key) api_key_obj = await service.verify_api_key(api_key)
if not api_key_obj: if not api_key_obj:
logger.warning(f" Clé API invalide: {method} {path}") logger.warning(f"🔒 Clé API invalide: {method} {path}")
return JSONResponse( return JSONResponse(
status_code=status.HTTP_401_UNAUTHORIZED, status_code=status.HTTP_401_UNAUTHORIZED,
content={ content={
@ -250,7 +255,7 @@ class ApiKeyMiddlewareHTTP(BaseHTTPMiddleware):
request.state.api_key = api_key_obj request.state.api_key = api_key_obj
request.state.authenticated_via = "api_key" 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) return await call_next(request)