Django 和 DRF 的审计日志
Posted
技术标签:
【中文标题】Django 和 DRF 的审计日志【英文标题】：auditlog with Django and DRF 【发布时间】：2017-04-06 00:31:59 【问题描述】：我需要在使用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，我不需要数据版本。
谢谢
【问题讨论】：
【参考方案1】：我通过修改审计日志代码实现了我所需要的 -
-
在审计日志的 LogEntry 模型中添加了必填字段。
修改receives.py的log_create、log_update、log_delete函数，在新增字段中保存信息。
使用它我已经完成了一半。现在我面临的唯一问题是，由于表中使用了 FK，因此 1 个表的模型实例还包含其他表的信息。
为了解决这个问题，我可以想出一个效果很好的解决方案，但我对此并不满意。 我在每个模型中添加了一个类似 include_in_model() 的函数，并修改了 auditlog 的 registry.py register() 函数来获取这些字段并仅使用它来将信息保存在 LogEntry 模型中。
这种方法需要我在每个模型类中创建这个 include_in_model() 函数，并为特定模型传递所需的字段。这样我就避免了 FK 相关信息。
【讨论】：
【参考方案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)
【讨论】：
【参考方案3】：首先你可以使用用户包：https://github.com/jcugat/django-custom-user，来解决电子邮件作为用户名字段。 然后您可以尝试通过以下方式专注于开发：http://django-reversion.readthedocs.io/en/stable/
【讨论】：
感谢您的建议。实际上回归不符合我的要求，我也提到了同样的问题。作为用户名的电子邮件已经完成。感谢您的回复。【参考方案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']
【讨论】：
以上是关于Django 和 DRF 的审计日志的主要内容，如果未能解决你的问题，请参考以下文章