JWT vs Session Tokens: Which to Use

eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiaWF0IjoxNTE2MjM5MDIyfQ.SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV_adQssw5c

├─ Header (algorithm, type)
├─ Payload (claims: user_id, exp, iat, etc.)
└─ Signature (HMAC or RSA)

1. User logs in → Server generates JWT
2. Client stores JWT (localStorage/cookie)
3. Client sends JWT in Authorization header
4. Server validates signature + expiration
5. Server extracts user_id from payload
   ✅ No database lookup needed

from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import jwt, JWTError
from datetime import datetime, timedelta
import os

app = FastAPI()
security = HTTPBearer()

SECRET_KEY = os.getenv("JWT_SECRET_KEY")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 15

def create_access_token(user_id: int) -> str:
    """Generate JWT access token."""
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    payload = {
        "sub": str(user_id),
        "exp": expire,
        "iat": datetime.utcnow(),
        "type": "access",
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)) -> int:
    """Validate JWT and extract user_id."""
    try:
        token = credentials.credentials
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        
        user_id = int(payload.get("sub"))
        token_type = payload.get("type")
        
        if user_id is None or token_type != "access":
            raise HTTPException(status_code=401, detail="Invalid token")
        
        return user_id
    
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

@app.post("/login")
async def login(username: str, password: str):
    user_id = 123
    
    access_token = create_access_token(user_id)
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/protected")
async def protected_route(user_id: int = Depends(get_current_user)):
    return {"message": f"Hello user {user_id}"}

package main

import (
    "net/http"
    "time"
    "github.com/golang-jwt/jwt/v5"
    "github.com/gin-gonic/gin"
)

var jwtSecret = []byte("your-secret-key")

type Claims struct {
    UserID int    `json:"sub"`
    Type   string `json:"type"`
    jwt.RegisteredClaims
}

func GenerateJWT(userID int) (string, error) {
    claims := Claims{
        UserID: userID,
        Type:   "access",
        RegisteredClaims: jwt.RegisteredClaims{
            ExpiresAt: jwt.NewNumericDate(time.Now().Add(15 * time.Minute)),
            IssuedAt:  jwt.NewNumericDate(time.Now()),
        },
    }
    
    token := jwt.NewWithClaims(jwt.SigningMethodHS256, claims)
    return token.SignedString(jwtSecret)
}

func AuthMiddleware() gin.HandlerFunc {
    return func(c *gin.Context) {
        tokenString := c.GetHeader("Authorization")
        if tokenString == "" {
            c.AbortWithStatusJSON(401, gin.H{"error": "Unauthorized"})
            return
        }
        
        // Remove "Bearer " prefix
        tokenString = tokenString[7:]
        
        token, err := jwt.ParseWithClaims(tokenString, &Claims{}, func(token *jwt.Token) (interface{}, error) {
            return jwtSecret, nil
        })
        
        if err != nil || !token.Valid {
            c.AbortWithStatusJSON(401, gin.H{"error": "Invalid token"})
            return
        }
        
        claims := token.Claims.(*Claims)
        c.Set("user_id", claims.UserID)
        c.Next()
    }
}

func main() {
    r := gin.Default()
    
    r.POST("/login", func(c *gin.Context) {
        // Verify credentials
        userID := 123
        
        token, _ := GenerateJWT(userID)
        c.JSON(200, gin.H{"access_token": token})
    })
    
    r.GET("/protected", AuthMiddleware(), func(c *gin.Context) {
        userID := c.GetInt("user_id")
        c.JSON(200, gin.H{"message": "Hello", "user_id": userID})
    })
    
    r.Run(":8000")
}

1. User logs in → Server generates random token
2. Server stores token + user_id in Redis/DB
3. Client stores token in cookie
4. Client sends cookie on every request
5. Server looks up session in Redis
   ✅ Instant revocation: delete session row

from fastapi import FastAPI, Depends, HTTPException, Response, Cookie
from redis.asyncio import Redis
import secrets

app = FastAPI()
redis = Redis.from_url("redis://localhost")

SESSION_EXPIRE_SECONDS = 3600  # 1 hour

async def create_session(user_id: int) -> str:
    """Create session and store in Redis."""
    session_token = secrets.token_urlsafe(32)
    
    await redis.setex(
        f"session:{session_token}",
        SESSION_EXPIRE_SECONDS,
        str(user_id)
    )
    
    return session_token

async def get_current_user(session_token: str = Cookie(None)) -> int:
    """Validate session token and get user_id."""
    if not session_token:
        raise HTTPException(status_code=401, detail="Not authenticated")
    
    user_id = await redis.get(f"session:{session_token}")
    if not user_id:
        raise HTTPException(status_code=401, detail="Invalid or expired session")
    
    # Extend session on activity
    await redis.expire(f"session:{session_token}", SESSION_EXPIRE_SECONDS)
    
    return int(user_id)

@app.post("/login")
async def login(username: str, password: str, response: Response):
    # Verify credentials
    user_id = 123
    
    session_token = await create_session(user_id)
    
    response.set_cookie(
        key="session_token",
        value=session_token,
        httponly=True,
        secure=True,
        samesite="lax",
        max_age=SESSION_EXPIRE_SECONDS,
    )
    
    return {"message": "Logged in"}

@app.post("/logout")
async def logout(session_token: str = Cookie(None)):
    """Revoke session immediately."""
    if session_token:
        await redis.delete(f"session:{session_token}")
    return {"message": "Logged out"}

@app.get("/protected")
async def protected_route(user_id: int = Depends(get_current_user)):
    return {"message": f"Hello user {user_id}"}

package main

import (
    "context"
    "crypto/rand"
    "encoding/base64"
    "net/http"
    "strconv"
    "time"
    
    "github.com/gin-gonic/gin"
    "github.com/redis/go-redis/v9"
)

var ctx = context.Background()
var rdb = redis.NewClient(&redis.Options{
    Addr: "localhost:6379",
})

func generateSessionToken() (string, error) {
    b := make([]byte, 32)
    _, err := rand.Read(b)
    if err != nil {
        return "", err
    }
    return base64.URLEncoding.EncodeToString(b), nil
}

func createSession(userID int) (string, error) {
    token, err := generateSessionToken()
    if err != nil {
        return "", err
    }
    
    err = rdb.Set(ctx, "session:"+token, userID, 1*time.Hour).Err()
    return token, err
}

func getSessionUserID(token string) (int, error) {
    val, err := rdb.Get(ctx, "session:"+token).Result()
    if err != nil {
        return 0, err
    }
    
    // Extend session
    rdb.Expire(ctx, "session:"+token, 1*time.Hour)
    
    return strconv.Atoi(val)
}

func authMiddleware() gin.HandlerFunc {
    return func(c *gin.Context) {
        token, err := c.Cookie("session_token")
        if err != nil {
            c.AbortWithStatusJSON(401, gin.H{"error": "Not authenticated"})
            return
        }
        
        userID, err := getSessionUserID(token)
        if err != nil {
            c.AbortWithStatusJSON(401, gin.H{"error": "Invalid session"})
            return
        }
        
        c.Set("user_id", userID)
        c.Next()
    }
}

func main() {
    r := gin.Default()
    
    r.POST("/login", func(c *gin.Context) {
        userID := 123  // After credential verification
        
        token, _ := createSession(userID)
        
        c.SetCookie(
            "session_token",
            token,
            3600,        // maxAge
            "/",         // path
            "",          // domain
            true,        // secure
            true,        // httpOnly
        )
        
        c.JSON(200, gin.H{"message": "Logged in"})
    })
    
    r.POST("/logout", func(c *gin.Context) {
        token, _ := c.Cookie("session_token")
        if token != "" {
            rdb.Del(ctx, "session:"+token)
        }
        c.JSON(200, gin.H{"message": "Logged out"})
    })
    
    r.GET("/protected", authMiddleware(), func(c *gin.Context) {
        userID := c.GetInt("user_id")
        c.JSON(200, gin.H{"message": "Hello", "user_id": userID})
    })
    
    r.Run(":8000")
}

# ✅ GOOD: Short expiration, HttpOnly cookie
response.set_cookie(
    key="access_token",
    value=jwt_token,
    httponly=True,      # Prevents XSS
    secure=True,        # HTTPS only
    samesite="strict",  # CSRF protection
    max_age=900,        # 15 minutes
)

// ❌ BAD: localStorage (vulnerable to XSS)
localStorage.setItem('token', jwt);

// ✅ GOOD: HttpOnly cookie (set by server)
// Client cannot access via JavaScript

# ✅ GOOD: HttpOnly, Secure, SameSite
response.set_cookie(
    key="session_token",
    value=session_token,
    httponly=True,
    secure=True,
    samesite="lax",  # Balance security and usability
    max_age=3600,
)

# ✅ GOOD: CSRF token for state-changing requests
@app.post("/transfer")
async def transfer(
    csrf_token: str,
    user_id: int = Depends(get_current_user)
):
    if not verify_csrf_token(csrf_token, user_id):
        raise HTTPException(403, "Invalid CSRF token")
    # Process transfer

# Redis Cluster for session storage
version: '3.8'
services:
  redis-cluster:
    image: redis:7-alpine
    command: redis-server --cluster-enabled yes
    ports:
      - "7000-7005:7000-7005"
    deploy:
      replicas: 6

# Redis Cluster client
from redis.asyncio.cluster import RedisCluster

redis = RedisCluster(
    startup_nodes=[
        {"host": "localhost", "port": 7000},
        {"host": "localhost", "port": 7001},
    ]
)

# Nginx load balancing (no sticky sessions needed)
upstream api_backend {
    server backend1:8000;
    server backend2:8000;
    server backend3:8000;
}

server {
    location /api {
        proxy_pass http://api_backend;
    }
}

# Instant revocation
await redis.delete(f"session:{session_token}")

# Revoke all sessions for user
sessions = await redis.keys(f"session:*")
for session_key in sessions:
    user_id = await redis.get(session_key)
    if int(user_id) == target_user_id:
        await redis.delete(session_key)

ACCESS_TOKEN_EXPIRE_MINUTES = 15  # Compromise: security vs UX

# Store revoked JWT IDs in Redis
async def revoke_jwt(token: str):
    payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    jti = payload.get("jti")  # JWT ID
    exp = payload.get("exp")
    
    # Store until expiration
    ttl = exp - int(time.time())
    await redis.setex(f"revoked:{jti}", ttl, "1")

async def is_jwt_revoked(token: str) -> bool:
    payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    jti = payload.get("jti")
    return await redis.exists(f"revoked:{jti}")

def create_token_pair(user_id: int):
    access_token = create_access_token(user_id, expires_minutes=15)
    refresh_token = create_refresh_token(user_id, expires_days=30)
    return access_token, refresh_token

# Refresh token stored in database (opaque)
async def refresh_access_token(refresh_token: str):
    # Validate refresh token in database
    user_id = await get_user_from_refresh_token(refresh_token)
    
    # Generate new token pair
    new_access, new_refresh = create_token_pair(user_id)
    
    # Invalidate old refresh token
    await revoke_refresh_token(refresh_token)
    
    return new_access, new_refresh

# Access token: opaque, stored in Redis
# Includes JWT claims for efficient permission checks

async def create_hybrid_token(user_id: int, roles: list[str]) -> str:
    token_id = secrets.token_urlsafe(32)
    
    # Store in Redis with embedded JWT claims
    token_data = {
        "user_id": user_id,
        "roles": roles,
        "created_at": int(time.time()),
    }
    
    await redis.setex(
        f"token:{token_id}",
        3600,  # 1 hour
        json.dumps(token_data)
    )
    
    return token_id

async def validate_hybrid_token(token_id: str) -> dict:
    data = await redis.get(f"token:{token_id}")
    if not data:
        raise HTTPException(401, "Invalid token")
    
    return json.loads(data)

# Short-lived JWT (15 min) for access
# Long-lived opaque token (30 days) for refresh

@app.post("/login")
async def login(username: str, password: str):
    user_id = 123
    
    access_token = create_jwt(user_id, expires_minutes=15)
    refresh_token = await create_refresh_token(user_id, expires_days=30)
    
    return {
        "access_token": access_token,
        "refresh_token": refresh_token,
        "token_type": "bearer",
    }

@app.post("/refresh")
async def refresh(refresh_token: str):
    user_id = await validate_refresh_token(refresh_token)
    
    new_access = create_jwt(user_id, expires_minutes=15)
    new_refresh = await rotate_refresh_token(refresh_token, user_id)
    
    return {
        "access_token": new_access,
        "refresh_token": new_refresh,
    }

// Axios interceptor for automatic token refresh
axios.interceptors.response.use(
  response => response,
  async error => {
    const originalRequest = error.config;
    
    if (error.response.status === 401 && !originalRequest._retry) {
      originalRequest._retry = true;
      
      try {
        const { data } = await axios.post('/refresh', {
          refresh_token: localStorage.getItem('refresh_token'),
        });
        
        localStorage.setItem('access_token', data.access_token);
        localStorage.setItem('refresh_token', data.refresh_token);
        
        originalRequest.headers['Authorization'] = `Bearer ${data.access_token}`;
        return axios(originalRequest);
      } catch (refreshError) {
        // Refresh failed — redirect to login
        window.location.href = '/login';
        return Promise.reject(refreshError);
      }
    }
    
    return Promise.reject(error);
  }
);

// Refresh token 5 minutes before expiration
function decodeJWT(token) {
  const base64Url = token.split('.')[1];
  const base64 = base64Url.replace(/-/g, '+').replace(/_/g, '/');
  return JSON.parse(window.atob(base64));
}

function scheduleTokenRefresh(accessToken) {
  const payload = decodeJWT(accessToken);
  const expiresAt = payload.exp * 1000;  // Convert to milliseconds
  const refreshAt = expiresAt - (5 * 60 * 1000);  // 5 min before
  
  const timeout = refreshAt - Date.now();
  
  if (timeout > 0) {
    setTimeout(async () => {
      const newTokens = await refreshTokens();
      scheduleTokenRefresh(newTokens.access_token);
    }, timeout);
  }
}