Wazobia JWT Authentication Middleware

A comprehensive authentication middleware package for Python web applications supporting JWT and Project-based authentication with JWKS (JSON Web Key Set) validation, Redis caching, and multi-framework support.

🚀 Features

• 🔐 Dual Authentication Methods: JWT tokens for user authentication and project tokens for service-to-service communication
• 🌐 Multi-Framework Support: Flask, FastAPI, and GraphQL (Strawberry/Graphene)
• 🔑 JWKS Integration: Dynamic public key fetching with RS512 algorithm support
• ⚡ Redis Caching: High-performance caching with configurable TTL
• 🚫 Token Revocation: Built-in support for checking revoked tokens
• 🏢 Multi-tenant: Support for project-specific JWT signing keys
• 📝 Type Safety: Full type hints for better IDE support and development experience
• 🔄 Auto Key Rotation: Automatic JWKS cache refresh and key rotation support
• 🛡️ Security: HMAC signature validation and comprehensive token validation

📦 Installation

Using pip

pip install -e ./libs/python/jwt-auth

Using requirements.txt

auth-middleware>=1.0.0

Development Installation

cd libs/python/jwt-auth
pip install -e .

⚙️ Environment Variables

Required Configuration

Variable	Description	Example	Default
REDIS_URL	Redis connection URL for caching	redis://localhost:6379/0	-
MERCURY_BASE_URL	Mercury service base URL (JWT issuer)	http://localhost:4000	http://localhost:4000
SIGNATURE_SHARED_SECRET	HMAC signature validation secret	your-secure-secret-key	-
SERVICE_ID	Current service identifier for project auth	dolos-service-uuid	-
NEXUS_ID	Default project UUID fallback	550e8400-e29b-41d4-a716-446655440000	-

Optional Configuration

Variable	Description	Example	Default
CACHE_EXPIRY_TIME	JWT token cache TTL (seconds)	3600	3600
JWKS_CACHE_TTL	JWKS public key cache TTL (seconds)	18000	18000

Environment File Example

Create a .env file in your project root:

# Redis Configuration
REDIS_URL=redis://localhost:6379/0

# Mercury Service (JWT Issuer)
MERCURY_BASE_URL=http://localhost:4000

# Security
SIGNATURE_SHARED_SECRET=your-very-secure-shared-secret-key

# Service Configuration
SERVICE_ID=550e8400-e29b-41d4-a716-446655440000  #Dolo / udjat/ coeus service ID
NEXUS_ID=550e8400-e29b-41d4-a716-446655440000 #Frontend Admin Dashboard ID

# Cache Configuration (Optional)
CACHE_EXPIRY_TIME=3600
JWKS_CACHE_TTL=18000

Usage Examples

Flask Integration

from flask import Flask, jsonify, g
from auth_middleware import flask_jwt_auth, flask_project_auth, flask_combined_auth

app = Flask(__name__)

# JWT Authentication
@app.route('/api/user/profile')
@flask_jwt_auth()
def get_user_profile():
    user = g.user
    return jsonify({
        'uuid': user['uuid'],
        'email': user['email'],
        'name': user['name']
    })

# Project Authentication
@app.route('/api/project/data')
@flask_project_auth()
def get_project_data():
    project = g.project
    return jsonify({
        'projectUuid': project['projectUuid'],
        'projectName': project['projectName']
    })

# Combined Authentication (both required)
@app.route('/api/secure/resource')
@flask_combined_auth()
def get_secure_resource():
    user = g.user
    project = g.project
    return jsonify({
        'user': user,
        'project': project,
        'data': 'sensitive information'
    })

if __name__ == '__main__':
    app.run(debug=True)

FastAPI Integration

from fastapi import FastAPI, Depends
from auth_middleware import (
    fastapi_jwt_auth,
    fastapi_project_auth,
    fastapi_combined_auth
)

app = FastAPI()

# JWT Authentication
@app.get('/api/user/profile')
async def get_user_profile(user = Depends(fastapi_jwt_auth)):
    return {
        'uuid': user['uuid'],
        'email': user['email'],
        'name': user['name']
    }

# Project Authentication
@app.get('/api/project/data')
async def get_project_data(project = Depends(fastapi_project_auth)):
    return {
        'projectUuid': project['projectUuid'],
        'projectName': project['projectName']
    }

# Combined Authentication
@app.get('/api/secure/resource')
async def get_secure_resource(auth_data = Depends(fastapi_combined_auth)):
    return {
        'user': auth_data['user'],
        'project': auth_data['project'],
        'data': 'sensitive information'
    }

GraphQL Integration (with Strawberry)

import strawberry
from datetime import datetime
from strawberry.types import Info
from auth_middleware import (
    JWTAuthPermission,
    ProjectAuthPermission,
    CombinedAuthPermission,
    requires_jwt_auth,
    requires_project_auth,
    requires_combined_auth,
    get_current_user,
    get_current_project,
)
from typing import List

# Example data models
@strawberry.type
class Message:
    text: str
    timestamp: datetime
    from_user: str

@strawberry.type
class User:
    uuid: str
    email: str
    name: str

@strawberry.type
class Query:
    # Project-protected field using permission_classes
    @strawberry.field(permission_classes=[ProjectAuthPermission])
    def hello(self, info: Info, name: str = "World") -> str:
        project_data = get_current_project(info)
        return f"Hello {name}! This is a public endpoint. {project_data}"
    
    # JWT-protected field using permission_classes
    @strawberry.field(permission_classes=[JWTAuthPermission])
    def whoami(self, info: Info) -> User:
        user_data = get_current_user(info)
        return User(
            uuid=user_data["uuid"],
            email=user_data["email"],
            name=user_data["name"]
        )
    
    # Project-protected field using decorator
    @strawberry.field
    @requires_project_auth
    async def hello_1(self, info: Info, name: str = "World_1") -> str:
        project_data = get_current_project(info)
        return f"Hello {name}! This is a public endpoint. {project_data}"
    
    # JWT-protected field using decorator
    @strawberry.field
    @requires_jwt_auth
    async def my_messages(self, info: Info) -> List[Message]:
        user = get_current_user(info)
        messages = [
            Message(
                text="Welcome to the authenticated area!",
                timestamp=datetime.now(),
                from_user="System"
            ),
            Message(
                text=f"Hello {user['name']}, you are authenticated!",
                timestamp=datetime.now(),
                from_user="System"
            ),
            Message(
                text="Your JWT token is valid",
                timestamp=datetime.now(),
                from_user="Auth Service"
            )
        ]
        return messages

    # Combined auth using decorator
    @strawberry.field
    @requires_combined_auth
    async def hello_2(self, info: Info, name: str = "World_2") -> str:
        user = info.context["user"]
        project = info.context["project"]
        return f"Hello {user['name']}! This is a public endpoint. {project}"
    
    # Combined auth using permission_classes
    @strawberry.field(permission_classes=[CombinedAuthPermission])
    async def hello_3(self, info: Info, name: str = "World_3") -> str:
        user = info.context["user"]
        project = info.context["project"]
        return f"Hello {user['name']}! This is a public endpoint. {project}"

schema = strawberry.Schema(query=Query)```


### GraphQL Integration (with Graphene)

```python
import graphene
from auth_middleware import (
    GraphQLAuthHelper,
    jwt_auth_required,
    project_auth_required,
    combined_auth_required,
    GqlContext,
    AuthenticatedRequest
)

# Initialize helper
graphql_auth = GraphQLAuthHelper()

class UserType(graphene.ObjectType):
    uuid = graphene.String()
    email = graphene.String()
    name = graphene.String()

class Query(graphene.ObjectType):
    # JWT protected query
    @jwt_auth_required
    async def resolve_current_user(self, info):
        user = info.context.req.user
        return UserType(
            uuid=user['uuid'],
            email=user['email'],
            name=user['name']
        )
    
    # Project protected query
    @project_auth_required
    async def resolve_project_info(self, info):
        project = info.context.req.project
        return {
            'projectUuid': project['projectUuid'],
            'projectName': project['projectName']
        }
    
    # Both authentications required
    @combined_auth_required
    async def resolve_secure_data(self, info):
        user = info.context.req.user
        project = info.context.req.project
        return {
            'user': user,
            'project': project,
            'sensitive_data': 'classified information'
        }

# Setup GraphQL with Flask
from flask import Flask
from flask_graphql import GraphQLView

app = Flask(__name__)

def get_context():
    # Create context with authenticated request
    return GqlContext(AuthenticatedRequest(request))

app.add_url_rule(
    '/graphql',
    view_func=GraphQLView.as_view(
        'graphql',
        schema=graphene.Schema(query=Query),
        graphiql=True,
        get_context=get_context
    )
)

Direct Middleware Usage

import asyncio
from auth_middleware import JwtAuthMiddleware, ProjectAuthMiddleware, AuthenticatedRequest

async def authenticate_request(request):
    # JWT Authentication
    jwt_auth = JwtAuthMiddleware()
    auth_req = AuthenticatedRequest(request)
    
    try:
        await jwt_auth.authenticate(auth_req)
        print(f"Authenticated user: {auth_req.user}")
    except Exception as e:
        print(f"JWT auth failed: {e}")
    
    # Project Authentication
    project_auth = ProjectAuthMiddleware()
    
    try:
        await project_auth.authenticate(auth_req)
        print(f"Authenticated project: {auth_req.project}")
    except Exception as e:
        print(f"Project auth failed: {e}")

# Run authentication
# asyncio.run(authenticate_request(your_request_object))

API Headers

JWT Authentication

Authorization: Bearer <jwt_token>

Project Authentication

x-app-id: <application_id>
x-app-secret: <application_secret>

Token Structure

The JWT tokens should have the following structure:

{
  "sub": {
    "uuid": "user-uuid",
    "email": "user@example.com",
    "name": "User Name"
  },
  "project_uuid": "project-uuid",
  "type": "access",
  "iss": "http://mercury.example.com",
  "aud": "http://athens.example.com",
  "exp": 1234567890,
  "nbf": 1234567890,
  "iat": 1234567890,
  "jti": "token-id"
}

Error Handling

The middleware raises exceptions with descriptive messages:

from flask import Flask, jsonify
from auth_middleware import flask_jwt_auth

app = Flask(__name__)

@app.route('/api/protected')
@flask_jwt_auth()
def protected_route():
    # This is only reached if authentication succeeds
    return jsonify({'message': 'Success'})

@app.errorhandler(401)
def handle_auth_error(e):
    return jsonify({'error': str(e)}), 401

Caching Strategy

• Project credentials: Cached for 15 minutes (configurable via CACHE_EXPIRY_TIME)
• JWT tokens: Cached for 1 hour after validation
• JWKS keys: Cached for 10 minutes
• Revoked tokens: Checked on every request

Development

Running Tests

# Install dev dependencies
pip install -e ".[dev]"

# Run tests
pytest

# With coverage
pytest --cov=auth_middleware

Code Quality

# Format code
black auth_middleware/

# Lint
flake8 auth_middleware/

# Type checking
mypy auth_middleware/

Architecture

The package mirrors the TypeScript implementation with Python-specific adaptations:

• Async/await support for all authentication methods
• Framework-specific adapters for Flask (sync) and FastAPI (async)
• Type hints throughout for better IDE support
• Singleton Redis connection manager for efficiency
• Decorator pattern for clean integration

License

MIT

Contributing

1. Fork the repository
2. Create a feature branch
3. Add tests for new functionality
4. Ensure all tests pass
5. Submit a pull request

Support

For issues and questions, please use the GitHub issue tracker.