Auth Middleware — Complete Backend Implementation Guide

DodaTech Updated 2026-06-28 6 min read

In this tutorial, you will learn about Auth Middleware. We cover key concepts, practical examples, and best practices to help you master this topic.

Authentication middleware intercepts incoming requests before they reach route handlers, verifying the user's identity, extracting user information, and attaching it to the request context for downstream use by authorization logic and business logic.

What You'll Learn

By the end of this lesson, you will implement JWT verification middleware, session auth middleware, role-based access control, route protection patterns, and composable middleware chains for Express, Django, and FastAPI.

Why It Matters

Centralizing authentication in middleware eliminates code duplication across routes, ensures consistent security enforcement, and makes it easy to add or modify auth logic without touching route handlers. Every DodaTech backend service uses a standardized auth middleware pipeline.

Real-World Use

An Express API has 50 route handlers. Without auth middleware, each handler must verify the JWT, check expiration, and extract user data. With middleware, a single authenticateJWT function handles all of this, and any route that needs auth simply registers the middleware.

Auth Middleware Architecture

flowchart LR
    REQ[Request] --> A[Extract Token]
    A --> B[Verify Token]
    B -->|Valid| C[Attach User to req]
    C --> D[Check Authorization]
    D -->|Permitted| E[Route Handler]
    B -->|Invalid| F[401 Response]
    D -->|Denied| G[403 Response]
    style A fill:#f90,color:#fff

JWT Auth Middleware (Express)

const jwt = require("jsonwebtoken");
const express = require("express");

const app = express();
const ACCESS_SECRET = process.env.ACCESS_SECRET || "your-secret";

function authenticate(req, res, next) {
  const authHeader = req.headers.authorization;

  if (!authHeader || !authHeader.startsWith("Bearer ")) {
    console.log(`[Auth] No token provided for ${req.path}`);
    return res.status(401).json({
      error: "Authentication required",
      code: "TOKEN_MISSING",
    });
  }

  const token = authHeader.split(" ")[1];

  try {
    const decoded = jwt.verify(token, ACCESS_SECRET, {
      algorithms: ["HS256", "RS256"],
    });

    req.user = {
      id: decoded.sub,
      email: decoded.email,
      role: decoded.role || "user",
      permissions: decoded.permissions || [],
    };

    console.log(`[Auth] User ${req.user.id} (${req.user.role}) accessing ${req.path}`);
    next();
  } catch (err) {
    if (err instanceof jwt.TokenExpiredError) {
      console.log(`[Auth] Expired token for ${req.path}`);
      return res.status(401).json({
        error: "Token expired",
        code: "TOKEN_EXPIRED",
      });
    }
    console.log(`[Auth] Invalid token: ${err.message}`);
    return res.status(403).json({
      error: "Invalid token",
      code: "TOKEN_INVALID",
    });
  }
}

function requireRole(...roles) {
  return (req, res, next) => {
    if (!req.user) {
      return res.status(401).json({ error: "Authentication required" });
    }
    if (!roles.includes(req.user.role)) {
      console.log(`[Auth] Role denied: ${req.user.role} not in ${roles}`);
      return res.status(403).json({ error: "Insufficient permissions" });
    }
    next();
  };
}

function requirePermission(permission) {
  return (req, res, next) => {
    if (!req.user) {
      return res.status(401).json({ error: "Authentication required" });
    }
    if (!req.user.permissions.includes(permission)) {
      return res.status(403).json({ error: `Missing permission: ${permission}` });
    }
    next();
  };
}

// Usage
app.get("/api/public", (req, res) => {
  res.json({ message: "No auth required" });
});

app.get("/api/profile", authenticate, (req, res) => {
  res.json({ user: req.user });
});

app.get("/api/admin", authenticate, requireRole("admin"), (req, res) => {
  res.json({ message: "Admin access granted" });
});

app.post("/api/orders", authenticate, requirePermission("orders:write"), (req, res) => {
  res.json({ message: "Order created" });
});

app.listen(3000);

Auth Middleware (FastAPI Python)

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt

app = FastAPI()
security = HTTPBearer(auto_error=False)
ACCESS_SECRET = "your-secret"

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
):
    if not credentials:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Authentication required",
        )

    try:
        payload = jwt.decode(
            credentials.credentials,
            ACCESS_SECRET,
            algorithms=["HS256"],
        )
        return {
            "id": payload["sub"],
            "role": payload.get("role", "user"),
            "permissions": payload.get("permissions", []),
        }
    except jwt.ExpiredSignatureError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token expired",
        )
    except jwt.InvalidTokenError:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Invalid token",
        )

def require_role(required_role: str):
    async def role_checker(user: dict = Depends(get_current_user)):
        if user["role"] != required_role:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"Role {required_role} required",
            )
        return user
    return role_checker

@app.get("/api/profile")
async def profile(user: dict = Depends(get_current_user)):
    return {"user": user}

@app.get("/api/admin")
async def admin(user: dict = Depends(require_role("admin"))):
    return {"message": "Admin access"}

Django Auth Middleware

# middleware.py
import jwt
from django.http import JsonResponse
from django.conf import settings

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

    def __call__(self, request):
        auth_header = request.META.get("HTTP_AUTHORIZATION", "")

        if auth_header.startswith("Bearer "):
            token = auth_header[7:]
            try:
                payload = jwt.decode(
                    token,
                    settings.JWT_SECRET,
                    algorithms=["HS256"],
                )
                request.user_id = payload["sub"]
                request.user_role = payload.get("role", "user")
            except jwt.ExpiredSignatureError:
                return JsonResponse({"error": "Token expired"}, status=401)
            except jwt.InvalidTokenError:
                return JsonResponse({"error": "Invalid token"}, status=403)

        response = self.get_response(request)
        return response

class AdminRequiredMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
        self.admin_paths = ["/admin/", "/api/admin/"]

    def __call__(self, request):
        for path in self.admin_paths:
            if request.path.startswith(path):
                role = getattr(request, "user_role", None)
                if role != "admin":
                    return JsonResponse(
                        {"error": "Admin access required"}, status=403
                    )
        return self.get_response(request)

# settings.py
# MIDDLEWARE = [
#     'middleware.JWTAuthMiddleware',
#     'middleware.AdminRequiredMiddleware',
#     ...
# ]

Common Mistakes

Not returning consistent error structures for auth failures makes client-side error handling difficult.
Running auth middleware after body Parsing wastes resources on unauthenticated requests.
Putting auth middleware inside route groups inconsistently, leaving some routes unprotected.
Including auth middleware on login and register endpoints, creating a chicken-and-egg problem.
Not distinguishing between missing token (401) and invalid permissions (403).
Using the same middleware for both authentication and authorization, making them hard to test independently.

Practice Questions

What is the difference between authentication and authorization middleware?

Authentication middleware verifies identity (who is this?). Authorization middleware checks permissions (what can they do?). They should be separate middleware functions, with auth running first.

How does the middleware chain order affect authentication?

Auth middleware should run early in the chain, before body parsing, logging, and Rate Limiting. This rejects unauthenticated requests quickly without wasting resources on processing.

Why should login endpoints exclude auth middleware?

Login endpoints must be accessible without authentication by definition. The middleware would reject the login request because there is no token yet. Public routes should be explicitly marked.

Challenge: Build a middleware chain with JWT verification, role-based access (admin/editor/user), permission-based access (resource:action), per-route middleware configuration, and consistent error responses with standard error codes.

FAQ

How do I handle optional authentication in middleware?

Some routes work both with and without authentication (public content + personalized recommendations). Create an optional auth middleware that attaches user info if available but does not reject the request if missing.

How do I test auth middleware in isolation?

Extract middleware as a pure function that takes (req, res, next) and mock the dependencies. Test: valid token, expired token, missing token, invalid signature, wrong algorithm, and token with missing claims.

Can auth middleware cache token verification?

Cache the JWKS response (public keys) with a 1-hour TTL. Do not cache token verification results — tokens must be verified on every request to detect revocation. Use Redis for the JWKS cache.

How do I handle token refresh in middleware?

The refresh endpoint should be public (no auth middleware). The refresh middleware should verify the refresh token and issue new access tokens. The auth middleware only handles access tokens.

Mini Project: Auth Middleware Tester

Build a CLI tool that simulates a middleware pipeline and tests authentication flows with different token scenarios.

import jwt
import time
import sys

class MiddlewarePipeline:
    def __init__(self):
        self.secret = "test-secret"
        self.middlewares = []

    def use(self, middleware):
        self.middlewares.append(middleware)
        return self

    def process(self, request):
        context = {"request": request, "response": None}
        chain = iter(self.middlewares)

        def next_middleware():
            try:
                mw = next(chain)
                mw(context, next_middleware)
            except StopIteration:
                pass

        next_middleware()
        return context.get("response") or {"status": 200, "body": "OK"}

def auth_middleware(context, next_mw):
    request = context["request"]
    token = request.get("headers", {}).get("authorization", "").replace("Bearer ", "")

    if not token:
        context["response"] = {"status": 401, "body": {"error": "No token"}}
        return

    try:
        payload = jwt.decode(token, "test-secret", algorithms=["HS256"])
        context["user"] = payload
        next_mw()
    except jwt.ExpiredSignatureError:
        context["response"] = {"status": 401, "body": {"error": "Expired token"}}
    except jwt.InvalidTokenError:
        context["response"] = {"status": 403, "body": {"error": "Invalid token"}}

def role_middleware(*roles):
    def mw(context, next_mw):
        user = context.get("user", {})
        if user.get("role") not in roles:
            context["response"] = {"status": 403, "body": {"error": "Insufficient role"}}
            return
        next_mw()
    return mw

pipeline = MiddlewarePipeline()
pipeline.use(auth_middleware)
pipeline.use(role_middleware("admin"))

token = jwt.encode({"sub": "1", "role": "admin", "exp": int(time.time()) + 300}, "test-secret")
result = pipeline.process({"headers": {"authorization": f"Bearer {token}"}})
print(f"Admin access: {result}")

token = jwt.encode({"sub": "2", "role": "user", "exp": int(time.time()) + 300}, "test-secret")
result = pipeline.process({"headers": {"authorization": f"Bearer {token}"}})
print(f"User access: {result}")

What's Next

Learn about security headers for web application security, then build the authentication project to tie everything together.

← Previous CSRF Protection — Complete Implementation Guide Next → Security Headers for Authentication — Complete Implementation Guide

Built by the developers of DodaTech

Doda Browser, DodaZIP & Durga Antivirus Pro

Home Browse Authentication Patterns