feat: implement API security measures phase 1
This commit is contained in:
@@ -0,0 +1,149 @@
|
||||
import logging
|
||||
import traceback
|
||||
|
||||
from django.conf import settings
|
||||
from django.core.exceptions import PermissionDenied, ValidationError as DjangoValidationError
|
||||
from django.http import Http404
|
||||
from rest_framework import exceptions, status
|
||||
from rest_framework.exceptions import APIException
|
||||
from rest_framework.response import Response
|
||||
from rest_framework.views import exception_handler as drf_exception_handler
|
||||
|
||||
logger = logging.getLogger("django.request")
|
||||
|
||||
|
||||
def _safe_error_response(
|
||||
detail: str,
|
||||
code: str,
|
||||
status_code: int,
|
||||
) -> Response:
|
||||
"""Return a consistent error response with no stack traces."""
|
||||
return Response(
|
||||
{"error": detail, "code": code},
|
||||
status=status_code,
|
||||
)
|
||||
|
||||
|
||||
def api_exception_handler(exc: Exception, context: dict) -> Response | None:
|
||||
"""
|
||||
Custom DRF exception handler that:
|
||||
- Never exposes stack traces, file paths, or Python internals
|
||||
- Maps common exceptions to user-safe messages
|
||||
- Logs full traceback to django.request logger
|
||||
- Returns consistent {error, code} format
|
||||
"""
|
||||
# Always log the full traceback
|
||||
logger.error(
|
||||
"API Exception: %s: %s\n%s",
|
||||
type(exc).__name__,
|
||||
exc,
|
||||
"".join(traceback.format_tb(exc.__traceback__)),
|
||||
)
|
||||
|
||||
# PermissionDenied -> 403
|
||||
if isinstance(exc, PermissionDenied):
|
||||
return _safe_error_response(
|
||||
"You do not have permission to perform this action.",
|
||||
"permission_denied",
|
||||
status.HTTP_403_FORBIDDEN,
|
||||
)
|
||||
|
||||
# Http404 -> 404
|
||||
if isinstance(exc, Http404):
|
||||
return _safe_error_response(
|
||||
"The requested resource was not found.",
|
||||
"not_found",
|
||||
status.HTTP_404_NOT_FOUND,
|
||||
)
|
||||
|
||||
# Django ValidationError -> 400
|
||||
if isinstance(exc, DjangoValidationError):
|
||||
return _safe_error_response(
|
||||
str(exc) if isinstance(exc.message, str) else "Validation error.",
|
||||
"validation_error",
|
||||
status.HTTP_400_BAD_REQUEST,
|
||||
)
|
||||
|
||||
# DRF APIException (includes AuthenticationFailed, NotAuthenticated, ParseError, etc.)
|
||||
if isinstance(exc, APIException):
|
||||
# Use DRF's standard handling but map to our format
|
||||
response = drf_exception_handler(exc, context)
|
||||
if response is not None:
|
||||
# Ensure response is our safe format
|
||||
detail = _extract_detail(response.data)
|
||||
return _safe_error_response(
|
||||
detail,
|
||||
_get_error_code(exc),
|
||||
response.status_code,
|
||||
)
|
||||
|
||||
# DRF Throttled
|
||||
if isinstance(exc, exceptions.Throttled):
|
||||
return _safe_error_response(
|
||||
"Request rate limit exceeded. Please try again later.",
|
||||
"throttled",
|
||||
exc.status_code,
|
||||
)
|
||||
|
||||
# AuthenticationFailed / NotAuthenticated
|
||||
if isinstance(exc, exceptions.AuthenticationFailed):
|
||||
return _safe_error_response(
|
||||
str(exc.detail) if hasattr(exc, "detail") else "Authentication failed.",
|
||||
"authentication_failed",
|
||||
status.HTTP_401_UNAUTHORIZED,
|
||||
)
|
||||
|
||||
if isinstance(exc, exceptions.NotAuthenticated):
|
||||
return _safe_error_response(
|
||||
"Authentication credentials were not provided.",
|
||||
"not_authenticated",
|
||||
status.HTTP_401_UNAUTHORIZED,
|
||||
)
|
||||
|
||||
# Catch-all for unhandled exceptions
|
||||
if not settings.DEBUG:
|
||||
return _safe_error_response(
|
||||
"Internal server error.",
|
||||
"internal_error",
|
||||
status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
)
|
||||
|
||||
# In DEBUG mode, let DRF's default handler show the traceback
|
||||
return drf_exception_handler(exc, context)
|
||||
|
||||
|
||||
def _extract_detail(data: dict | list | str) -> str:
|
||||
"""Extract the first meaningful error string from DRF error data."""
|
||||
if isinstance(data, str):
|
||||
return data
|
||||
if isinstance(data, list):
|
||||
for item in data:
|
||||
result = _extract_detail(item)
|
||||
if result:
|
||||
return result
|
||||
if isinstance(data, dict):
|
||||
# Try 'detail' first, then first field error
|
||||
if "detail" in data:
|
||||
return _extract_detail(data["detail"])
|
||||
for _key, value in data.items():
|
||||
result = _extract_detail(value)
|
||||
if result:
|
||||
return result
|
||||
return "An error occurred."
|
||||
|
||||
|
||||
def _get_error_code(exc: APIException) -> str:
|
||||
"""Map exception class to a stable error code string."""
|
||||
mapping: dict[type, str] = {
|
||||
exceptions.AuthenticationFailed: "authentication_failed",
|
||||
exceptions.NotAuthenticated: "not_authenticated",
|
||||
exceptions.PermissionDenied: "permission_denied",
|
||||
exceptions.NotFound: "not_found",
|
||||
exceptions.MethodNotAllowed: "method_not_allowed",
|
||||
exceptions.NotAcceptable: "not_acceptable",
|
||||
exceptions.UnsupportedMediaType: "unsupported_media_type",
|
||||
exceptions.Throttled: "throttled",
|
||||
exceptions.ParseError: "parse_error",
|
||||
exceptions.ValidationError: "validation_error",
|
||||
}
|
||||
return mapping.get(type(exc), "error")
|
||||
Reference in New Issue
Block a user