wazobia-auth-middleware 1.0.0
Authentication middleware for Python services with JWT and Project authentication
License: MIT
Requires Python: >=3.8

# 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
```bash
pip install -e ./libs/python/jwt-auth
```

### Using requirements.txt
```txt
auth-middleware>=1.0.0
```

### Development Installation
```bash
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:

```bash
# 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

```python
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

```python
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

```python
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
```http
Authorization: Bearer <jwt_token>
```

### Project Authentication
```http
x-app-id: <application_id>
x-app-secret: <application_secret>
```

## Token Structure

The JWT tokens should have the following structure:

```json
{
  "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:

```python
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

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

# Run tests
pytest

# With coverage
pytest --cov=auth_middleware
```

### Code Quality

```bash
# 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.