Insufficient Logging in Fastapi

Insufficient logging in Fastapi applications creates dangerous blind spots that attackers can exploit without detection. Fastapi's async nature and middleware architecture create unique logging challenges that developers must address.

The most critical manifestation occurs in Fastapi's dependency injection system. When authentication dependencies fail silently or log at inappropriate levels, attackers can probe endpoints without triggering alerts. Consider this vulnerable pattern:

async def check_auth(x_api_key: str = Header(...)): 
    if x_api_key != os.getenv('VALID_KEY'):
        raise HTTPException(status_code=401)
    # No logging of failed authentication attempts

Without logging failed authentication attempts, brute force attacks go unnoticed. Fastapi's exception handlers can also swallow errors silently:

async def get_user(user_id: int = Path(...)): 
    try:
        return await get_user_from_db(user_id)
    except Exception:
        raise HTTPException(status_code=500)
        # No logging of which user IDs were probed

Rate limiting bypass attempts are another critical gap. Fastapi's async endpoints can process multiple requests simultaneously, making rate limit circumvention harder to detect without proper logging:

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    # Missing: logging of request timing for anomaly detection

Fastapi's Pydantic models can also mask validation failures. When model validation fails, the default behavior doesn't log the invalid input:

class UserCreate(BaseModel):
    username: str
    password: str

@app.post("/users")
async def create_user(user: UserCreate):
    # If validation fails, Fastapi returns 422 but doesn't log the invalid payload
    pass

Detecting insufficient logging in Fastapi requires examining both the application code and runtime behavior. The most effective approach combines static analysis with runtime scanning.

Static analysis should focus on Fastapi's middleware and dependency injection patterns. Look for these anti-patterns:

import ast

def find_silent_exceptions(tree):
    problematic = []
    for node in ast.walk(tree):
        if isinstance(node, ast.ExceptHandler):
            if not node.name or node.name == 'Exception':
                # Check if this except block contains logging
                if not any(isinstance(n, ast.Expr) and 
                          isinstance(n.value, ast.Call) and 
                          isinstance(n.value.func, ast.Name) and 
                          n.value.func.id == 'logger' for n in node.body):
                    problematic.append(node)
    return problematic

Runtime detection with middleBrick reveals logging gaps through black-box scanning. The scanner tests authentication bypass attempts and monitors for missing audit trails:

# middleBrick CLI example
middlebrick scan https://api.example.com/v1/users \
  --test-auth \
  --test-bola \
  --test-rate-limiting

# Output shows:
# [FAIL] Authentication bypass detected - no failed login attempts logged
# [WARN] Rate limit bypass possible - no request rate logging found
# [FAIL] BOLA vulnerability - no access control audit trail

Fastapi's built-in logging configuration can be audited for proper levels and handlers:

import logging
from fastapi import FastAPI

app = FastAPI()

# Check if this exists and is properly configured
if not app.logger.handlers:
    app.logger.addHandler(logging.StreamHandler())
    app.logger.setLevel(logging.INFO)

# Verify critical events are logged at appropriate levels
@app.middleware("http")
async def audit_middleware(request: Request, call_next):
    response = await call_next(request)
    # Check if this logs all requests with user context
    if not hasattr(audit_middleware, 'logged_requests'):
        raise RuntimeError("Audit logging middleware not properly implemented")
    return response

Fastapi provides several native mechanisms for implementing comprehensive logging. The most effective approach combines middleware-based audit logging with structured exception handling.

Implement a comprehensive audit middleware that captures all requests with user context:

from fastapi import Request, Response
import logging
import json
from datetime import datetime

class AuditMiddleware:
    def __init__(self, app):
        self.app = app
        
    async def __call__(self, request: Request, call_next):
        start_time = datetime.utcnow()
        
        # Extract user context from request
        user_id = request.headers.get('X-User-Id', 'anonymous')
        ip_address = request.client.host if request.client else 'unknown'
        
        try:
            response: Response = await call_next(request)
            status_code = response.status_code
        except Exception as e:
            status_code = 500
            raise
        finally:
            # Log the complete request lifecycle
            duration = (datetime.utcnow() - start_time).total_seconds()
            
            log_entry = {
                'timestamp': start_time.isoformat(),
                'method': request.method,
                'path': str(request.url),
                'user_id': user_id,
                'ip_address': ip_address,
                'status_code': status_code,
                'duration_ms': round(duration * 1000),
                'user_agent': request.headers.get('user-agent', ''),
                'content_length': request.headers.get('content-length', '0')
            }
            
            # Log at INFO level for successful requests
            # Log at WARNING for client errors (4xx)
            # Log at ERROR for server errors (5xx)
            if 400 <= status_code < 500:
                logger.warning(f"Client error: {log_entry}")
            elif 500 <= status_code < 600:
                logger.error(f"Server error: {log_entry}")
            else:
                logger.info(f"Request completed: {log_entry}")
                
            # Special handling for authentication failures
            if status_code == 401:
                logger.warning(f"Unauthorized access attempt: {log_entry}")
                
            # Rate limit monitoring
            if duration < 0.01:  # Sub-10ms requests may indicate abuse
                logger.warning(f"Suspiciously fast request: {log_entry}")
                
        return response

app.add_middleware(AuditMiddleware)

Structured exception handling with comprehensive logging:

from fastapi import HTTPException
from starlette.exceptions import HTTPException as StarletteHTTPException
import logging

@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
    # Log detailed information about the exception
    logger.error({
        'type': 'HTTPException',
        'request_id': request.state.request_id,
        'method': request.method,
        'path': str(request.url),
        'user_id': request.headers.get('X-User-Id', 'anonymous'),
        'exception': {
            'status_code': exc.status_code,
            'detail': exc.detail,
            'headers': dict(exc.headers) if exc.headers else {}
        }
    })
    
    # Return the original response
    return JSONResponse(
        content=exc.detail,
        status_code=exc.status_code,
        headers=exc.headers
    )

# Custom exception handler for business logic errors
class BusinessLogicException(Exception):
    def __init__(self, message, user_id=None, metadata=None):
        super().__init__(message)
        self.user_id = user_id
        self.metadata = metadata or {}

@app.exception_handler(BusinessLogicException)
async def business_logic_exception_handler(request: Request, exc: BusinessLogicException):
    # Log business logic failures with user context
    logger.warning({
        'type': 'BusinessLogicException',
        'user_id': exc.user_id or request.headers.get('X-User-Id', 'anonymous'),
        'message': str(exc),
        'metadata': exc.metadata,
        'request_path': str(request.url),
        'request_method': request.method
    })
    
    return JSONResponse(
        content={'detail': 'Business logic error occurred'},
        status_code=400
    )

Fastapi's dependency injection system can be enhanced with logging:

from fastapi import Depends, HTTPException
import logging
from typing import Optional

async def authenticated_user(
    x_api_key: str = Header(...),
    logger: logging.Logger = Depends(get_logger)
) -> dict:
    """
    Dependency that authenticates users and logs all attempts.
    """
    try:
        user = await authenticate_user(x_api_key)
        if user:
            logger.info(f"User authenticated: {user['id']}")
            return user
        else:
            logger.warning(f"Failed authentication attempt for API key: {x_api_key[:8]}...")
            raise HTTPException(status_code=401, detail="Invalid credentials")
    except Exception as e:
        logger.error(f"Authentication error: {str(e)}")
        raise HTTPException(status_code=500, detail="Authentication service unavailable")

async def get_logger() -> logging.Logger:
    """Dependency to provide configured logger."""
    return logging.getLogger(__name__)