Nosql Injection in Fastapi with Jwt Tokens

Nosql Injection occurs when user-controlled input is interpreted as part of a NoSQL query without proper validation or parameterization. In Fastapi applications that rely on JWT Tokens for authentication, this combination can expose endpoints to injection when the token payload or decoded claims are used directly in constructing database queries. JWT Tokens often carry user identifiers, roles, or scopes, and if these values are embedded into NoSQL queries without sanitization, an attacker may manipulate them to alter query logic.

For example, consider a Fastapi route that decodes a JWT Token to extract a tenant identifier and uses that identifier in a MongoDB query. If the tenant ID is taken directly from the JWT payload and concatenated into the query, an attacker who can influence the token (via weak signing, algorithm confusion, or token leakage) can inject crafted input such as {'$ne': ''} or {'$where': 'return true'}. This can lead to authentication bypass, unauthorized data access, or data exfiltration.

The risk is amplified when JWT Tokens are accepted without strict validation of claims, issuer, audience, or expiration. An attacker might present a modified token with injected fields that the application trusts. Because Fastapi routes often deserialize the token into Pydantic models, developers may assume safety, but if decoded values are passed to NoSQL drivers as part of query filters, the application effectively builds queries from attacker-controlled data.

In a NoSQL context, injection patterns differ across databases. In MongoDB, operators like $gt, $in, and $regex can be abused; in DynamoDB, condition expressions may be manipulated. If a JWT claim such as user_role is used directly in a DynamoDB ConditionExpression without validation, an attacker could set the role to admin AND #attr = :val to escalate privileges. The interplay between JWT Tokens and NoSQL databases requires strict schema enforcement, input sanitization, and separation of trusted data from query construction.

middleBrick scans such attack surfaces by testing unauthenticated endpoints and analyzing OpenAPI specifications alongside runtime behavior. It checks whether JWT Token claims are reflected in database interactions and whether NoSQL queries are parameterized. The scanner flags instances where dynamic query building includes unchecked token-derived values, providing prioritized findings with severity and remediation guidance.

Remediation focuses on strict validation of JWT Tokens, avoiding direct use of token claims in NoSQL queries, and using parameterized patterns. Below are concrete code examples for Fastapi that illustrate secure handling.

First, always verify the token signature, issuer, audience, and expiration using a well-audited library such as python-jose or PyJWT. Do not trust payload values without validation.

from fastapi import Depends, HTTPException, status
from jose import JWTError, jwt
from pydantic import BaseModel

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"

class TokenData(BaseModel):
    username: str | None = None
    tenant_id: str | None = None
    scopes: list[str] = []

def decode_token(token: str):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        tenant_id: str = payload.get("tenant_id")
        scopes: list[str] = payload.get("scopes", [])
        if username is None or tenant_id is None:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Invalid authentication credentials",
                headers={"WWW-Authenticate": "Bearer"},
            )
        return TokenData(username=username, tenant_id=tenant_id, scopes=scopes)
    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )

Second, never directly embed decoded claims into NoSQL query structures. Instead, map claims to allowed values or use them as filters on already-parameterized queries. For MongoDB with PyMongo, use static field names and pass user data only as parameter values.

from pymongo import MongoClient
from fastapi import Depends

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

async def get_user_data(token_data: TokenData = Depends(decode_token)):
    # Safe: tenant_id is used as a value, not as part of the query operator
    collection = db["tenant_data"]
    cursor = collection.find({ "tenant_id": token_data.tenant_id })
    return list(cursor)

For DynamoDB, use ExpressionAttributeValues and avoid concatenating token claims into ConditionExpression strings. Define explicit placeholder mappings.

import boto3
from botocore.exceptions import ClientError

async def get_item_safe(token_data: TokenData):
    client = boto3.client("dynamodb")
    try:
        response = client.get_item(
            TableName="TenantTable",
            Key={
                "tenant_id": {"S": token_data.tenant_id},
                "entity_id": {"S": "known-id"}
            },
            ConditionExpression="attribute_exists(tenant_id)",
            ExpressionAttributeNames={},
            ExpressionAttributeValues={":val": {"S": token_data.tenant_id}}
        )
        return response.get("Item")
    except ClientError as e:
        raise HTTPException(status_code=400, detail=str(e))

Third, apply least privilege to the database credentials used by Fastapi and scope token claims to minimal required permissions. Regularly rotate secrets and audit token usage. middleBrick’s Pro plan supports continuous monitoring for such patterns, integrating with CI/CD pipelines to catch risky code before deployment.