Ssrf Server Side in Flask with Cockroachdb

Server-side request forgery (SSRF) in a Flask application that uses CockroachDB can arise when user-supplied data is used to form network requests or construct database connection logic. SSRF is listed among the 12 security checks middleBrick runs in parallel, including checks for Input Validation and SSRF itself. In this stack, an attacker may trick the backend into making unintended requests to internal services, the database metadata endpoints, or other sensitive resources reachable from the application server.

Consider a Flask route that accepts a hostname or port from a client to test connectivity to CockroachDB. If the input is not validated and is directly passed to a request library or used to build a database URI, an attacker can supply an internal address (e.g., 169.254.169.254, the AWS metadata service) or a sensitive internal hostname. Even though CockroachDB is a distributed SQL database with its own SQL interface, SSRF here is not about SQL injection; it is about the server-side logic that initiates outbound connections based on untrusted input.

In practice, this can happen when developers build ad‑hoc diagnostic endpoints or configuration tools that construct CockroachDB connection strings from user input. For example, an endpoint that accepts a node_host parameter to ping a specific CockroachDB node could be abused to scan internal services, reach the metadata service, or interact with other internal APIs that are not exposed publicly. middleBrick’s Input Validation and SSRF checks would flag such endpoints by testing whether the application can be induced to make requests to unexpected internal targets.

Another scenario involves server-side template rendering or background tasks where a hostname is read from user-controlled configuration and used to open a database session. If the hostname can be set to an internal service, the Flask app may leak internal network topology or inadvertently trigger requests to services that assume only internal clients can connect. Because CockroachDB often exposes admin APIs and SQL endpoints on specific ports, an SSRF-capable endpoint can become a pivot point for further internal reconnaissance.

Remediation focuses on strict input validation, avoiding dynamic connection construction from user input, and ensuring that any network calls are limited to explicitly allowed endpoints. The following patterns demonstrate secure handling when working with CockroachDB in Flask.

1) Use a fixed connection string with certificate validation

Do not build the CockroachDB URI from user input. Instead, store connection parameters as environment variables or configuration, and validate any user input against a strict allowlist if host selection is required.

import os
import psycopg2
from flask import Flask, request, jsonify

app = Flask(__name__)

# Fixed connection parameters; never concatenate user input into the URI
COCKROACH_URI = os.environ.get(
    "COCKROACH_URI",
    "postgresql://root@cockroachdb-public:26257/defaultdb?sslmode=verify-full"
)

# Example: allowlist of permitted hostnames for multi-tenant scenarios
ALLOWED_HOSTS = {"cluster-east.example.com", "cluster-west.example.com"}

def get_db():
    # Optionally validate a hostname parameter against the allowlist
    hostname = request.args.get("hostname")
    if hostname and hostname not in ALLOWED_HOSTS:
        raise ValueError("Hostname not permitted")
    # Use the fixed URI; if hostname selection is required, map it safely
    effective_uri = COCKROACH_URI
    return psycopg2.connect(effective_uri)

@app.route("/health")
def health():
    try:
        conn = get_db()
        with conn.cursor() as cur:
            cur.execute("SELECT 1")
        return jsonify({"status": "ok"})
    except Exception as e:
        return jsonify({"error": str(e)}), 500

2) Avoid dynamic SQL or schema names from user input

Even when using an allowlist for hostnames, do not inject user input into SQL object names or schema identifiers. Use parameterized queries for data, and validate identifiers against a strict pattern or allowlist.

import psycopg2
from flask import Flask, request, jsonify

app = Flask(__name__)

conn = psycopg2.connect(
    "postgresql://root@cockroachdb-public:26257/defaultdb?sslmode=verify-full"
)

# Safe: parameterized query for data values
@app.route("/user")
def get_user():
    user_id = request.args.get("id")
    if not user_id or not user_id.isdigit():
        return jsonify({"error": "invalid user id"}), 400
    with conn.cursor() as cur:
        cur.execute("SELECT username, email FROM users WHERE id = %s", (user_id,))
        row = cur.fetchone()
    return jsonify(dict(row) if row else {})

# Unsafe pattern to avoid: using string formatting for identifiers
# DO NOT DO THIS:
# query = f"SELECT * FROM {table_name} WHERE id = {user_id}"

3) Harden SSRF surface in Flask

Apply generic SSRF mitigations: disable redirects for HTTP clients, restrict URL schemes to HTTPS/HTTP only, enforce strict timeouts, and avoid sending sensitive headers to user-controlled URLs. If you must accept hostnames, resolve them to IPs and compare against an internal network deny-list (e.g., 127.0.0.0/8, 10.0.0.0/8, 172.16.0.0/12, 192.168.0.0/16, ::1, fc00::/7).

import requests
from flask import Flask, request, jsonify

app = Flask(__name__)

INTERNAL_SUBNETS = {"127.0.0.0/8", "10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16", "::1", "fc00::/7"}

def is_internal(url: str) -> bool:
    from ipaddress import ip_address, IPv4Address, IPv6Address
    from urllib.parse import urlparse
    try:
        parsed = urlparse(url)
        host = parsed.hostname or ""
        ip = ip_address(host)
        for net in INTERNAL_SUBNETS:
            if ip in ip_address(net):  # simplified for example; use ipaddress.ip_network in practice
                return True
    except Exception:
        return False
    return False

@app.route("/fetch")
def fetch_url():
    target = request.args.get("url")
    if not target:
        return jsonify({"error": "missing url"}), 400
    # Reject non-HTTPS/HTTP schemes
    if not target.startswith("http://") and not target.startswith("https://"):
        return jsonify({"error": "unsupported scheme"}), 400
    if is_internal(target):
        return jsonify({"error": "request to internal resource denied"}), 403
    try:
        resp = requests.get(target, timeout=5, allow_redirects=False)
        return jsonify({"status": resp.status_code, "size": len(resp.content)})
    except Exception as e:
        return jsonify({"error": str(e)}), 500