Vulnerable Components in Fastapi with Cockroachdb

When FastAPI applications interact with CockroachDB, several common implementation patterns introduce security risks that middleBrick detects across its 12 security checks. The combination creates vulnerabilities primarily through three vectors: connection handling, query construction, and data exposure.

Connection String Exposure

CockroachDB connection strings often contain sensitive parameters that developers inadvertently expose through FastAPI configuration. When connection details appear in environment variables without proper protection or are logged during debugging, middleBrick flags this as a Data Exposure finding. A typical vulnerable pattern involves hardcoding connection parameters in application code:

from fastapi import FastAPI
import cockroachdb

# Vulnerable: connection details in code
DATABASE_URL = "postgresql://root@localhost:26257/defaultdb?sslmode=require"

def get_db():
    return cockroachdb.connect(DATABASE_URL)

Improper Query Parameterization

FastAPI path parameters combined with CockroachDB queries create BOLA/IDOR risks when developers concatenate user input directly into SQL statements. Consider this vulnerable endpoint:

from fastapi import FastAPI, Path
import cockroachdb

app = FastAPI()

@app.get("/users/{user_id}")
def get_user(user_id: int = Path(...)):
    conn = cockroachdb.connect("postgresql://root@localhost:26257/defaultdb")
    cursor = conn.cursor()
    # Vulnerable: direct string concatenation enables IDOR
    query = f"SELECT username, email FROM users WHERE id = {user_id}"
    cursor.execute(query)
    return cursor.fetchone()

This pattern allows attackers to enumerate user IDs horizontally (accessing other users' data) or vertically (accessing administrative records). middleBrick's BOLA/IDOR check identifies such patterns by correlating path parameters with database query construction.

Over-privileged Database Connections

CockroachDB's role-based access control is frequently misconfigured in FastAPI applications. When the application connects using a superuser account or overly permissive roles, any injected vulnerability becomes critical. middleBrick's Authentication and Privilege Escalation checks identify when database connections lack principle of least privilege:

import os
import cockroachdb

# Vulnerable: application uses root-like privileges
connection_params = {
    "host": os.getenv("DB_HOST"),
    "port": os.getenv("DB_PORT", 26257),
    "database": os.getenv("DB_NAME"),
    "user": "root",  # Over-privileged
    "ssl_mode": "require"
}

def create_connection():
    return cockroachdb.connect(**connection_params)

Additionally, unauthenticated LLM security checks detect when database connection strings or sensitive query patterns might be exposed through AI-assisted development tools or logging mechanisms.

Secure implementation requires addressing connection management, query construction, and privilege separation specifically for CockroachDB's distributed architecture.

Secure Connection Management with Environment Isolation

Use CockroachDB's secure connection string format with certificate-based authentication and store credentials outside application code:

import os
from fastapi import Depends, FastAPI
import cockroachdb
from dotenv import load_dotenv

load_dotenv()

app = FastAPI()

def get_db_connection():
    # Secure: certificates and credentials from secure sources
    return cockroachdb.connect(
        host=os.getenv("DB_HOST"),
        port=int(os.getenv("DB_PORT", 26257)),
        database=os.getenv("DB_NAME"),
        user=os.getenv("DB_USER"),
        password=os.getenv("DB_PASSWORD"),
        ssl_mode="verify-full",
        ssl_root_cert="/path/to/ca.pem",
        ssl_cert="/path/to/client.pem",
        ssl_key="/path/to/client.key"
    )

@app.get("/secure-endpoint")
def secure_endpoint(db=Depends(get_db_connection)):
    # Connection uses certificate-based authentication
    with db.cursor() as cursor:
        cursor.execute("SELECT current_user()")
        return {"database_user": cursor.fetchone()[0]}

Parameterized Queries with Type Validation

Prevent BOLA/IDOR by using CockroachDB's parameterized queries with FastAPI's type validation:

from fastapi import FastAPI, HTTPException, Path
import cockroachdb

app = FastAPI()

@app.get("/users/{user_id}")
def get_user(
    user_id: int = Path(..., gt=0, description="Valid user ID")
):
    conn = get_db_connection()
    try:
        with conn.cursor() as cursor:
            # Secure: parameterized query prevents IDOR
            cursor.execute(
                "SELECT username, email, tenant_id FROM users WHERE id = %s",
                (user_id,)
            )
            result = cursor.fetchone()
            if not result:
                raise HTTPException(status_code=404, detail="User not found")
            return {
                "username": result[0],
                "email": result[1],
                "tenant_id": result[2]
            }
    finally:
        conn.close()

Role-Based Access Control Implementation

Configure CockroachDB roles specific to FastAPI application needs and implement connection pooling with appropriate privileges:

import os
import cockroachdb
from contextlib import contextmanager

# Application-specific role with minimal privileges
@contextmanager
def app_db_session():
    conn = cockroachdb.connect(
        host=os.getenv("DB_HOST"),
        database=os.getenv("DB_NAME"),
        user=os.getenv("DB_APP_USER"),  # Dedicated application user
        password=os.getenv("DB_APP_PASSWORD"),
        ssl_mode="verify-full"
    )
    try:
        yield conn
    finally:
        conn.close()

# CockroachDB setup (run separately):
# CREATE ROLE api_user WITH LOGIN PASSWORD 'secure_password';
# GRANT SELECT ON users TO api_user;
# GRANT USAGE ON SCHEMA public TO api_user;

middleBrick's Pro plan continuous monitoring can detect when database privilege changes occur and alert on permission escalations. The GitHub Action integration can validate these connection patterns during CI/CD pipeline runs.