Logging Monitoring Failures in Fastapi

Logging monitoring failures in Fastapi applications often stem from improper exception handling and inadequate logging configuration. When Fastapi applications encounter errors, the default behavior may not provide sufficient visibility into what went wrong, leaving security teams blind to potential attacks.

A common manifestation occurs in exception handlers where developers catch exceptions but fail to log critical details. Consider this vulnerable pattern:

@app.exception_handler(SomeException)
def handle_exception(request: Request, exc: SomeException):
    return JSONResponse(
        status_code=500,
        content={"detail": "Internal Server Error"}
    )

This handler swallows the exception without logging it, making it impossible to detect repeated attack attempts or identify the root cause of failures. Attackers can exploit this by repeatedly triggering exceptions to cause denial of service or to probe for information.

Another Fastapi-specific issue arises with dependency injection. When dependencies fail, Fastapi's dependency injection system may not propagate errors correctly:

@app.get("/sensitive-data")
def get_sensitive_data(db: Database = Depends(get_db)):
    # If get_db fails, the error might not be logged
    return db.query_sensitive_data()

Without proper logging in the dependency injection chain, authentication failures, database connection issues, or other critical errors go unnoticed. This creates blind spots where attackers can repeatedly attempt unauthorized access without triggering alerts.

Fastapi's background tasks also present logging challenges. Tasks run asynchronously and may fail silently:

@app.post("/process")
async def process_data(data: DataModel):
    await app.state.background_tasks.add_task(
        process_in_background, data
    )
    return JSONResponse(status_code=202)

If process_in_background raises an exception, it might not propagate to the main request handler, leaving the failure unlogged. Attackers can exploit this by submitting malformed data that causes background tasks to fail, creating a denial of service condition while remaining undetected.

Detecting logging monitoring failures in Fastapi requires examining both the application code and runtime behavior. Code analysis should focus on exception handlers, dependency injection patterns, and background task implementations.

middleBrick's Fastapi-specific scanning identifies several critical patterns. The scanner examines exception handlers to ensure they log sufficient details before returning responses. It checks for the presence of logging statements in exception handlers:

@app.exception_handler(SomeException)
def handle_exception(request: Request, exc: SomeException):
    logger.error(f"Exception occurred: {exc}", exc_info=True)
    return JSONResponse(
        status_code=500,
        content={"detail": "Internal Server Error"}
    )

The scanner also verifies that dependency injection failures are properly logged. It looks for patterns where database connections, authentication services, or other critical dependencies might fail silently:

@app.get("/api/data")
async def get_data(
    db: Database = Depends(get_db),
    auth: AuthService = Depends(get_auth)
):
    try:
        return await db.fetch_data()
    except Exception as e:
        logger.error(f"Database query failed: {e}")
        raise HTTPException(status_code=500)

For background tasks, middleBrick checks whether tasks are wrapped with proper error handling and logging. The scanner identifies tasks that might run without any error capture:

async def process_in_background(data: DataModel):
    try:
        # Processing logic
        pass
    except Exception as e:
        logger.error(f"Background task failed: {e}", exc_info=True)

middleBrick also tests the runtime behavior by sending malformed requests to trigger exceptions and verifying that the application logs them appropriately. The scanner checks whether the application provides different responses for different types of failures, which could leak information about the system's internal state.

The scanner examines Fastapi's logging configuration to ensure it captures all relevant information. This includes checking that request logging middleware is properly configured to log request details, response status codes, and execution times:

class RequestLoggingMiddleware:
    async def __call__(self, request: Request, call_next):
        start_time = time.time()
        try:
            response = await call_next(request)
            logger.info(f"{request.method} {request.url} {response.status_code} {time.time() - start_time:.2f}s")
            return response
        except Exception as e:
            logger.error(f"{request.method} {request.url} ERROR: {e}")
            raise

Remediating logging monitoring failures in Fastapi requires implementing comprehensive logging strategies that cover all application components. Start with a centralized logging configuration that captures all relevant events:

import logging
import sys
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from pythonjsonlogger import jsonlogger

app = FastAPI()

# Configure structured JSON logging
logger = logging.getLogger()
logger.setLevel(logging.INFO)

json_handler = logging.StreamHandler(sys.stdout)
formatter = jsonlogger.JsonFormatter(
    fmt='%(asctime)s %(name)s %(levelname)s %(message)s'
)
json_handler.setFormatter(formatter)
logger.addHandler(json_handler)

# Global exception handler with comprehensive logging
@app.exception_handler(Exception)
def global_exception_handler(request: Request, exc: Exception):
    logger.error(f"Unhandled exception: {exc}", exc_info=True)
    
    if isinstance(exc, HTTPException):
        return JSONResponse(
            status_code=exc.status_code,
            content={"detail": exc.detail}
        )
    
    return JSONResponse(
        status_code=500,
        content={"detail": "Internal Server Error"}
    )

# Request logging middleware
class RequestLoggingMiddleware:
    def __init__(self, app):
        self.app = app
    
    async def __call__(self, request: Request, call_next):
        start_time = time.time()
        try:
            response = await call_next(request)
            logger.info(f"{request.method} {request.url} {response.status_code} {time.time() - start_time:.2f}s")
            return response
        except Exception as e:
            logger.error(f"{request.method} {request.url} ERROR: {e}")
            raise

app.add_middleware(RequestLoggingMiddleware)

# Dependency injection with logging
async def get_db():
    try:
        db = Database()
        logger.info("Database connection established")
        return db
    except Exception as e:
        logger.error(f"Database connection failed: {e}")
        raise HTTPException(status_code=503, detail="Service unavailable")

# Background task with error handling
async def process_in_background(data: DataModel):
    try:
        # Processing logic
        logger.info(f"Background task started for data: {data.id}")
        # ... processing ...
        logger.info(f"Background task completed for data: {data.id}")
    except Exception as e:
        logger.error(f"Background task failed for data: {data.id}: {e}", exc_info=True)
        raise

@app.post("/process")
async def process_data(data: DataModel):
    await app.state.background_tasks.add_task(
        process_in_background, data
    )
    return JSONResponse(status_code=202)

For authentication and authorization failures, implement specific logging that captures relevant details without exposing sensitive information:

@app.exception_handler(Unauthorized)
def unauthorized_handler(request: Request, exc: Unauthorized):
    logger.warning(f"Unauthorized access attempt from {request.client.host}")
    return JSONResponse(
        status_code=401,
        content={"detail": "Not authenticated"}
    )

@app.exception_handler(PermissionDenied)
def permission_denied_handler(request: Request, exc: PermissionDenied):
    logger.warning(f"Permission denied for user {request.state.user_id} on {request.url}")
    return JSONResponse(
        status_code=403,
        content={"detail": "Insufficient permissions"}
    )

Implement structured logging for API endpoints to capture request and response details:

@app.middleware("http")
async def logging_middleware(request: Request, call_next):
    request_start_time = time.time()
    
    try:
        response = await call_next(request)
        
        logger.info({
            "event": "api_request",
            "method": request.method,
            "path": str(request.url),
            "status_code": response.status_code,
            "duration": time.time() - request_start_time,
            "user_agent": request.headers.get("user-agent"),
            "client_ip": request.client.host
        })
        
        return response
    except Exception as e:
        logger.error({
            "event": "api_error",
            "method": request.method,
            "path": str(request.url),
            "error": str(e),
            "client_ip": request.client.host
        })
        raise