62 lines
1.9 KiB
Python
62 lines
1.9 KiB
Python
import logging
|
|
import uuid
|
|
|
|
from rest_framework.exceptions import PermissionDenied
|
|
from rest_framework.views import exception_handler
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _client_ip(request) -> str | None:
|
|
if request is None:
|
|
return None
|
|
forwarded = request.META.get("HTTP_X_FORWARDED_FOR")
|
|
if forwarded:
|
|
return forwarded.split(",")[0].strip()
|
|
return request.META.get("REMOTE_ADDR")
|
|
|
|
|
|
def custom_exception_handler(exc, context):
|
|
response = exception_handler(exc, context)
|
|
if response is None:
|
|
return None
|
|
|
|
request = context.get("request")
|
|
request_id = (
|
|
getattr(request, "META", {}).get("HTTP_X_REQUEST_ID") or str(uuid.uuid4())
|
|
)
|
|
|
|
code = getattr(exc, "default_code", "error")
|
|
message = str(exc.detail) if hasattr(exc, "detail") else str(exc)
|
|
details = exc.detail if isinstance(getattr(exc, "detail", None), dict) else {}
|
|
|
|
# Record access.permission_denied audit event for all 403s (M1 integration).
|
|
if isinstance(exc, PermissionDenied) and request is not None:
|
|
try:
|
|
from tracking.audit import record as audit_record
|
|
view = context.get("view")
|
|
resource_type = type(view).__name__ if view else None
|
|
audit_record(
|
|
"access.permission_denied",
|
|
actor=getattr(request, "user", None),
|
|
actor_ip=_client_ip(request),
|
|
object_type=resource_type,
|
|
metadata={
|
|
"path": request.path,
|
|
"method": request.method,
|
|
"request_id": request_id,
|
|
},
|
|
)
|
|
except Exception:
|
|
logger.exception("Failed to write access.permission_denied audit event")
|
|
|
|
response.data = {
|
|
"error": {
|
|
"code": code,
|
|
"message": message,
|
|
"details": details,
|
|
"request_id": request_id,
|
|
}
|
|
}
|
|
return response
|