Django 和 DRF 的审计日志

Question

我需要在使用Django 1.8 和Django-Rest-Framework 3.2.2 的项目之一中实现审计日志功能。我已扩展 BaseUserManager class 以创建用户模型，因为我必须在我的应用程序中使用电子邮件作为用户名（如果此信息很重要）。

下面是我的数据库设计，它将保存日志：

**fields    type    desc**

id           pk      ( auto_increment)  
cust_id   FK  customer 
customer_name   FK  customer
user_id FK  user
user_name   FK  user
module  Varchar(100) sales,order,billing,etc
action  Varchar(10) Create/Update/Delete
previous_value  varchar(500)    
current_value   varchar(500)    
Datetime    Datetime    timestamp of change

我已经尝试过https://pypi.python.org/pypi/django-audit-log，但根据我的要求，它有 2 个问题-

1. 它没有按照我的要求捕获数据，我理解这是我的问题，因此我修改了它的代码并将我的字段添加到它的模型中。
2. 未捕获模块信息。行为是随机的。

我正在寻求建议以继续使用此功能。哪个包最适合我的任务。

P.S 我也试过Django-reversion，我不需要数据版本。

谢谢

Answer 1

我通过修改审计日志代码实现了我所需要的 -

1. 在审计日志的 LogEntry 模型中添加了必填字段。
2. 修改receives.py的log_create、log_update、log_delete函数，在新增字段中保存信息。

使用它我已经完成了一半。现在我面临的唯一问题是，由于表中使用了 FK，因此 1 个表的模型实例还包含其他表的信息。

为了解决这个问题，我可以想出一个效果很好的解决方案，但我对此并不满意。 我在每个模型中添加了一个类似 include_in_model() 的函数，并修改了 auditlog 的 registry.py register() 函数来获取这些字段并仅使用它来将信息保存在 LogEntry 模型中。

这种方法需要我在每个模型类中创建这个 include_in_model() 函数，并为特定模型传递所需的字段。这样我就避免了 FK 相关信息。

Answer 2

Django Simple History 是我过去在生产项目中使用过的一款出色的应用程序，它会为您提供针对您的用户的每个模型的审核。

此外，您应该创建自己的身份验证类来负责记录请求。假设用户使用令牌向您的 API 进行身份验证。它在每个 HTTP 请求的标头中发送到您的 API，如下所示：Authorization: Bearer <My Token>。然后我们应该记录与请​​求关联的用户、时间、用户的 IP 和正文。

这很简单：

settings.py

REST_FRAMEWORK = {
    'DEFAULT_AUTHENTICATION_CLASSES': (
        'common.authentication.MyTokenAuthenticationClass'
    ),
    ...
}

common/authentication.py

from django.utils import timezone
from django.utils.translation import ugettext_lazy as _

from ipware.ip import get_real_ip
from rest_framework import authentication
from rest_framework import exceptions

from accounts.models import Token, AuditLog


class MyTokenAuthenticationClass(authentication.BaseAuthentication):

    def authenticate(self, request):

        # Grab the Athorization Header from the HTTP Request
        auth = authentication.get_authorization_header(request).split()

        if not auth or auth[0].lower() != b'bearer':
            return None

        # Check that Token header is properly formatted and present, raise errors if not
        if len(auth) == 1:
            msg = _('Invalid token header. No credentials provided.')
            raise exceptions.AuthenticationFailed(msg)
        elif len(auth) > 2:
            msg = _('Invalid token header. Credentials string should not contain spaces.')
            raise exceptions.AuthenticationFailed(msg)

        try:
            token = Token.objects.get(token=auth[1])
            # Using the `ipware.ip` module to get the real IP (if hosted on ElasticBeanstalk or Heroku)
            token.last_ip = get_real_ip(request)
            token.last_login = timezone.now()
            token.save()

            # Add the saved token instance to the request context
            request.token = token

        except Token.DoesNotExist:
            raise exceptions.AuthenticationFailed('Invalid token.')

        # At this point, insert the Log into your AuditLog table:
        AuditLog.objects.create(
            user_id=token.user,
            request_payload=request.body,
            # Additional fields
            ...
        )

        # Return the Authenticated User associated with the Token
        return (token.user, token)

Answer 3

首先你可以使用用户包：https://github.com/jcugat/django-custom-user，来解决电子邮件作为用户名字段。 然后您可以尝试通过以下方式专注于开发：http://django-reversion.readthedocs.io/en/stable/

Answer 4

另一种解决方案是使用django auditlog 并使用自定义中间件，该中间件不会直接捕获“request.user”，但在需要时，此时 DRF 将设置正确的“request.user” ' 以便它不再缺少审核日志中的用户名。

创建一个名为（例如）auditlog_middleware.py 的文件，并将其包含在您的 settings.py 中的 MIDDLEWARE 中，而不是默认的审计日志中间件。

from __future__ import unicode_literals

import threading
import time

from django.conf import settings
from django.db.models.signals import pre_save
from django.utils.functional import curry
from django.apps import apps
from auditlog.models import LogEntry
from auditlog.compat import is_authenticated

# Use MiddlewareMixin when present (Django >= 1.10)
try:
    from django.utils.deprecation import MiddlewareMixin
except ImportError:
    MiddlewareMixin = object


threadlocal = threading.local()


class AuditlogMiddleware(MiddlewareMixin):
    """
    Middleware to couple the request's user to log items. This is accomplished by currying the signal receiver with the
    user from the request (or None if the user is not authenticated).
    """

    def process_request(self, request):
        """
        Gets the current user from the request and prepares and connects a signal receiver with the user already
        attached to it.
        """
        # Initialize thread local storage
        threadlocal.auditlog = {
            'signal_duid': (self.__class__, time.time()),
            'remote_addr': request.META.get('REMOTE_ADDR'),
        }

        # In case of proxy, set 'original' address
        if request.META.get('HTTP_X_FORWARDED_FOR'):
            threadlocal.auditlog['remote_addr'] = request.META.get('HTTP_X_FORWARDED_FOR').split(',')[0]

        # Connect signal for automatic logging
        set_actor = curry(self.set_actor, request=request, signal_duid=threadlocal.auditlog['signal_duid'])
        pre_save.connect(set_actor, sender=LogEntry, dispatch_uid=threadlocal.auditlog['signal_duid'], weak=False)

    def process_response(self, request, response):
        """
        Disconnects the signal receiver to prevent it from staying active.
        """
        if hasattr(threadlocal, 'auditlog'):
            pre_save.disconnect(sender=LogEntry, dispatch_uid=threadlocal.auditlog['signal_duid'])

        return response

    def process_exception(self, request, exception):
        """
        Disconnects the signal receiver to prevent it from staying active in case of an exception.
        """
        if hasattr(threadlocal, 'auditlog'):
            pre_save.disconnect(sender=LogEntry, dispatch_uid=threadlocal.auditlog['signal_duid'])

        return None

    @staticmethod
    def set_actor(request, sender, instance, signal_duid, **kwargs):
        """
        Signal receiver with an extra, required 'user' kwarg. This method becomes a real (valid) signal receiver when
        it is curried with the actor.
        """
        if hasattr(threadlocal, 'auditlog'):
            if not hasattr(request, 'user') or not is_authenticated(request.user):
                return
            if signal_duid != threadlocal.auditlog['signal_duid']:
                return
            try:
                app_label, model_name = settings.AUTH_USER_MODEL.split('.')
                auth_user_model = apps.get_model(app_label, model_name)
            except ValueError:
                auth_user_model = apps.get_model('auth', 'user')
            if sender == LogEntry and isinstance(request.user, auth_user_model) and instance.actor is None:
                instance.actor = request.user

            instance.remote_addr = threadlocal.auditlog['remote_addr']