Files
claudetools/api/middleware/README.md
Mike Swanson 390b10b32c Complete Phase 6: MSP Work Tracking with Context Recall System
Implements production-ready MSP platform with cross-machine persistent memory for Claude.

API Implementation:
- 130 REST API endpoints across 21 entities
- JWT authentication on all endpoints
- AES-256-GCM encryption for credentials
- Automatic audit logging
- Complete OpenAPI documentation

Database:
- 43 tables in MariaDB (172.16.3.20:3306)
- 42 SQLAlchemy models with modern 2.0 syntax
- Full Alembic migration system
- 99.1% CRUD test pass rate

Context Recall System (Phase 6):
- Cross-machine persistent memory via database
- Automatic context injection via Claude Code hooks
- Automatic context saving after task completion
- 90-95% token reduction with compression utilities
- Relevance scoring with time decay
- Tag-based semantic search
- One-command setup script

Security Features:
- JWT tokens with Argon2 password hashing
- AES-256-GCM encryption for all sensitive data
- Comprehensive audit trail for credentials
- HMAC tamper detection
- Secure configuration management

Test Results:
- Phase 3: 38/38 CRUD tests passing (100%)
- Phase 4: 34/35 core API tests passing (97.1%)
- Phase 5: 62/62 extended API tests passing (100%)
- Phase 6: 10/10 compression tests passing (100%)
- Overall: 144/145 tests passing (99.3%)

Documentation:
- Comprehensive architecture guides
- Setup automation scripts
- API documentation at /api/docs
- Complete test reports
- Troubleshooting guides

Project Status: 95% Complete (Production-Ready)
Phase 7 (optional work context APIs) remains for future enhancement.

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-01-17 06:00:26 -07:00

304 lines
7.6 KiB
Markdown

# 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