Race Condition in Fastapi with Api Keys

A race condition in FastAPI involving API keys typically occurs when authorization checks and state-modifying operations are not performed as a single, atomic unit. Consider an endpoint that reads a resource, validates an API key, and then updates a counter or status field based on that validation. If two concurrent requests pass the key validation step before either has updated the shared state, the application may allow an invalid or duplicate operation, such as double-spending a credit or enabling access beyond intended limits.

In FastAPI, this can manifest when API key validation (e.g., checking a database or cache for key status) is not tightly coupled with the business logic that follows. For example, a request may validate that a key is active and has quota, but between that check and the subsequent write, another request can modify the quota or key status. The framework does not provide built-in locking for business state, so without explicit synchronization at the data store level, the window between read and write becomes the race window.

An attacker can exploit this by issuing many simultaneous requests that rely on a shared resource, such as a rate limit counter or a one-time action enabled by a valid API key. If the validation logic does not enforce atomicity, the attacker may succeed in performing an action that should have been blocked after the first use. This is distinct from authentication failures; the API key itself is valid, but the surrounding logic fails under concurrency. The standard OWASP API Top 10 category for this type of issue is Broken Object Level Authorization (BOLA) when object-level permissions are misapplied, and it can intersect with business logic flaws and unsafe consumption patterns.

middleBrick scans for such logic flaws across its 12 security checks, including BOLA/IDOR and Property Authorization, and maps findings to frameworks like OWASP API Top 10 and SOC2. The scanner tests unauthenticated attack surfaces, so it can identify endpoints where concurrency issues may exist without requiring credentials, focusing on how inputs and state changes interact under load.

Remediation guidance emphasizes ensuring that authorization and state changes occur within a transaction or use row-level locking where supported. Developers should design endpoints so that validation and mutation are a single atomic operation at the data layer, avoiding separate read-then-write patterns for shared state.

To mitigate race conditions when using API keys in FastAPI, enforce atomic validation and state updates at the data store level. Avoid checking key validity and then separately updating a quota or status; instead, perform both within a single database transaction or use an atomic decrement operation.

Below is an example of a vulnerable pattern followed by a corrected version. The vulnerable code checks the API key and quota in separate steps, creating a race window:

from fastapi import FastAPI, Depends, HTTPException, Header
import sqlite3

app = FastAPI()

def get_db():
    return sqlite3.connect("keys.db")

@app.get("/use-credit")
async def use_credit(x_api_key: str = Header(None), db: sqlite3.Connection = Depends(get_db)):
    cursor = db.cursor()
    cursor.execute("SELECT is_active, remaining FROM api_keys WHERE key_value = ?", (x_api_key,))
    row = cursor.fetchone()
    if not row or not row[0] or row[1] <= 0:
        raise HTTPException(status_code=402, detail="Invalid or exhausted key")
    # Race condition: another request can modify remaining between read and write
    db.execute("UPDATE api_keys SET remaining = remaining - 1 WHERE key_value = ?", (x_api_key,))
    db.commit()
    return {"status": "used"}

In this example, two concurrent requests can both read remaining > 0 and then both decrement, allowing usage beyond the quota. The fix is to perform the check and update atomically:

import sqlite3
from fastapi import FastAPI, Depends, HTTPException, Header

app = FastAPI()

def get_db():
    return sqlite3.connect("keys.db")

@app.get("/use-credit-safe")
async def use_credit_safe(x_api_key: str = Header(None), db: sqlite3.Connection = Depends(get_db)):
    cursor = db.cursor()
    # Atomic decrement with condition in a single statement
    cursor.execute(
        "UPDATE api_keys SET remaining = remaining - 1 WHERE key_value = ? AND is_active = 1 AND remaining > 0",
        (x_api_key,)
    )
    db.commit()
    cursor.execute("SELECT changes()")
    if cursor.fetchone()[0] == 0:
        raise HTTPException(status_code=402, detail="Invalid or exhausted key")
    return {"status": "used"}

This approach uses a single UPDATE with a WHERE clause that includes all validity conditions. The database ensures atomicity, so concurrent requests cannot overspend the quota. For high-concurrency scenarios, consider using row-level locking (e.g., SELECT … FOR UPDATE in supported databases) within a transaction to serialize access to the key record.

middleBrick’s CLI tool can be used to scan endpoints for such patterns by running middlebrick scan <url>, while the GitHub Action can add API security checks to your CI/CD pipeline to fail builds if risk scores drop below your threshold. The MCP Server enables scanning APIs directly from your AI coding assistant within the development environment.