How to Build Custom Security Middleware

Custom security middleware in Django allows you to enforce application-wide policies such as authentication, rate limiting, CSRF enhancements, and audit logging at the HTTP layer. This guide covers design patterns, implementation, and configuration for robust security middleware.

1. Middleware Workflow in Django

Django middleware wraps the request/response cycle. Each middleware class must implement __init__ and __call__ (or process_view, process_exception hooks) to intercept and modify requests or responses.

2. Authentication Enforcement Middleware

Ensure all API endpoints require authenticated users, except login and static routes:

# middleware/authentication.py
from django.http import JsonResponse
from django.urls import reverse

EXEMPT_URLS = [reverse('login'), reverse('signup')]

class AuthenticationMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        # Allow exempt URLs
        if request.path in EXEMPT_URLS:
            return self.get_response(request)

        # Enforce authentication
        if not request.user.is_authenticated:
            return JsonResponse({'detail': 'Authentication required.'}, status=401)

        return self.get_response(request)

Tip: Use reverse to avoid hard-coding URLs and update exemptions centrally.

3. Per-User Rate Limiting

Throttle requests per user using Django cache (Redis/Memcached):

# middleware/rate_limit.py
import time
from django.core.cache import cache
from django.http import JsonResponse

class RateLimitMiddleware:
    RATE = 100        # max requests
    WINDOW = 60 * 5   # seconds

    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        user = request.user
        if user.is_authenticated:
            key = f"rl:{user.id}"
            count, start = cache.get(key, (0, time.time()))
            now = time.time()
            if now - start > self.WINDOW:
                count, start = 0, now
            count += 1
            cache.set(key, (count, start), timeout=self.WINDOW)
            if count > self.RATE:
                return JsonResponse({'detail': 'Rate limit exceeded.'}, status=429)
        return self.get_response(request)

4. CSRF Protection Enhancement

Add custom checks for high-risk endpoints (e.g., payment processing):

# middleware/csrf_enhance.py
from django.middleware.csrf import CsrfViewMiddleware
from django.http import HttpResponseForbidden

class CsrfEnhanceMiddleware(CsrfViewMiddleware):
    def process_view(self, request, callback, callback_args, callback_kwargs):
        # Use default CSRF validation first
        response = super().process_view(request, callback, callback_args, callback_kwargs)
        if response:
            return response

        # Additional check: require custom header for critical endpoints
        if request.path.startswith('/payments/') and 'X-Auth-Token' not in request.headers:
            return HttpResponseForbidden('Missing X-Auth-Token header.')
        return None

5. Audit Logging Middleware

Log critical request details for compliance and traceability:

# middleware/audit.py
import logging
from django.utils.deprecation import MiddlewareMixin

logger = logging.getLogger('audit')

class AuditLoggingMiddleware(MiddlewareMixin):
    def process_request(self, request):
        # Log request start
        user = request.user.username if request.user.is_authenticated else 'anonymous'
        logger.info(f"Request start: user={user}, path={request.path}, method={request.method}")

    def process_response(self, request, response):
        # Log response completion
        user = request.user.username if request.user.is_authenticated else 'anonymous'
        logger.info(f"Request end: user={user}, path={request.path}, status={response.status_code}")
        return response

Note: Configure your logger in settings.py to write to file or external system (e.g., ELK, Splunk).

6. Middleware Ordering and Configuration

Order middleware carefully in settings.py:

Position	Middleware	Purpose
1	`'django.middleware.security.SecurityMiddleware'`	Core security headers
2	`'middleware.authentication.AuthenticationMiddleware'`	Enforce login
3	`'middleware.rate_limit.RateLimitMiddleware'`	Throttle requests
4	`'middleware.csrf_enhance.CsrfEnhanceMiddleware'`	Enhanced CSRF checks
5	`'middleware.audit.AuditLoggingMiddleware'`	Audit request/response
6	`'django.middleware.common.CommonMiddleware'`	Standard middleware

7. Testing Your Middleware

Write unit tests for each middleware behavior:

# tests/test_middleware.py
from django.test import RequestFactory, TestCase
from middleware.rate_limit import RateLimitMiddleware
from middleware.authentication import AuthenticationMiddleware
from django.contrib.auth.models import AnonymousUser, User

class MiddlewareTestCase(TestCase):
    def setUp(self):
        self.factory = RequestFactory()
        self.user = User.objects.create_user('test', 't@test.com', 'pass')

    def test_authentication_exempt(self):
        req = self.factory.get('/login/')
        req.user = AnonymousUser()
        mw = AuthenticationMiddleware(lambda r: JsonResponse({}))
        resp = mw(req)
        self.assertEqual(resp.status_code, 200)

    def test_authentication_required(self):
        req = self.factory.get('/secure/')
        req.user = AnonymousUser()
        mw = AuthenticationMiddleware(lambda r: JsonResponse({}))
        resp = mw(req)
        self.assertEqual(resp.status_code, 401)

    def test_rate_limit_exceeded(self):
        req = self.factory.get('/api/')
        req.user = self.user
        mw = RateLimitMiddleware(lambda r: JsonResponse({}))
        # Simulate 101 requests
        for _ in range(101):
            resp = mw(req)
        self.assertEqual(resp.status_code, 429)

8. Best Practices & Tips

Keep middleware lightweight to avoid adding latency.
Store rate-limit counters in Redis for distributed deployments.
Use Django’s built-in SecurityMiddleware first for headers like HSTS.
Document middleware order and purpose clearly for maintainability.

Pro Tip: Leverage Django signals (e.g., request_finished) for additional cleanup or analytics outside middleware flow.