claudetools/api/middleware

ClaudeTools API Middleware

This package provides JWT authentication, authorization, and error handling middleware for the ClaudeTools FastAPI application.

Overview

The middleware package consists of three main modules:

1. auth.py - JWT token management and password hashing
2. error_handler.py - Custom exception classes and global error handlers
3. init.py - Package exports and convenience imports

Authentication (auth.py)

Password Hashing

The middleware uses Argon2 for password hashing (with bcrypt fallback for compatibility):

from api.middleware import hash_password, verify_password

# Hash a password
hashed = hash_password("user_password")

# Verify a password
is_valid = verify_password("user_password", hashed)

JWT Token Management

Create and verify JWT tokens for API authentication:

from api.middleware import create_access_token, verify_token
from datetime import timedelta

# Create a token
token = create_access_token(
    data={
        "sub": "mike@azcomputerguru.com",
        "scopes": ["msp:read", "msp:write"],
        "machine": "windows-workstation"
    },
    expires_delta=timedelta(hours=1)
)

# Verify a token
payload = verify_token(token)
# Returns: {"sub": "mike@...", "scopes": [...], "exp": ..., ...}

Protected Routes

Use dependency injection to protect API routes:

from fastapi import APIRouter, Depends
from api.middleware import get_current_user

router = APIRouter()

@router.get("/protected")
async def protected_route(current_user: dict = Depends(get_current_user)):
    """This route requires authentication."""
    return {
        "message": "Access granted",
        "user": current_user.get("sub"),
        "scopes": current_user.get("scopes")
    }

Optional Authentication

For routes with optional authentication:

from typing import Optional
from fastapi import APIRouter, Depends
from api.middleware import get_optional_current_user

router = APIRouter()

@router.get("/content")
async def get_content(user: Optional[dict] = Depends(get_optional_current_user)):
    """This route works with or without authentication."""
    if user:
        return {"content": "Premium content", "user": user.get("sub")}
    return {"content": "Public content"}

Scope-Based Authorization

Require specific permission scopes:

from fastapi import APIRouter, Depends
from api.middleware import get_current_user, require_scopes

router = APIRouter()

@router.post("/admin/action")
async def admin_action(
    current_user: dict = Depends(get_current_user),
    _: None = Depends(require_scopes("msp:admin"))
):
    """This route requires the 'msp:admin' scope."""
    return {"message": "Admin action performed"}

@router.post("/write")
async def write_data(
    current_user: dict = Depends(get_current_user),
    _: None = Depends(require_scopes("msp:write"))
):
    """This route requires the 'msp:write' scope."""
    return {"message": "Data written"}

Error Handling (error_handler.py)

Custom Exception Classes

The middleware provides several custom exception classes:

• ClaudeToolsException - Base exception class
• AuthenticationError (401) - Authentication failures
• AuthorizationError (403) - Permission denied
• NotFoundError (404) - Resource not found
• ValidationError (422) - Business logic validation errors
• ConflictError (409) - Resource conflicts
• DatabaseError (500) - Database operation failures

Using Custom Exceptions

from api.middleware import NotFoundError, ValidationError, AuthenticationError

# Raise a not found error
raise NotFoundError(
    "User not found",
    resource_type="User",
    resource_id="123"
)

# Raise a validation error
raise ValidationError(
    "Username already exists",
    field="username"
)

# Raise an authentication error
raise AuthenticationError("Invalid credentials")

Exception Response Format

All exceptions return a consistent JSON format:

{
  "error": "Error message",
  "details": {
    "field": "username",
    "resource_type": "User",
    "resource_id": "123"
  },
  "path": "/api/v1/users/123"
}

Registering Exception Handlers

In your FastAPI application initialization:

from fastapi import FastAPI
from api.middleware import register_exception_handlers

app = FastAPI()

# Register all exception handlers
register_exception_handlers(app)

Complete FastAPI Example

Here's a complete example of using the middleware in a FastAPI application:

from fastapi import FastAPI, Depends, HTTPException
from api.middleware import (
    get_current_user,
    require_scopes,
    register_exception_handlers,
    NotFoundError,
    ValidationError
)

# Create FastAPI app
app = FastAPI(title="ClaudeTools API")

# Register exception handlers
register_exception_handlers(app)

# Public endpoint
@app.get("/")
async def root():
    return {"message": "Welcome to ClaudeTools API"}

# Protected endpoint (requires authentication)
@app.get("/api/v1/sessions")
async def list_sessions(current_user: dict = Depends(get_current_user)):
    """List sessions - requires authentication."""
    return {
        "sessions": [],
        "user": current_user.get("sub")
    }

# Admin endpoint (requires authentication + admin scope)
@app.delete("/api/v1/sessions/{session_id}")
async def delete_session(
    session_id: str,
    current_user: dict = Depends(get_current_user),
    _: None = Depends(require_scopes("msp:admin"))
):
    """Delete a session - requires admin scope."""
    # Check if session exists
    if not session_exists(session_id):
        raise NotFoundError(
            "Session not found",
            resource_type="Session",
            resource_id=session_id
        )

    # Delete the session
    delete_session_from_db(session_id)
    return {"message": "Session deleted"}

# Write endpoint (requires authentication + write scope)
@app.post("/api/v1/clients")
async def create_client(
    client_data: dict,
    current_user: dict = Depends(get_current_user),
    _: None = Depends(require_scopes("msp:write"))
):
    """Create a client - requires write scope."""
    # Validate client data
    if client_exists(client_data["name"]):
        raise ValidationError(
            "Client with this name already exists",
            field="name"
        )

    # Create the client
    client = create_client_in_db(client_data)
    return {"client": client}

Configuration

The middleware uses settings from api/config.py:

• JWT_SECRET_KEY - Secret key for signing JWT tokens
• JWT_ALGORITHM - Algorithm for JWT (default: HS256)
• ACCESS_TOKEN_EXPIRE_MINUTES - Token expiration time (default: 60)

Ensure these are set in your .env file:

JWT_SECRET_KEY=your-base64-encoded-secret-key
JWT_ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=60

Token Payload Structure

JWT tokens should contain:

{
  "sub": "mike@azcomputerguru.com",
  "scopes": ["msp:read", "msp:write", "msp:admin"],
  "machine": "windows-workstation",
  "exp": 1234567890,
  "iat": 1234567890,
  "jti": "unique-token-id"
}

Permission Scopes

The system uses three permission scopes:

• msp:read - Read sessions, clients, work items
• msp:write - Create/update sessions, work items
• msp:admin - Manage clients, credentials, delete operations

Notes

• Password hashing uses Argon2 (more secure than bcrypt) due to compatibility issues with Python 3.13
• JWT tokens are stateless and contain all necessary user information
• The system does not use a traditional User model - authentication is based on email addresses
• All exceptions are automatically caught and formatted consistently
• Token verification includes expiration checking