Nosql Injection in Flask with Bearer Tokens

Nosql Injection occurs when untrusted input is interpreted as part of a NoSQL query, allowing an attacker to alter query logic. In Flask applications that use Bearer Tokens for API authentication, the presence of tokens does not inherently protect endpoints from injection if the token is accepted as user-controlled input or if the token is used to gate endpoints that then process untrusted data directly in NoSQL queries.

Consider a Flask route that authenticates via a Bearer Token and then builds a NoSQL query from request parameters. If the route extracts the token from the Authorization header for logging or rate-limiting and later uses request-derived values in a MongoDB query without validation, the token context becomes part of the authentication surface while the query remains vulnerable. For example, a developer might check the token before constructing a query like {'$or': [{request.args.get('username') : request.args.get('password')}]}, effectively letting an authenticated user manipulate query structure. The token itself is not injected, but the route’s logic combines authentication with unsafe query building, creating an opening for injection.

Attackers can exploit this by sending crafted query fragments in parameters, such as username[$ne]=admin or using operator injection like password[$regex]=^.*, to bypass intended filters. If the Bearer Token is accepted as a parameter (e.g., passed as a query or header that the application reuses in query construction), the risk increases because the token may be reflected in logs or error messages, aiding reconnaissance. The combination of token-based access control and unchecked NoSQL input can therefore expose endpoints that would otherwise be protected by simple authentication checks.

OpenAPI/Swagger descriptions that define securitySchemes for Bearer Auth but do not enforce strict input validation on subsequent query parameters can mislead developers into assuming authentication equals safety. middleBrick scans such endpoints during its 12 parallel checks, including Authentication and Input Validation, and can surface mismatches where a Bearer Token gate exists but the underlying query remains injectable. This highlights the importance of treating authenticated requests as untrusted for data handling purposes.

Remediation focuses on strict input validation, parameterized queries, and avoiding the direct inclusion of token-derived or user-derived data into NoSQL query construction. Below are two concrete Flask examples: one vulnerable pattern and one corrected pattern.

Vulnerable Example

from flask import Flask, request, jsonify
from pymongo import MongoClient

app = Flask(__name__)
client = MongoClient('mongodb://localhost:27017')
db = client['mydb']

@app.route('/search')
def search_user():
    token = request.headers.get('Authorization', '').replace('Bearer ', '')
    username = request.args.get('username')
    password = request.args.get('password')
    # Vulnerable: directly embedding user input into query
    user = db.users.find_one({'$or': [{username: password}]})
    return jsonify({'user': str(user) if user else None})

Remediated Example

from flask import Flask, request, jsonify
from pymongo import MongoClient
import re

app = Flask(__name__)
client = MongoClient('mongodb://localhost:27017')
db = client['mydb']

def is_valid_username(value):
    # Allow only alphanumeric and underscores, length 1–64
    return re.match(r'^[A-Za-z0-9_]{1,64}$', value) is not None

def is_valid_token(token):
    # Bearer token format validation, e.g., hex or base64-like
    return re.match(r'^[A-Za-z0-9\-_=]+\.[A-Za-z0-9\-_=]+(\.[A-Za-z0-9\-_=]+)?$', token) is not None

@app.route('/search')
def search_user():
    auth = request.headers.get('Authorization', '')
    if not auth.startswith('Bearer '):
        return jsonify({'error': 'Unauthorized'}), 401
    token = auth[len('Bearer '):]
    if not is_valid_token(token):
        return jsonify({'error': 'Invalid token'}), 401

    username = request.args.get('username')
    password = request.args.get('password')

    if not is_valid_username(username) or not is_valid_username(password):
        return jsonify({'error': 'Invalid input'}), 400

    # Safe: using parameterized queries with dictionary keys constructed safely
    query = {
        '$or': [
            {username: password}
        ]
    }
    # Further safeguard: ensure username field is a known safe field
    allowed_fields = {'admin_user', 'service_user'}
    if username not in allowed_fields:
        return jsonify({'error': 'Forbidden field'}), 403

    user = db.users.find_one(query)
    return jsonify({'user': str(user) if user else None})

Key remediation steps include validating and whitelisting input, using parameterized queries, avoiding reflection of raw user input into query operators, and strictly checking the Authorization header format. These practices reduce the attack surface even when Bearer Tokens are used for authentication.