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

fix(security): improve api key authentication and error handling

Browse source
This commit is contained in:
Fanilo-Nantenaina 2026-01-20 16:18:04 +03:00
parent 67ef83c4e3
commit 211dd4fd23
2 changed files with 48 additions and 25 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
api.py
View file

@ -187,8 +187,18 @@ def custom_openapi():
) )
openapi_schema["components"]["securitySchemes"] = { openapi_schema["components"]["securitySchemes"] = {
"HTTPBearer": {"type": "http", "scheme": "bearer", "bearerFormat": "JWT"}, "HTTPBearer": {
"ApiKeyAuth": {"type": "apiKey", "in": "header", "name": "X-API-Key"}, "type": "http",
"scheme": "bearer",
"bearerFormat": "JWT",
"description": "Authentification JWT pour utilisateurs (POST /auth/login)",
},
"ApiKeyAuth": {
"type": "apiKey",
"in": "header",
"name": "X-API-Key",
"description": "Clé API pour intégrations externes (format: sdk_live_xxx)",
},
} }
openapi_schema["security"] = [{"HTTPBearer": []}, {"ApiKeyAuth": []}] openapi_schema["security"] = [{"HTTPBearer": []}, {"ApiKeyAuth": []}]

55
middleware/security.py
View file

@ -132,26 +132,30 @@ class ApiKeyMiddlewareHTTP(BaseHTTPMiddleware):
method = request.method method = request.method
if self._is_excluded_path(path): if self._is_excluded_path(path):
logger.debug(f" Route publique: {method} {path}")
return await call_next(request) return await call_next(request)
auth_header = request.headers.get("Authorization") auth_header = request.headers.get("Authorization")
has_jwt = auth_header and auth_header.startswith("Bearer ") api_key_header = request.headers.get("X-API-Key")
api_key = request.headers.get("X-API-Key") if auth_header and auth_header.startswith("Bearer "):
has_api_key = bool(api_key) token = auth_header.split(" ")[1]
if has_jwt: if token.startswith("sdk_live_"):
logger.warning(
" API Key envoyée dans Authorization au lieu de X-API-Key"
)
api_key_header = token
else:
logger.debug(f" JWT détecté pour {method} {path}") logger.debug(f" JWT détecté pour {method} {path}")
return await call_next(request) return await call_next(request)
if has_api_key: if api_key_header:
logger.debug(f" API Key détectée pour {method} {path}") logger.debug(f" API Key détectée pour {method} {path}")
return await self._handle_api_key_auth( return await self._handle_api_key_auth(
request, api_key, path, method, call_next request, api_key_header, path, method, call_next
) )
logger.warning(f" Aucune authentification pour {method} {path}") logger.warning(f" Aucune authentification: {method} {path}")
return JSONResponse( return JSONResponse(
status_code=status.HTTP_401_UNAUTHORIZED, status_code=status.HTTP_401_UNAUTHORIZED,
content={ content={
@ -170,7 +174,7 @@ class ApiKeyMiddlewareHTTP(BaseHTTPMiddleware):
method: str, method: str,
call_next: Callable, call_next: Callable,
): ):
"""Gère l'authentification par API Key""" """Gère l'authentification par API Key avec vérification STRICTE"""
try: try:
from database.db_config import async_session_factory from database.db_config import async_session_factory
from services.api_key import ApiKeyService from services.api_key import ApiKeyService
@ -181,7 +185,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 pour {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={
@ -192,7 +196,7 @@ class ApiKeyMiddlewareHTTP(BaseHTTPMiddleware):
is_allowed, rate_info = await service.check_rate_limit(api_key_obj) is_allowed, rate_info = await service.check_rate_limit(api_key_obj)
if not is_allowed: if not is_allowed:
logger.warning(f"⚠️ Rate limit dépassé: {api_key_obj.name}") logger.warning(f"⚠️ Rate limit: {api_key_obj.name}")
return JSONResponse( return JSONResponse(
status_code=status.HTTP_429_TOO_MANY_REQUESTS, status_code=status.HTTP_429_TOO_MANY_REQUESTS,
content={"detail": "Rate limit dépassé"}, content={"detail": "Rate limit dépassé"},
@ -203,28 +207,37 @@ class ApiKeyMiddlewareHTTP(BaseHTTPMiddleware):
) )
has_access = await service.check_endpoint_access(api_key_obj, path) has_access = await service.check_endpoint_access(api_key_obj, path)
if not has_access: if not has_access:
logger.warning( import json
f"Accès refusé: {api_key_obj.name} → {method} {path}"
allowed = (
json.loads(api_key_obj.allowed_endpoints)
if api_key_obj.allowed_endpoints
else ["Tous"]
) )
logger.warning(
f" ACCÈS REFUSÉ: {api_key_obj.name}\n"
f" Endpoint demandé: {path}\n"
f" Endpoints autorisés: {allowed}"
)
return JSONResponse( return JSONResponse(
status_code=status.HTTP_403_FORBIDDEN, status_code=status.HTTP_403_FORBIDDEN,
content={ content={
"detail": "Accès non autorisé à cet endpoint", "detail": "Accès non autorisé à cet endpoint",
"endpoint": path, "endpoint_requested": path,
"api_key_name": api_key_obj.name, "api_key_name": api_key_obj.name,
"allowed_endpoints": ( "allowed_endpoints": allowed,
api_key_obj.allowed_endpoints "hint": "Cette clé API n'a pas accès à cet endpoint. Contactez l'administrateur.",
if api_key_obj.allowed_endpoints
else "Tous"
),
}, },
) )
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"✅ API Key valide: {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)
@ -232,7 +245,7 @@ class ApiKeyMiddlewareHTTP(BaseHTTPMiddleware):
logger.error(f" Erreur validation API Key: {e}", exc_info=True) logger.error(f" Erreur validation API Key: {e}", exc_info=True)
return JSONResponse( return JSONResponse(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
content={"detail": "Erreur interne lors de la validation"}, content={"detail": f"Erreur interne: {str(e)}"},
) )