Ssrf Server Side in Flask with Mongodb

Server-side request forgery (SSRF) in a Flask application that uses MongoDB can arise when user-controlled input is used to construct network requests or when internal endpoints are indirectly exposed. A common pattern is an endpoint that fetches data from a service and then stores or logs that data into MongoDB. If the target service is user-supplied and not strictly validated, an attacker can direct requests to internal resources such as the Flask metadata service (http://169.254.169.254), cloud instance metadata, or internal MongoDB connections that are not bound to external interfaces.

In this stack, SSRF can be triggered via HTTP parameters that downstream code uses to build a request, and the consequences are compounded if the application later writes the response into MongoDB without strict schema and type checks. For example, an attacker might supply a malicious internal IP or a local file URL, causing the server to make unintended requests and potentially leak sensitive data. Because MongoDB connections in Flask are often established with connection strings or URIs, if those values are derived from user input or from an SSRF-tainted response, the attack surface extends to the database layer.

Moreover, if the Flask app exposes administrative or diagnostic routes that interact with MongoDB (e.g., listing collections or running ad hoc queries), SSRF can be used to reach these internal endpoints through a proxied request. The combination of Flask’s flexible routing and MongoDB’s flexible document model can inadvertently allow an attacker to probe internal services, enumerate network topology, or exfiltrate data through side channels.

Remediation focuses on strict input validation, network segregation, and safe MongoDB interactions. Avoid constructing MongoDB URIs or query filters from unvalidated user input. Use allowlists for expected values and reject unexpected schemes or hosts in URLs used for outbound requests.

Input validation and URL allowlisting

Validate and sanitize any user input used to form URLs or database filters. For outbound requests, use an allowlist of permitted hosts and paths, and reject URLs with private IP ranges, cloud metadata endpoints, or non-HTTP(S) schemes.

import re
from urllib.parse import urlparse

def is_safe_url(url: str) -> bool:
    # Allow only specific external services
    allowed_hosts = {"api.example.com", "data.example.com"}
    parsed = urlparse(url)
    if parsed.scheme not in {"http", "https"}:
        return False
    if parsed.hostname is None:
        return False
    # Reject private IPs and localhost
    private_ip_pattern = re.compile(
        r"^(127\.0\.0\.1|localhost|\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}|" 
        r"10\\.|172\\.(1[6-9]|2\\d|3[01])\\.|192\\.168\\.)", re.I
    )
    if private_ip_pattern.match(parsed.hostname):
        return False
    if parsed.hostname not in allowed_hosts:
        return False
    return True

Safe MongoDB interaction in Flask

Use a trusted MongoDB connection string managed by the server environment, and avoid dynamically building URIs from user input. When querying, use schema validation and parameterized queries to avoid injection and unintended document access.

from flask import Flask, request, jsonify
from pymongo import MongoClient
from bson.objectid import ObjectId

app = Flask(__name__)
# Use a fixed connection string from configuration or environment
client = MongoClient("mongodb://localhost:27017/")
db = client["mydatabase"]
collection = db["items"]

@app.route("/item", methods=["GET"])
def get_item():
    item_id = request.args.get("id")
    if not item_id:
        return jsonify({"error": "missing id"}), 400
    # Validate ObjectId format strictly
    if not ObjectId.is_valid(item_id):
        return jsonify({"error": "invalid id format"}), 400
    doc = collection.find_one({"_id": ObjectId(item_id)})
    if doc is None:
        return jsonify({"error": "not found"}), 404
    # Exclude internal fields before returning
    doc.pop("_id", None)
    return jsonify(doc)

@app.route("/search", methods=["GET"])
def search_items():
    q = request.args.get("q", "")
    # Use parameterized query; avoid building dynamic filter objects from raw input
    pipeline = [
        {"$match": {"$text": {"$search": q}}},
        {"$project": {"name": 1, "description": 1, "_id": 0}}
    ]
    results = list(collection.aggregate(pipeline))
    return jsonify(results)

Defensive SSRF mitigations

• Do not follow redirects to private IPs; disable redirect handling or validate each redirect location.
• Use a network-level firewall or service mesh to prevent outbound connections to metadata endpoints from application containers.
• Apply principle of least privilege: the MongoDB user used by Flask should have minimal required permissions and should not be able to change its own URI or configuration.