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
SecurityMiddlewarefirst 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.