Nosql Injection in Fastapi with Api Keys

Nosql Injection occurs when user-controlled input is improperly used to construct queries against NoSQL databases such as MongoDB. In Fastapi, endpoints that accept query or body parameters and directly forward them to a NoSQL driver can become vulnerable if input is not validated or sanitized. Using Api Keys for authentication does not inherently prevent this class of injection; it governs access to the endpoint but does not alter how the endpoint builds database queries.

When an API key is required, developers may assume the endpoint is protected and therefore skip rigorous input validation. This false sense of security can lead to patterns where the API key is checked early, and the remaining parameters are passed directly into a dictionary that is used in a MongoDB $where or aggregation pipeline. For example, a route that filters user data by a JSON field may concatenate request values into the query object without type checks or allowlist constraints, enabling an attacker to inject operators like $ne, $gt, or $exists to bypass intended filters.

Consider a Fastapi route that retrieves user records based on filters provided in JSON. If the route requires an Api Key in a header but does not sanitize the JSON payload, an attacker can send crafted operators to manipulate the query logic. A request with a valid Api Key might include {"username": {"$ne": null}} to retrieve unexpected records, or {"role": {"$in": ["admin", "superuser"]}} to escalate privileges. Because the API key is valid, the request proceeds to the database layer where the unsanitized input alters the semantics of the query, leading to unauthorized data access or enumeration. This pattern is especially risky when combined with overly permissive CORS or when the endpoint exposes internal fields such as hashed passwords or session tokens.

An additional risk vector arises when the NoSQL query is constructed dynamically using string formatting or concatenation. For instance, building a query by interpolating header values or path parameters into JSON structures can allow injection of operators or malicious payloads. Even with Api Key enforcement, if the server trusts the client-supplied keys for constructing queries, the attack surface remains wide. The API key identifies the caller but does not sanitize the structure of the database command, meaning injection techniques such as sending nested operators or malformed JSON can bypass logical constraints intended by the developer.

Real-world parallels can be found in reported findings where endpoints exposed NoSQL injection despite requiring authentication tokens. Detection typically involves sending operators that change the logical path of the query and observing differences in response size, timing, or error messages. Because the vulnerability resides in how the application constructs the query, authentication mechanisms like Api Keys alone are insufficient. Defense requires strict input validation, type checking, and use of parameterized patterns that avoid direct inclusion of user data in query construction.

Remediation centers on treating all user-supplied data as untrusted, regardless of the presence of an Api Key. Implement strict validation using Pydantic models and avoid building queries from raw dictionaries. Below are concrete patterns to secure Fastapi endpoints that use Api Keys.

First, define a Pydantic model that explicitly lists allowed fields and types, and use it to parse incoming data. This prevents unexpected operators from being interpreted as query instructions.

from fastapi import FastAPI, Header, HTTPException, Depends
from pydantic import BaseModel, Field
from typing import Optional

app = FastAPI()

# Example API key validation dependency (simplified)
def verify_api_key(x_api_key: str = Header(...)):
    valid_keys = {"trusted-key-123", "trusted-key-456"}
    if x_api_key not in valid_keys:
        raise HTTPException(status_code=401, detail="Invalid API Key")
    return x_api_key

class FilterParams(BaseModel):
    username: Optional[str] = Field(None, max_length=64)
    role: Optional[str] = Field(None, max_length=32)
    # Explicitly disallow operators by not including them in the model

@app.get("/users")
async def get_users(
    filter_params: FilterParams,
    api_key: str = Depends(verify_api_key)
):
    # Safe: filter_params is a validated model; no direct injection into query
    query = {"username": filter_params.username, "role": filter_params.role}
    # Further sanitize: remove None values if your database expects exact matches
    query = {k: v for k, v in query.items() if v is not None}
    # Use parameterized methods or ORM constructs instead of raw dict expansion
    # Example with a hypothetical database client:
    # results = db.collection.find(query)
    return {"filters": query}

Second, when constructing queries for MongoDB or similar stores, prefer the driver’s built-in mechanisms and avoid dynamic operator injection. Never directly update a query dictionary with user input. Instead, map validated fields explicitly.

import pymongo
from fastapi import Depends

client = pymongo.MongoClient("mongodb://localhost:27017")
db = client["mydb"]

def get_db_collection():
    return db["users"]

def safe_find(collection, filters):
    # Build query using explicit field names; operators are not derived from user input
    query = {}
    if filters.get("username"):
        query["username"] = filters["username"]
    if filters.get("role"):
        query["role"] = filters["role"]
    return collection.find(query)

@app.get("/users/safe")
async def get_users_safe(
    filter_params: FilterParams,
    api_key: str = Depends(verify_api_key)
):
    collection = get_db_collection()
    results = safe_find(collection, filter_params.dict(exclude_none=True))
    return [doc for doc in results]

Third, apply allowlists and reject known operator patterns if you must accept broader JSON input. Use middleware or a validation layer to detect and block common injection indicators such as $where, $eval, or $function.

import re
from fastapi import Request

operator_patterns = re.compile(r'\$\w+')

@app.middleware("http")
async def block_nosql_injection(request: Request, call_next):
    if request.method in {"POST", "PUT", "PATCH"}:
        body = await request.json()
        if isinstance(body, dict):
            for key, value in body.items():
                if isinstance(value, dict) and operator_patterns.search(str(value)):
                    raise HTTPException(status_code=400, detail="Potential NoSQL injection detected")
    response = await call_next(request)
    return response

By combining Api Key authentication with strict input models, explicit query building, and operator pattern detection, you reduce the risk of Nosql Injection. Remember that the API key identifies the caller; it does not sanitize data. Always validate and constrain inputs before they reach the database layer.