Token Generation during User Login

To implement security, the first step involves generating a JSON Web Token (JWT) upon successful user authentication. The following example demonstrates a login endpoint using the Sanic framework. This view accepts user credentials, validates them against a database, and returns a signed token if the credentials are correct.

import jwt
import datetime
from sanic import Blueprint, response
from sanic.views import HTTPMethodView
from Database.models import UserSchema

auth_bp = Blueprint('auth', url_prefix='/api/v1')

class LoginView(HTTPMethodView):
    def __init__(self):
        self.required_fields = ['username', 'password']

    def generate_token(self, user_id, role):
        """Generates a JWT token with an expiration time."""
        expiration = datetime.datetime.utcnow() + datetime.timedelta(hours=24)
        payload = {
            'sub': user_id,
            'role': role,
            'exp': expiration
        }
        return jwt.encode(payload, 'your_super_secret_key', algorithm='HS256')

    async def post(self, request):
        username = request.json.get('username', '').strip()
        password = request.json.get('password', '').strip()

        if not username or not password:
            return response.json({'error': 'Missing credentials'}, status=400)

        # Fetch user from database
        user = await UserSchema.find_one({'username': username})

        if user and UserSchema.verify_password(password, user['password_hash']):
            token = self.generate_token(user['user_id'], user['role'])
            return response.json({
                'status': 'success',
                'token': token,
                'user_id': str(user['_id'])
            })

        return response.json({'error': 'Invalid credentials'}, status=401)

auth_bp.add_route(LoginView.as_view(), '/login')

JWT Configuraton Settings

Centralizing configuraton makes the application easier to maintain. The configuraton module below defines the secret key, the signing algorithm, the token lifespan, and validation options. This setup ensures that the signature and expiration date are verified during decoding.

# config/jwt_settings.py

import jwt

class JWTConfig:
    SECRET_KEY = "my_secure_app_secret"
    ALGORITHM = 'HS256'
    EXPIRATION_SECONDS = 86400  # 24 hours
    LEEWAY = 60  # Seconds to account for clock skew

    OPTIONS = {
        'verify_signature': True,
        'verify_exp': True,
        'verify_nbf': False,
        'verify_iat': True,
        'verify_aud': False
    }

Authentication Middleware Decorator

To protect routes, we create a decorator that intercepts the request, validates the JWT from the Authorization header, and either allows the request to proceed or returns an error. This implementation also includes logic to handle specific JWT exceptions such as expiration or invalid signatures.

# middleware/auth.py
import jwt
from functools import wraps
from sanic import response
from jwt.exceptions import (
    InvalidTokenError, DecodeError, ExpiredSignatureError,
    InvalidAudienceError, InvalidIssuerError
)
from config.jwt_settings import JWTConfig
import time
import datetime

def protected_route(wrapped):
    @wraps(wrapped)
    async def decorator(request, *args, **kwargs):
        auth_header = request.headers.get('Authorization')
        
        if not auth_header:
            return response.json({'error': 'Authorization header missing'}, status=401)

        # Extract token (assuming "Bearer <token>" format or just the token)
        token_parts = auth_header.split()
        if len(token_parts) != 2 or token_parts[0].lower() != 'bearer':
            return response.json({'error': 'Invalid authorization header format'}, status=401)
            
        token = token_parts[1]

        try:
            # Decode the token
            payload = jwt.decode(
                token,
                JWTConfig.SECRET_KEY,
                leeway=JWTConfig.LEEWAY,
                options=JWTConfig.OPTIONS
            )
            
            # Attach user info to request context for use in the view
            request.ctx.user = payload
            
            # Execute the original view function
            resp = await wrapped(request, *args, **kwargs)
            
            # Optional: Logic to check if token needs refreshing
            current_timestamp = int(time.time())
            # If the token is about to expire (e.g., within 30 minutes), issue a new one
            if payload['exp'] - current_timestamp < 1800:
                new_exp = datetime.datetime.utcnow() + datetime.timedelta(seconds=JWTConfig.EXPIRATION_SECONDS)
                new_payload = {k: v for k, v in payload.items() if k != 'exp'}
                new_payload['exp'] = new_exp
                
                new_token = jwt.encode(new_payload, JWTConfig.SECRET_KEY, algorithm=JWTConfig.ALGORITHM)
                
                # Modify response to include new token
                resp_json = resp.json
                resp_json['refreshed_token'] = new_token
                return response.json(resp_json)

            return resp

        except ExpiredSignatureError:
            return response.json({'error': 'Token has expired'}, status=401)
        except (InvalidTokenError, DecodeError, InvalidAudienceError, InvalidIssuerError) as e:
            return response.json({'error': f'Invalid token: {str(e)}'}, status=401)
        except Exception as e:
            return response.json({'error': f'Server error: {str(e)}'}, status=500)

    return decorator
</token>

Securing View Functions

Applying the security layer is straightforward. Simply decorate the view methods that require authentication with the @protected_route decorator. The example below shows a protected resource that returns data only if a valid token is provided in the request header.

from sanic import Blueprint, response
from sanic.views import HTTPMethodView
from middleware.auth import protected_route

resource_bp = Blueprint('resources', url_prefix='/api/v1')

class DashboardView(HTTPMethodView):
    @protected_route
    async def get(self, request):
        # Access user info attached by the decorator
        user_info = request.ctx.user
        return response.json({
            'message': 'Access granted',
            'user_role': user_info.get('role'),
            'data': [1, 2, 3]
        })

resource_bp.add_route(DashboardView.as_view(), '/dashboard')