drf-rack/net/xeaf/rack/middleware/custom_logging_middleware.py

# DRF Rack
# Библиотека классов расширений для Django REST Framework
#
# Автор: Николай В. Анохин <n.anokhin@xeaf.net>
# Все права защищены. Лицензия: MIT

"""
Описание класса CustomLoggingMiddleware
"""

import json
import logging
import traceback
from logging import Logger
from typing import Any

from django.conf import settings
from django.http import HttpRequest
from django.http import HttpResponse
from rest_framework import status


class CustomLoggingMiddleware:
    """
    Класс для логирования запросов и ответов DRF
    """

    DEFAULT_SENSITIVE_KEYS = {'password', 'old_password', 'new_password', 'token', 'access_token'}
    """ Словарь ключей, которые нужно замаскировать в логах """

    MAX_LOG_LENGTH = 2048
    """ Максимальная длина объекта данных лога """

    sensitive_keys: set = {}
    """ Актуальный словарь ключей, которые нужно замаскировать в логах """

    logger: Logger
    """ Объект доступа к логгеру """

    def __init__(self, get_response):
        """
        Инициализация
        """

        self.get_response = get_response
        self.logger = logging.getLogger(settings.LOGGER_NAME)

        self.sensitive_keys = set(self.DEFAULT_SENSITIVE_KEYS)
        if settings.LOGGER_SENSITIVE_KEYS:
            self.sensitive_keys.update(settings.LOGGER_SENSITIVE_KEYS)

    def __call__(self, request: HttpRequest):
        """
        Собираем данные запроса и отправляем в лог
        """

        # Обрабатываем данные запроса
        path = request.path
        method = request.method
        query_params = self._process_query_params(request)

        # Передаем запрос дальше по конвейеру Django / DRF
        response = self.get_response(request)

        # Обрабатываем данные запроса, ответа и перехваченного исключения
        body = self._process_body(request)
        exc = getattr(response, "_raised_exception", None)
        status_code = response.status_code
        request_id = request.META.get('HTTP_X_REQUEST_ID', '')
        response_content = self._process_response(response)

        # Обрабатываем данные пользователя
        user_id = ""
        if hasattr(request, 'user'):
            if request.user.is_authenticated:
                user_id = str(request.user.id)

        # Формируем единое сообщение для лога
        log_parts = [
            f"{status_code} [{method}] {path}"
        ]

        if user_id:
            log_parts.append(f"User ID: {user_id}")
        if request_id:
            log_parts.append(f"Request ID: {request_id}")
        if query_params:
            log_parts.append(f"Query Params: {query_params}")
        if body:
            log_parts.append(f"Request Body: {body}")
        if response_content:
            log_parts.append(f"Response Content: {response_content}")

        log_message = "\n".join(log_parts) + "\n"

        # Создаем и записываем лог
        self._write_log(log_message, status_code, exc)

        return response

    def _process_query_params(self, request: HttpRequest) -> dict[str, Any]:
        """
        Обрабатывает параметры запроса и маскирует конфиденциальные данные

        :param request: Объект запроса

        :return: Обработанные параметры запроса
        """

        raw_query_params = dict(request.GET.lists())
        return self._mask_sensitive_data(raw_query_params)

    def _process_body(self, request: HttpRequest) -> str:
        """
        Обрабатывает тело запроса и маскирует конфиденциальные данные

        :param request: Объект запроса

        :return: Обработанное тело запроса
        """

        if request.body:
            return self._process_data(request.body)

        return ""

    def _process_response(self, response: HttpResponse) -> str:
        """
        Обрабатываем данные ответа

        :param response: HttpResponse

        :return: Строковое представление данных ответа
        """

        if hasattr(response, 'content'):
            return self._process_data(response.content)

        return ""

    def _process_data(self, data: Any) -> str:
        """
        Обрабатывает данные запроса или ответа и возвращает их в виде строки

        :param data: Данные запроса или ответа

        :return: Обработанные данные в виде строки
        """

        data_len = len(data)
        if data_len > self.MAX_LOG_LENGTH:
            return f"<Data too large to log (len={data_len})>"

        try:
            raw_response = data.decode('utf-8')
            try:
                data_json = json.loads(raw_response)
                masked_response = self._mask_sensitive_data(data_json)
                result = json.dumps(masked_response, ensure_ascii=False, indent=4)
            except json.JSONDecodeError:
                result = raw_response
        except UnicodeDecodeError:
            result = f"<Binary/Streaming Data (len={data_len})>"

        return result

    def _write_log(self, log_message: str, status_code: int, exc: Exception | None):
        """
        Записывает лог с учетом исключения

        :param log_message: Сообщение лога
        :param status_code: Код статуса ответа
        :param exc: Исключение
        """

        if exc or status_code >= status.HTTP_400_BAD_REQUEST:

            if exc:
                tb_string = ''.join(traceback.format_exception(
                    type(exc),  # тип исключения
                    exc,  # сам объект исключения
                    exc.__traceback__  # объект traceback
                ))
                log_message += f"\n{tb_string}\n"

            if status_code == status.HTTP_400_BAD_REQUEST:
                self.logger.debug(log_message)
            elif status_code == status.HTTP_401_UNAUTHORIZED:
                self.logger.warning(log_message)
            elif status_code == status.HTTP_403_FORBIDDEN:
                self.logger.warning(log_message)
            elif status_code == status.HTTP_404_NOT_FOUND:
                self.logger.error(log_message)
            else:
                self.logger.critical(log_message)

        else:
            self.logger.debug(log_message)

    def _mask_sensitive_data(self, data: Any) -> Any:
        """
        Рекурсивно маскирует конфиденциальные данные в словарях и списках

        :param data: Данные для маскировки

        :return: Маскированные данные
        """

        if isinstance(data, dict):
            return {
                key: '***' if key.lower() in self.sensitive_keys else self._mask_sensitive_data(value)
                for key, value in data.items()
            }
        elif isinstance(data, list):
            return [self._mask_sensitive_data(item) for item in data]

        return data