使用 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 个问题-

它没有按照我的要求捕获数据，我知道这是我的问题，所以我修改了它的代码并将我的字段添加到它的模型中。
它没有捕获模块信息。行为是随机的。

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

P.S我也试过Django-reversion，我没有数据版本控制的要求。

谢谢

Answer 1

首先可以user包：https://github.com/jcugat/django-custom-user，解决Email as Username字段。然后你可以尝试着重开发：http://django-reversion.readthedocs.io/en/stable/

Answer 2

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

在审计日志的 LogEntry 模型中添加了必填字段。
修改了 receivers.py 的 log_create、log_update、log_delete 功能，以在新添加的字段中保存信息。

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

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

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

Answer 3

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

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

这很简单：

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 4

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

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

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']

Answer 5

我知道这个答案来得太晚了，但就这样吧

因为 DRF 在 View 级别而不是 Middleware 级别进行身份验证，所以当 AuditlogMiddleware 运行时用户尚未附加到请求，导致 AnonymousUser

您可以在验证后附上 AuditlogMiddleware 中的逻辑这个逻辑连接了一些信号

此解决方案适合：

你不必用它装饰每个 View
它一般不会对 AuditlogMiddleware 或 audit_log 实现做任何假设。所以如果代码改变，这应该仍然有效
它不会强制或重复 DRF 身份验证。

#token_authentication_wrapper.py
from auditlog.middleware import AuditlogMiddleware
from rest_framework.authentication import TokenAuthentication


class TokenAuthenticationWrapper(TokenAuthentication):
    def authenticate(self, request):
        user, token = super().authenticate(request)
        request.user = user # necessary for preventing recursion
        AuditlogMiddleware().process_request(request)
        return user, token

继承您最喜欢的身份验证服务，例如BasicAuthentication SessionAuthentication, TokenAuthentication, 等等...

并在 setting.py

    'DEFAULT_AUTHENTICATION_CLASSES': [
        'path.to.file.token_authentication_wrapper.TokenAuthenticationWrapper',
    ]

使用 Django 和 DRF 的审计日志

auditlog with Django and DRF

python

django

django-rest-framework

audit-logging

django-auditlog