Injection Flaws in Fastapi with Mongodb

Injection flaws occur when untrusted data is interpreted as part of a command or query. In a Fastapi application using Mongodb, this typically manifests as NoSQL injection when user-controlled input is concatenated into query structures or aggregation pipelines without proper validation or sanitization. Because Mongodb queries are expressed as JSON-like documents, attackers can inject operators that alter query logic, bypass authentication, or extract data.

Fastapi does not inherently sanitize inputs; developers construct queries using parameters that may come from path, query, or body sources. If these parameters are directly embedded into a dict that becomes a Mongodb query, special operators such as $ne, $in, $or, or $where can be injected. For example, an attacker providing {"username": {"$ne": ""}} as a JSON payload can bypass intended filters.

Consider a login endpoint that builds a filter from request body:

from fastapi import FastAPI, HTTPException
from pymongo import MongoClient

app = FastAPI()
client = MongoClient("mongodb://localhost:27017")
db = client["mydb"]

@app.post("/login")
async def login(user: dict):
    # Unsafe: directly using user input in query
    found = db.users.find_one(user)
    if found:
        return {"status": "ok"}
    raise HTTPException(status_code=401, detail="Invalid credentials")

If the caller sends {"username": "alice", "password": {"$ne": ""}}, the query matches any user named alice with a non-empty password field, effectively bypassing password checks. This is a classic NoSQL injection pattern.

Additionally, aggregation pipelines can be vulnerable if stages are built from unchecked input. An attacker might inject a $skip or $limit stage to manipulate pagination or data exposure. For instance:

pipeline = [
    {"$match": {"status": "active"}},
    *user_supplied_stages  # unsafe concatenation
]
result = db.collection.aggregate(pipeline)

If user_supplied_stages contains malicious stages, the pipeline behavior is altered. The combination of Fastapi's flexibility in accepting JSON and Mongodb's expressive query language increases the attack surface when input validation is omitted.

Beyond injection, improper error handling can leak stack traces or database structure, aiding further exploitation. Automated scans from tools like middleBrick help identify such risky patterns in unauthentiated scans, highlighting endpoints where user input directly influences query construction without sanitization.

Remediation centers on strict input validation, avoiding direct embedding of user input into query documents, and using parameterized patterns. The primary rule is to treat all user data as values, not as part of the query structure.

1. Use a strict schema and map known fields explicitly. Instead of passing a raw dict to find_one, extract and validate expected fields:

from fastapi import FastAPI, HTTPException
from pymongo import MongoClient
from pydantic import BaseModel, constr

app = FastAPI()
client = MongoClient("mongodb://localhost:27017")
db = client["mydb"]

class LoginInput(BaseModel):
    username: constr(min_length=1, max_length=50)
    password: constr(min_length=1, max_length=128)

@app.post("/login")
async def login(input: LoginInput):
    # Safe: only known fields are used
    found = db.users.find_one({"username": input.username, "password": input.password})
    if found:
        return {"status": "ok"}
    raise HTTPException(status_code=401, detail="Invalid credentials")

2. When filtering is needed, construct queries using a denylist or allowlist of operators. Never forward raw user input to $in or $or. If you need array operations, validate each element strictly:

allowed_statuses = {"active", "pending"}
# Unsafe: {"status": {"$in": user_supplied_list}}
# Safe:
if not all(s in allowed_statuses for s in user_supplied_list):
    raise HTTPException(status_code=400, detail="Invalid status")
query = {"status": {"$in": list(allowed_statuses)}}
results = db.collection.find(query)

3. For aggregation pipelines, avoid concatenating external input. Define pipeline templates and inject only safe, validated parameters:

base_pipeline = [
    {"$match": {"active": True}},
    {"$sort": {"created_at": -1}}
]
# If dynamic stages are required, validate each stage structure rigorously
for stage in user_stages:
    if stage.get("$limit") and isinstance(stage["$limit"], int) and stage["$limit"] > 0:
        base_pipeline.append(stage)
    else:
        raise HTTPException(status_code=400, detail="Invalid pipeline stage")
results = db.collection.aggregate(base_pipeline)

4. Use MongoDB driver features such as collation for case-sensitive matching and avoid JavaScript evaluation ($where) entirely. Regularly review indexes and ensure that queries leverage them to reduce unexpected full collection scans that could be abused.

5. Complement these coding practices with runtime security insights. middleBrick’s scans can surface endpoints where query building appears dynamic, enabling developers to prioritize fixes. The CLI (middlebrick scan <url>) and GitHub Action integrations help embed checks into development workflows, while the Web Dashboard tracks improvements over time.