Implementing an Audit Trail Middleware in Django for Tracking User Actions in Django
Table of contents
This is part 2 of the series create and use custom signals in Django.
In this article, you will do some updates on the previously written code in the previous article and learn how to wrap the signal in middleware for it to be automatically fired up with every request.
The base application for this blog post is hosted on this github repository .
1 Update Signal definitions
In this section, you will update some deprecated methods in the code base. Head over to the `apps/blogs/signals.py` file and remove this block:
audit_trail_signal = django.dispatch.Signal(
providing_args=['user', 'request', 'model', 'event_category', 'method', 'summary']
)
remove the providing_args argument from the django.dispatch.Signal method. The updated file should look like the one below.
import django.dispatch
from django.dispatch import receiver
from .models import AuditTrail
import logging
import datetime
# creates a custom signal and specifies the args required.
audit_trail_signal = django.dispatch.Signal(
# providing_args=['user', 'request', 'model', 'event_category', 'method', 'summary']
)
logger = logging.getLogger(__name__)
# helper func that gets the client ip
def get_client_ip(request):
x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
if x_forwarded_for:
ip = x_forwarded_for.split(',')[0]
else:
ip = request.META.get('REMOTE_ADDR')
return ip
@receiver(audit_trail_signal)
def log_audit_trail(sender, user, request, model, event_category, method, summary, **kwargs):
try:
user_agent_info = request.META.get(
'HTTP_USER_AGENT', '<unknown>')[:255],
print(summary)
auditTrail = AuditTrail.objects.create(
user=user,
user_agent_info=user_agent_info,
changed_object=model,
event_category=event_category,
login_IP=get_client_ip(request),
is_deleted=False,
action=method,
change_summary=summary
)
logger.info(
f"Audit trail created {auditTrail.id} for user {auditTrail.username} and object {auditTrail.changed_object}")
except Exception as e:
logger.error("log_user_logged_in request: %s, error: %s" %
(request, e))
2. Custom AuditTrailMiddleware.
In this section, you will learn how to set up a custom middleware, why it is needed and its importance.
As per the official django documentation, a middleware is a framework of hooks into Django's request/response processing. It’s a light, low-level “plugin” system for globally altering Django’s input or output.
With you now understanding what middleware is, you can go ahead and create a custom AuditTrailMiddleware.
The AuditTrailMiddleware will intercept the response to every API call and trigger the audit_trail signal that you had earlier created for every request whether successful or error thrown with the corresponding information.
Head over to your idea and at the apps/blogs level create a new file middlewares.py and paste the following lines of code.
from django.utils.deprecation import MiddlewareMixin
from .signals import audit_trail_signal
import traceback
class AuditTrailMiddleware(MiddlewareMixin):
def process_response(self, request, response):
if response.status_code >= 200 and response.status_code < 300:
# Get the view class from the resolver match info
# view_class = request.resolver_match.func.cls
view_class = None
if hasattr(request.resolver_match.func, 'cls'):
view_class = request.resolver_match.func.cls
# Get the model from the view's queryset attribute
# model = view_class.queryset.model.__name__
model = None
if hasattr(view_class, 'queryset'):
model = view_class.queryset.model.__name__
audit_trail_signal.send(
sender=request.user.__class__,
request=request,
response=response,
user=request.user,
model=model,
event_category=model,
method=request.method,
summary="{} {}".format(request.method, request.path),
)
return response
else:
# An error occurred, send the audit trail signal with an error message
# Get the view class and model as described above
view_class = None
if hasattr(request.resolver_match.func, 'cls'):
view_class = request.resolver_match.func.cls
elif hasattr(request.resolver_match.func, 'view_class'):
view_class = request.resolver_match.func.view_class
model = None
if view_class is not None and hasattr(view_class, 'queryset'):
model = view_class.queryset.model.__name__
# Build the error message from the traceback
error_message = traceback.format_exc()
audit_trail_signal.send(
sender=request.user.__class__,
request=request,
response=response,
user=request.user,
model=model,
event_category="Blog",
method=request.method,
summary="{} {} - ERROR".format(request.method, request.path),
detail=error_message,
)
return response
def process_exception(self, request, exception):
# An exception occurred, send the audit trail signal with an error message
# Get the view class and model as described above
view_class = None
if hasattr(request.resolver_match.func, 'cls'):
view_class = request.resolver_match.func.cls
elif hasattr(request.resolver_match.func, 'view_class'):
view_class = request.resolver_match.func.view_class
model = None
if view_class is not None and hasattr(view_class, 'queryset'):
model = view_class.queryset.model.__name__
# Build the error message from the traceback
error_message = traceback.format_exc()
audit_trail_signal.send(
sender=request.user.__class__,
request=request,
response=None,
user=request.user,
model=model,
event_category="Blog",
method=request.method,
summary="{} {} - ERROR".format(request.method, request.path),
detail=error_message,
)
With the AuditTrailMiddleware implementation in place, it needs to be registered for it to work thus head over to your settings file, the middleware definitions and add in the AuditTraiMiddleware at the end of the list as shown below
MIDDLEWARE = [
...,
'apps.blogs.middlewares.AuditTrailMiddleware' # path to your middleware class
]
With this in place, the audit trail signal will be automatically triggered for every request-response cycle.
Head over to your apps/blogs/views/py and comment out the line that calls the audit trail signal on the PostAPI class.
class PostAPI(generics.ListCreateAPIView):
serializer_class = PostSerializer
queryset = Post.objects.all().order_by("-created_at")
permission_classes = (permissions.IsAuthenticated,)
def perform_create(self, serializer):
data = self.request.data
new_post = serializer.save(
user=self.request.user, title=data["title"], content=data["content"]
)
all_posts = Post.objects.all().order_by("-created_at")
serializer = PostSerializer(all_posts, many=True)
# audit_trail_signal.send(sender=self.request.user.__class__, request=self.request,
# user=self.request.user, model="Blog", event_category="Blog", method="CREATE", summary="Create a new post")
return Response(serializer.data, status=status.HTTP_201_CREATED)
To test this exhaustively, re-start your development server and head over to the swagger endpoint and hit any posts / tags API endpoints.
http://localhost:8000/swagger/
With this, you have learnt on how to set up a custom Django middleware and use it to track user actions via an audit trail.
Happy hacking.