Bài 17: Middleware

Middleware Là Gì?

Middleware là layer xử lý requests/responses giữa web server và Django views.

# Request/Response flow:# 1. Browser → Web Server (nginx, Apache)# 2. Web Server → WSGI Server (Gunicorn, uWSGI)# 3. WSGI Server → Django Middleware (request processing)# 4. Middleware → View (your code)# 5. View → Middleware (response processing)# 6. Middleware → WSGI Server# 7. WSGI Server → Web Server# 8. Web Server → Browser # Middleware can:# - Process requests before they reach views# - Process responses before they're sent to browser# - Short-circuit request (return response without calling view)# - Modify request/response objects# - Handle exceptions # Common uses:# - Authentication# - Session management# - CSRF protection# - Security headers# - Logging# - Request/response modification

Middleware Order

# settings.pyMIDDLEWARE = [    'django.middleware.security.SecurityMiddleware',         # 1. Security    'django.contrib.sessions.middleware.SessionMiddleware',  # 2. Sessions    'django.middleware.common.CommonMiddleware',             # 3. Common    'django.middleware.csrf.CsrfViewMiddleware',            # 4. CSRF    'django.contrib.auth.middleware.AuthenticationMiddleware', # 5. Auth    'django.contrib.messages.middleware.MessageMiddleware',  # 6. Messages    'django.middleware.clickjacking.XFrameOptionsMiddleware', # 7. Clickjacking] # Order matters!# Request: Top to bottom (1→7)# Response: Bottom to top (7→1) # Example flow:# Request:  Security → Session → Common → CSRF → Auth → Messages → View# Response: View → Messages → Auth → CSRF → Common → Session → Security

Built-in Middleware

SecurityMiddleware

# Django Security Middleware'django.middleware.security.SecurityMiddleware' # Provides:# - HTTPS redirect# - Strict-Transport-Security header# - Content security headers# - Browser XSS protection # settings.pySECURE_SSL_REDIRECT = True  # Redirect HTTP to HTTPSSECURE_HSTS_SECONDS = 31536000  # 1 yearSECURE_HSTS_INCLUDE_SUBDOMAINS = TrueSECURE_HSTS_PRELOAD = TrueSECURE_CONTENT_TYPE_NOSNIFF = TrueSECURE_BROWSER_XSS_FILTER = True

SessionMiddleware

# Session management'django.contrib.sessions.middleware.SessionMiddleware' # Enables request.session# Must come before AuthenticationMiddleware # Example usage:def my_view(request):    request.session['key'] = 'value'    value = request.session.get('key')

CommonMiddleware

# Common operations'django.middleware.common.CommonMiddleware' # Provides:# - URL normalization (trailing slash)# - ETags# - Content-Length header# - Forbidden User-Agent strings # settings.pyAPPEND_SLASH = True  # Add trailing slash to URLsPREPEND_WWW = False  # Add www. to domainDISALLOWED_USER_AGENTS = []  # Block user agents

CsrfViewMiddleware

# CSRF protection'django.middleware.csrf.CsrfViewMiddleware' # Protects against Cross-Site Request Forgery# Requires {% csrf_token %} in forms # In views:def my_view(request):    # POST requests checked for CSRF token    if request.method == 'POST':        # Process form        pass # Exempt view from CSRFfrom django.views.decorators.csrf import csrf_exempt @csrf_exemptdef api_view(request):    # No CSRF check    pass

AuthenticationMiddleware

# User authentication'django.contrib.auth.middleware.AuthenticationMiddleware' # Adds request.user attribute# Requires SessionMiddleware # Usage in views:def my_view(request):    if request.user.is_authenticated:        username = request.user.username    else:        username = 'Anonymous'

MessageMiddleware

# Flash messages'django.contrib.messages.middleware.MessageMiddleware' # Enables django.contrib.messagesfrom django.contrib import messages def my_view(request):    messages.success(request, 'Operation successful')

ClickjackingMiddleware

# Clickjacking protection'django.middleware.clickjacking.XFrameOptionsMiddleware' # Adds X-Frame-Options header# Prevents page from being loaded in iframe # settings.pyX_FRAME_OPTIONS = 'DENY'  # or 'SAMEORIGIN'

Creating Custom Middleware

Function-Based Middleware (Old Style)

# middleware.pydef simple_middleware(get_response):    """    One-time configuration and initialization.    """    print("Middleware initialized")        def middleware(request):        # Code executed for each request before view        print(f"Request: {request.path}")                # Call next middleware or view        response = get_response(request)                # Code executed for each response after view        print(f"Response status: {response.status_code}")                return response        return middleware # settings.pyMIDDLEWARE = [    # ...    'myapp.middleware.simple_middleware',]

Class-Based Middleware (New Style)

# middleware.pyclass SimpleMiddleware:    def __init__(self, get_response):        """One-time configuration"""        self.get_response = get_response        print("Middleware initialized")        def __call__(self, request):        """Called for each request"""        # Before view        print(f"Before view: {request.path}")                # Process request        response = self.get_response(request)                # After view        print(f"After view: {response.status_code}")                return response # settings.pyMIDDLEWARE = [    # ...    'myapp.middleware.SimpleMiddleware',]

Middleware with All Hooks

class FullMiddleware:    def __init__(self, get_response):        self.get_response = get_response        def __call__(self, request):        # Process request before view        response = self.get_response(request)        # Process response after view        return response        def process_view(self, request, view_func, view_args, view_kwargs):        """        Called just before Django calls the view.        Return None to continue processing or HttpResponse to short-circuit.        """        print(f"About to call view: {view_func.__name__}")        return None        def process_exception(self, request, exception):        """        Called when view raises exception.        Return None or HttpResponse.        """        print(f"Exception occurred: {exception}")        return None        def process_template_response(self, request, response):        """        Called after view returns TemplateResponse.        Must return TemplateResponse.        """        print("Processing template response")        return response

Practical Middleware Examples

Request Logging Middleware

import loggingimport time logger = logging.getLogger(__name__) class RequestLoggingMiddleware:    def __init__(self, get_response):        self.get_response = get_response        def __call__(self, request):        # Log request        start_time = time.time()                logger.info(f"Request: {request.method} {request.path}")        logger.info(f"User: {request.user}")        logger.info(f"IP: {self.get_client_ip(request)}")                # Process request        response = self.get_response(request)                # Log response        duration = time.time() - start_time        logger.info(f"Response: {response.status_code}")        logger.info(f"Duration: {duration:.2f}s")                return response        def get_client_ip(self, 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

Request Timing Middleware

import time class TimingMiddleware:    def __init__(self, get_response):        self.get_response = get_response        def __call__(self, request):        # Start timer        request.start_time = time.time()                # Process request        response = self.get_response(request)                # Calculate duration        duration = time.time() - request.start_time                # Add header        response['X-Request-Duration'] = f"{duration:.4f}s"                # Log slow requests        if duration > 1.0:            print(f"SLOW REQUEST: {request.path} took {duration:.2f}s")                return response

User Agent Blocking Middleware

from django.http import HttpResponseForbidden class BlockUserAgentMiddleware:    def __init__(self, get_response):        self.get_response = get_response        self.blocked_agents = [            'bot',            'crawler',            'spider',            'scraper',        ]        def __call__(self, request):        user_agent = request.META.get('HTTP_USER_AGENT', '').lower()                # Check if blocked        for blocked in self.blocked_agents:            if blocked in user_agent:                return HttpResponseForbidden('Access denied')                return self.get_response(request)

Custom Headers Middleware

class CustomHeadersMiddleware:    def __init__(self, get_response):        self.get_response = get_response        def __call__(self, request):        response = self.get_response(request)                # Add custom headers        response['X-Powered-By'] = 'Django'        response['X-Version'] = '1.0.0'        response['X-Developer'] = 'Your Name'                # Add CORS headers        response['Access-Control-Allow-Origin'] = '*'        response['Access-Control-Allow-Methods'] = 'GET, POST, PUT, DELETE'                return response

Maintenance Mode Middleware

from django.http import HttpResponsefrom django.conf import settings class MaintenanceModeMiddleware:    def __init__(self, get_response):        self.get_response = get_response        def __call__(self, request):        # Check maintenance mode        if getattr(settings, 'MAINTENANCE_MODE', False):            # Allow staff users            if request.user.is_authenticated and request.user.is_staff:                pass            else:                return HttpResponse(                    '<h1>Site Under Maintenance</h1>'                    '<p>We will be back soon!</p>',                    status=503                )                return self.get_response(request) # settings.pyMAINTENANCE_MODE = False  # Set to True for maintenance

IP Whitelist Middleware

from django.http import HttpResponseForbiddenfrom django.conf import settings class IPWhitelistMiddleware:    def __init__(self, get_response):        self.get_response = get_response        self.whitelist = getattr(settings, 'IP_WHITELIST', [])        def __call__(self, request):        # Get client IP        ip = self.get_client_ip(request)                # Check whitelist        if self.whitelist and ip not in self.whitelist:            return HttpResponseForbidden(                f'Access denied for IP: {ip}'            )                return self.get_response(request)        def get_client_ip(self, 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 # settings.pyIP_WHITELIST = ['127.0.0.1', '192.168.1.1']

Request Rate Limiting Middleware

from django.core.cache import cachefrom django.http import HttpResponse class RateLimitMiddleware:    def __init__(self, get_response):        self.get_response = get_response        self.rate_limit = 100  # requests per minute        self.period = 60  # seconds        def __call__(self, request):        # Get client identifier        ip = self.get_client_ip(request)        cache_key = f'rate_limit:{ip}'                # Get current count        requests = cache.get(cache_key, 0)                if requests >= self.rate_limit:            return HttpResponse(                'Rate limit exceeded. Please try again later.',                status=429            )                # Increment count        cache.set(cache_key, requests + 1, self.period)                response = self.get_response(request)                # Add rate limit headers        response['X-RateLimit-Limit'] = str(self.rate_limit)        response['X-RateLimit-Remaining'] = str(self.rate_limit - requests - 1)                return response        def get_client_ip(self, 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

Authentication Token Middleware

from django.contrib.auth import get_user_modelfrom django.http import JsonResponse User = get_user_model() class TokenAuthMiddleware:    def __init__(self, get_response):        self.get_response = get_response        def __call__(self, request):        # Check for API token        token = request.META.get('HTTP_AUTHORIZATION', '')                if token.startswith('Token '):            token = token[6:]  # Remove 'Token ' prefix                        try:                # Validate token and get user                user = self.get_user_from_token(token)                request.user = user            except:                return JsonResponse(                    {'error': 'Invalid token'},                    status=401                )                return self.get_response(request)        def get_user_from_token(self, token):        # Implement token validation        # Example: check database, verify signature, etc.        pass

Response Modification Middleware

class ResponseModificationMiddleware:    def __init__(self, get_response):        self.get_response = get_response        def __call__(self, request):        response = self.get_response(request)                # Modify HTML responses only        if response.get('Content-Type', '').startswith('text/html'):            # Add content to response            footer = b'<!-- Generated by Django -->'            response.content = response.content + footer                        # Update Content-Length            response['Content-Length'] = len(response.content)                return response

Exception Handling Middleware

import logging logger = logging.getLogger(__name__) class ExceptionHandlingMiddleware:    def __init__(self, get_response):        self.get_response = get_response        def __call__(self, request):        return self.get_response(request)        def process_exception(self, request, exception):        """Handle exceptions"""        # Log exception        logger.error(            f"Exception in {request.path}",            exc_info=exception        )                # Send email to admins        if not settings.DEBUG:            self.notify_admins(request, exception)                # Return custom error page        return render(request, 'errors/500.html', {            'exception': str(exception)        }, status=500)        def notify_admins(self, request, exception):        from django.core.mail import mail_admins        subject = f'Error on {request.path}'        message = f'Exception: {exception}\nUser: {request.user}'        mail_admins(subject, message)

Middleware Best Practices

Performance Considerations

class PerformantMiddleware:    def __init__(self, get_response):        self.get_response = get_response                # Do expensive operations once at startup        self.config = self.load_config()        def __call__(self, request):        # Keep request processing fast        # Avoid database queries if possible        # Use caching                response = self.get_response(request)                return response        def load_config(self):        """Load config once at startup"""        return {}

Conditional Middleware

class ConditionalMiddleware:    def __init__(self, get_response):        self.get_response = get_response        def __call__(self, request):        # Skip middleware for certain paths        if request.path.startswith('/api/'):            return self.get_response(request)                # Skip for static files        if request.path.startswith('/static/'):            return self.get_response(request)                # Process only for specific conditions        response = self.get_response(request)                return response

Bài Tập

Bài 1: Request Logger

Task: Create comprehensive request logging:

# Requirements:# 1. Log all requests with:#    - Timestamp#    - Method (GET, POST, etc.)#    - Path#    - User (if authenticated)#    - IP address#    - User agent#    - Response status code#    - Response time # 2. Save logs to:#    - File (requests.log)#    - Database (RequestLog model) # 3. Features:#    - Exclude static files#    - Exclude admin paths (optional)#    - Log request body for POST requests#    - Log query parameters#    - Anonymize sensitive data (passwords)

Bài 2: API Rate Limiter

Task: Implement API rate limiting:

# Requirements:# 1. Rate limits:#    - Anonymous: 10 requests/minute#    - Authenticated: 100 requests/minute#    - Premium users: 1000 requests/minute # 2. Use Redis for counting# 3. Return 429 status when exceeded# 4. Add headers:#    - X-RateLimit-Limit#    - X-RateLimit-Remaining#    - X-RateLimit-Reset # 5. Different limits per endpoint# 6. Whitelist certain IPs# 7. Admin interface to view rate limits

Bài 3: Security Middleware

Task: Create security middleware:

# Features:# 1. Block suspicious requests:#    - SQL injection attempts#    - XSS attempts#    - Path traversal attempts # 2. IP blocking:#    - Maintain blacklist#    - Auto-block after X failed attempts#    - Temporary bans # 3. User agent filtering:#    - Block known bots#    - Require valid user agent # 4. Referrer checking:#    - Block requests from suspicious domains # 5. Security headers:#    - CSP (Content Security Policy)#    - HSTS#    - X-Content-Type-Options

Bài 4: Multi-tenant Middleware

Task: Implement multi-tenancy:

# Requirements:# 1. Detect tenant from:#    - Subdomain (tenant1.example.com)#    - Or path (example.com/tenant1/)#    - Or custom header (X-Tenant-ID) # 2. Set tenant context:#    - request.tenant attribute#    - Filter all queries by tenant # 3. Database per tenant:#    - Route to correct database#    - Or use shared database with tenant_id # 4. Tenant-specific:#    - Templates (themes)#    - Settings#    - Static files # 5. Handle invalid tenant gracefully

Tài Liệu Tham Khảo

Previous: Bài 16: Messages Framework | Next: Bài 18: Settings and Configuration

#hooks

Django Middleware