JWT Authentication - PyJWT, Token Creation, Verification

Introduction

JSON Web Tokens (JWT) provide a compact, URL-safe means of representing claims to be transferred between two parties. This tutorial covers token creation, verification, and implementation in Python.

JWT Token Creation

import jwt
from datetime import datetime, timedelta

SECRET_KEY = "your-secret-key-here"
ALGORITHM = "HS256"

def create_access_token(data: dict, expires_delta: timedelta = None):
    to_encode = data.copy()
    
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(hours=24)
    
    to_encode.update({"exp": expire})
    
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

def create_token_pair(user_id: int, username: str):
    access_token = create_access_token(
        data={"sub": str(user_id), "username": username},
        expires_delta=timedelta(hours=1)
    )
    
    refresh_token = create_access_token(
        data={"sub": str(user_id), "type": "refresh"},
        expires_delta=timedelta(days=7)
    )
    
    return {
        "access_token": access_token,
        "refresh_token": refresh_token,
        "token_type": "bearer"
    }

# Create token for user
token = create_token_pair(1, "john")
print(token['access_token'])

JWT Token Verification

import jwt
from jwt.exceptions import ExpiredSignatureError, InvalidTokenError
from datetime import datetime

def decode_token(token: str) -> dict:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except ExpiredSignatureError:
        raise Exception("Token has expired")
    except InvalidTokenError:
        raise Exception("Invalid token")

def verify_token(token: str) -> bool:
    try:
        jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return True
    except:
        return False

def get_current_user(token: str) -> User:
    payload = decode_token(token)
    user_id = payload.get("sub")
    if not user_id:
        raise Exception("Invalid token payload")
    
    user = User.query.get(int(user_id))
    if not user:
        raise Exception("User not found")
    
    return user

# Protected endpoint example
def get_user_profile(token: str):
    user = get_current_user(token)
    return {
        "id": user.id,
        "username": user.username,
        "email": user.email
    }

JWT with Flask/FastAPI

# fastapi implementation
from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

app = FastAPI()
security = HTTPBearer()

def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
    token = credentials.credentials
    try:
        payload = decode_token(token)
        user_id = payload.get("sub")
        return User.get(int(user_id))
    except:
        raise HTTPException(status_code=401, detail="Invalid token")

@app.get("/profile")
async def get_profile(current_user: User = Depends(get_current_user)):
    return {"username": current_user.username, "email": current_user.email}

Practice Problems

1. Implement token refresh endpoint
2. Add token blacklisting for logout functionality
3. Create tokens with custom claims and roles
4. Implement token rotation for enhanced security
5. Add token expiration and renewal prompts