Security Misconfiguration in Fastapi with Mongodb

Security misconfiguration in a Fastapi application using MongoDB often arises from a mismatch between API surface and database access patterns. When endpoints accept user-supplied query parameters or body fields and directly forward them to MongoDB queries without strict validation, transformation, or least-privilege enforcement, the attack surface expands. For example, dynamically building filter dictionaries from request JSON (e.g., using dict(request.query_params) or request.body()) can inadvertently expose fields that should never be filterable, such as is_admin or deleted_at, enabling horizontal or vertical privilege escalation.

Another common misconfiguration is missing or permissive CORS in Fastapi, which can allow unauthorized origins to invoke endpoints and leverage the MongoDB connection string or backend logic inferred from responses. If the MongoDB instance is reachable from the same host or network as the Fastapi app and relies on default authentication settings or unencrypted transport, an attacker who compromises the API path may gain direct database access. Additionally, failing to normalize or sanitize user input before constructing aggregation pipelines can lead to injection, where attacker-controlled keys or operators are interpreted as pipeline stages rather than literal values.

Operational missteps also contribute, such as leaving debug endpoints or detailed error messages enabled in Fastapi that reveal stack traces or MongoDB driver details. These details can expose collection names, field structures, or internal identifiers useful for further exploitation. Without schema-aware validation and strict allowlisting of queryable fields, even well-intentioned features like flexible search or filtering become vectors for unauthorized data enumeration and exfiltration.

Remediation centers on strict schema validation, explicit allowlists, and safe query construction. Use Pydantic models to define input shapes and avoid passing raw user data directly to MongoDB. For read operations, explicitly project only permitted fields and avoid dynamic filter generation from untrusted sources.

from fastapi import Fastapi, HTTPException, Depends
from pydantic import BaseModel, Field
from typing import Optional
from pymongo import MongoClient
from bson import ObjectId
import re

app = Fastapi()
client = MongoClient("mongodb://user:password@mongodb:27017/")
db = client["secure_db"]
users = db["users"]

class UserFilter(BaseModel):
    email: Optional[str] = Field(None, description="Exact email match")
    name_pattern: Optional[str] = Field(None, max_length=100, description="Substring match on name")
    # Explicitly exclude dangerous fields
    class Config:
        orm_mode = True

# Safe endpoint with allowlisted filter fields
def get_safe_filter(data: dict) -> dict:
    filter_keys = {"email", "name_pattern"}
    safe = {k: v for k, v in data.items() if k in filter_keys and v is not None}
    if "email" in safe:
        if not re.match(r"^[^@\s]+@[^@\s]+\.[^@\s]+$", safe["email"]):
            raise HTTPException(status_code=400, detail="Invalid email")
    if "name_pattern" in safe:
        safe["name"] = {"$regex": re.escape(safe["name_pattern"]), "$options": "i"}
        del safe["name_pattern"]
    return safe

@app.get("/users")
def list_users(filter: UserFilter):
    query = get_safe_filter(filter.dict())
    cursor = users.find(query, {"name": 1, "email": 1, "status": 1})  # Explicit projection
    results = list(cursor)
    return {"count": len(results), "users": results}

@app.post("/users")
def create_user(data: dict):
    # Validate and normalize before insert
    if "email" not in data or not re.match(r"^[^@\s]+@[^@\s]+\.[^@\s]+$", data["email"]):
        raise HTTPException(status_code=400, detail="Invalid email")
    safe_data = {
        "email": data["email"],
        "name": data.get("name", ""),
        "status": data.get("status", "active"),
        "created_at": datetime.utcnow()
    }
    result = users.insert_one(safe_data)
    return {"id": str(result.inserted_id)}

# Parameterized aggregation to avoid injection
def build_pipeline(rules: list):
    pipeline = []
    for rule in rules:
        if rule["field"] in ("status", "role") and rule["op"] == "eq":
            pipeline.append({"$match": {rule["field"]: rule["value"]}})
        else:
            raise HTTPException(status_code=400, detail="Unsupported rule")
    return pipeline

@app.post("/users/aggregate")
def run_aggregate(rules: list):
    pipeline = build_pipeline(rules)
    cursor = users.aggregate(pipeline)
    return list(cursor)

Key practices demonstrated:

• Strict input schemas with Pydantic to prevent unexpected fields.
• Allowlisted filter fields and regex validation to avoid injection and data leakage.
• Explicit projection in find to limit returned sensitive fields.
• Parameterized aggregation construction that only permits safe operators and fields.
• Normalized inserts with required fields and server-side timestamps.

Additionally, ensure MongoDB is configured with role-based access control, TLS encryption in transit, and network isolation. In Fastapi, enable CORS with restrictive origins and avoid exposing stack traces in production to reduce information leakage.