# 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): ```python 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: ```python 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: ```python 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: ```python 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: ```python 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 ```python 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: ```json { "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: ```python 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: ```python 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: ```bash JWT_SECRET_KEY=your-base64-encoded-secret-key JWT_ALGORITHM=HS256 ACCESS_TOKEN_EXPIRE_MINUTES=60 ``` ## Token Payload Structure JWT tokens should contain: ```json { "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