Memory Leak in Fastapi with Api Keys

A memory leak in a FastAPI application that uses API keys typically arises not from the key validation logic itself, but from how the application handles and stores per-request or per-client state when keys are validated on each request. FastAPI, being a modern async framework, can inadvertently retain references to request-scoped data when responses or background tasks capture variables that should be released. For example, if a developer attaches parsed key metadata (such as the key identifier, scopes, or rate-limit counters) to the request state via request.state or stores per-client caches in a global dictionary keyed by the API key, these objects may persist beyond the request lifecycle if references are unintentionally held. This is especially risky when keys are used to index caches, in-memory stores, or logging contexts that are never cleared, leading to unbounded growth in the runtime process.

When API keys are validated via middleware or dependency injection, the validation routine may create objects (e.g., database sessions, HTTP client sessions, or structured logging contexts) and attach them to the request or response cycle. If these objects hold references to the API key or related data and are not explicitly cleaned up—such as failing to close connections or not using context managers—memory usage increases with each request. In an async environment, if a background task or a callback captures the request or a key-derived object without proper lifecycle management, the garbage collector may be unable to reclaim the memory. Over time, this manifests as a memory leak, observable through steadily increasing resident memory usage, slower response times, or eventual process restarts due to out-of-memory conditions.

The interaction with unauthenticated attack surface testing—where the API is probed without credentials—can amplify the risk. If the endpoint performing key validation does not enforce strict input checks on the key parameter, an attacker might send malformed or excessively large key values that trigger repeated allocations (e.g., creating many temporary objects or log entries). Because the scan runs unauthenticated against the attack surface, such malformed inputs can be exercised repeatedly, accelerating memory consumption. Moreover, if the API key validation path lacks proper bounds or normalization, edge-case inputs may cause data structures to grow unexpectedly. This ties into broader findings such as Input Validation and Unsafe Consumption, where improper handling of key formats leads to resource exhaustion.

In the context of the 12 parallel security checks run by middleBrick, a memory leak may not directly appear as a distinct category but can be inferred from findings like Rate Limiting (excessive allocations due to missing throttling), Data Exposure (leaked objects in logs), or Inventory Management (unreleased resources). The scanner’s runtime analysis can surface symptoms—such as irregular memory patterns during sequential requests—while correlating with OpenAPI/Swagger spec analysis to identify missing cleanup semantics in documented operations. Because middleBrick tests the unauthenticated attack surface, it can detect endpoints where key-related allocations are not bounded, providing evidence of resource mismanagement without needing authenticated access.

To mitigate memory leaks when using API keys in FastAPI, focus on deterministic resource management and avoiding long-lived references tied to key identifiers. Use context managers for allocations that must be cleaned up, and prefer scoped, short-lived objects over global caches indexed by key. Below are concrete, realistic code examples that demonstrate secure patterns.

First, ensure API key validation does not attach persistent data to the request. Instead of storing key-derived objects on request.state, compute what you need inline and avoid retaining references. Use a dependency that validates the key and returns minimal metadata, and rely on FastAPI’s dependency system to clean up after the request.

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import APIKeyHeader
import contextlib

app = FastAPI()
api_key_header = APIKeyHeader(name="X-API-Key")

# A lightweight validator that does not cache globally
def get_api_key(key: str = Depends(api_key_header)):
    # Perform validation (e.g., lookup in a bounded cache or database)
    if not key or len(key) > 256:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Invalid key")
    # Return only necessary metadata, not the raw key if avoidable
    return {"key_id": key[:8]}  # truncated example

@app.get("/items")
async def read_items(claims: dict = Depends(get_api_key)):
    # Use context managers for any resource that needs cleanup
    with contextlib.closing(create_db_session()) as session:
        data = session.query(Item).filter_by(tenant=claims["key_id"]).all()
    return {"data": data}

@contextlib.contextmanager
def create_db_session():
    # Example of a session that is properly closed
    session = db_engine.connect()
    try:
        yield session
    finally:
        session.close()

Second, avoid global dictionaries that grow with each key. If you must maintain per-client state, use bounded data structures with expiration, such as an LRU cache with a size limit and TTL, and ensure cleanup hooks are registered. Do not store references to request or response objects.

from fastapi import FastAPI, Depends
from collections import OrderedDict
import threading
import time

app = FastAPI()

# Bounded cache with TTL to prevent unbounded growth
class BoundedKeyCache:
    def __init__(self, max_size=1000, ttl=300):
        self.max_size = max_size
        self.ttl = ttl
        self._store = OrderedDict()
        self._lock = threading.Lock()

    def get(self, key):
        with self._lock:
            if key in self._store:
                entry = self._store.pop(key)
                if time.time() - entry["ts"] < self.ttl:
                    self._store[key] = entry
                    return entry["value"]
                else:
                    return None
            return None

    def set(self, key, value):
        with self._lock:
            if key in self._store:
                self._store.pop(key)
            elif len(self._store) >= self.max_size:
                self._store.popitem(last=False)
            self._store[key] = {"value": value, "ts": time.time()}

cache = BoundedKeyCache(max_size=1000, ttl=300)

def get_api_key_state(key: str = Depends(api_key_header)):
    state = cache.get(key)
    if state is None:
        # Validate and produce state, then store bounded
        state = {"valid": True}
        cache.set(key, state)
    return state

Third, ensure that background tasks and async callbacks do not capture the request or key-derived objects. If you must run background work, pass only serializable, minimal data and avoid holding references to request-scoped items.

from fastapi import BackgroundTasks

def send_audit_log(data: dict):
    # Process data without capturing request or key objects
    with open("/var/log/audit.log", "a") as f:
        f.write(str(data) + "\n")

@app.post("/action")
async def perform_action(background_tasks: BackgroundTasks, claims: dict = Depends(get_api_key)):
    # Pass only necessary serializable data
    background_tasks.add_task(send_audit_log, {"action": "do_something", "key_id": claims["key_id"]})
    return {"status": "accepted"}

By combining bounded caches, context managers for resource cleanup, and careful handling of background tasks, you reduce the risk of memory leaks while maintaining secure API key usage. These practices align with input validation and resource management checks that middleBrick can surface, helping you address findings before deployment.