Data Exposure in Fastapi with Cockroachdb

The combination of FastAPI and CockroachDB can inadvertently expose sensitive data when application logic does not enforce proper authorization and data scoping. Data Exposure in this context refers to situations where one user can read, modify, or enumerate data that belongs to another user. With CockroachDB, a distributed SQL database, the risk often arises from how queries are constructed and how rows are filtered in the application layer rather than being enforced at the database level.

Consider an endpoint that retrieves a user profile by ID. If the endpoint uses a direct lookup without verifying that the requesting user is authorized to view that specific profile, an Insecure Direct Object Reference (IDOR) can occur. For example, an attacker can iterate through numeric or predictable identifiers and access other users' data because the query does not include a tenant or user scope.

In FastAPI, a typical vulnerable route might look like this:

from fastapi import FastAPI, Depends, HTTPException
import cockroachdb

app = FastAPI()

def get_db():
    # Returns a database connection; credentials handled externally
    conn = cockroachdb.connect(
        host="localhost",
        port 26257,
        database="mydb",
        user="app_user"
    )
    return conn

@app.get("/users/{user_id}")
def read_user(user_id: int, db = Depends(get_db)):
    cursor = db.cursor()
    cursor.execute("SELECT id, email, name, ssn FROM users WHERE id = %s", (user_id,))
    row = cursor.fetchone()
    if row is None:
        raise HTTPException(status_code=404, detail="User not found")
    return {"id": row[0], "email": row[1], "name": row[2], "ssn": row[3]}

In this example, the query does not check whether the authenticated user is the same as the requested user_id. If authentication is present but authorization is missing, an attacker can supply any integer ID and potentially access sensitive fields such as SSN. CockroachDB will return the row if it exists, and FastAPI will expose it without additional checks.

Data Exposure is also relevant when responses include fields that should never be returned to certain roles. For instance, an endpoint that lists users might unintentionally return admin-level fields (e.g., is_superuser, password_hash) to non-privileged clients. Because CockroachDB scans the table according to the query’s SELECT clause, if the application requests broad columns without role-based filtering, sensitive data can be leaked through the API response.

Another scenario involves missing row-level security policies. CockroachDB supports row-level security, but if the application relies solely on the database to enforce scoping and the FastAPI layer does not validate ownership, indirect exposures can occur. For example, an attacker might manipulate query parameters to request data they should not see, and without explicit tenant or user filtering in the SQL, the database may return cross-user data.

To mitigate Data Exposure in this stack, developers must ensure that every data access path includes explicit ownership or tenant checks, minimizes returned sensitive fields, and validates that the requesting context aligns with the requested resource. Relying on the database to return only what the requester is allowed to see is essential, and this must be enforced in the application code and query design.

Remediation focuses on embedding authorization into queries, reducing exposed fields, and leveraging CockroachDB capabilities where appropriate. The following examples demonstrate secure patterns for FastAPI with CockroachDB.

1. Enforce user ownership in the WHERE clause

Always include the authenticated user’s identifier in the query. This ensures that even if an attacker manipulates the URL parameter, the database will not return data that does not belong to them.

from fastapi import FastAPI, Depends, HTTPException
import cockroachdb
from typing import Optional

app = FastAPI()

# Simulated auth dependency
def get_current_user():
    return {"user_id": 123, "role": "user"}

def get_db():
    conn = cockroachdb.connect(
        host="localhost",
        port=26257,
        database="mydb",
        user="app_user"
    )
    return conn

@app.get("/users/{user_id}")
def read_user(user_id: int,
               current_user = Depends(get_current_user),
               db = Depends(get_db)):
    cursor = db.cursor()
    # Enforce ownership at the database level
    cursor.execute(
        "SELECT id, email, name FROM users WHERE id = %s AND owner_id = %s",
        (user_id, current_user["user_id"])
    )
    row = cursor.fetchone()
    if row is None:
        raise HTTPException(status_code=404, detail="User not found or access denied")
    return {"id": row[0], "email": row[1], "name": row[2]}

2. Apply role-based field filtering

Select only the columns that are necessary for the caller’s role. Avoid SELECT * and explicitly exclude sensitive fields for non-admin roles.

def get_user_projection(role: str):
    if role == "admin":
        return "id, email, name, is_superuser, created_at"
    return "id, email, name"

@app.get("/users/me")
def read_me(
    current_user = Depends(get_current_user),
    db = Depends(get_db)
):
    cursor = db.cursor()
    fields = get_user_projection(current_user["role"])
    cursor.execute(f"SELECT {fields} FROM users WHERE id = %s", (current_user["user_id"],))
    row = cursor.fetchone()
    if row is None:
        raise HTTPException(status_code=404, detail="User not found")
    return dict(zip(["id", "email", "name", "is_superuser", "created_at"][:len(row)], row))

3. Use parameterized queries to prevent injection and ensure scoping

Never concatenate identifiers or values. Use placeholders and pass parameters separately to maintain correctness and security.

@app.get("/accounts")
def list_accounts(
    account_type: Optional[str] = None,
    current_user = Depends(get_current_user),
    db = Depends(get_db)
):
    cursor = db.cursor()
    base_query = "SELECT id, name, currency FROM accounts WHERE owner_id = %s"
    params: list = [current_user["user_id"]]
    if account_type:
        base_query += " AND type = %s"
        params.append(account_type)
    cursor.execute(base_query, tuple(params))
    rows = cursor.fetchall()
    return [{"id": r[0], "name": r[1], "currency": r[2]} for r in rows]

4. Leverage CockroachDB placeholders correctly

CockroachDB uses the $1, $2 style for positional placeholders in some driver configurations. Ensure your driver and query style match. The examples above use %s which works with many Python drivers (e.g., psycopg2) that translate to the correct CockroachDB placeholder syntax. If using raw SQL with placeholders, verify the driver’s placeholder style to avoid errors.

# Example using positional placeholders if required by driver
cursor.execute(
    "SELECT id, email FROM users WHERE id = $1 AND owner_id = $2",
    (user_id, current_user["user_id"])
)

5. Avoid exposing internal details in error messages

Do not return database constraint details or raw errors to the client. Use generic messages and log detailed errors server-side.

@app.get("/items/{item_id}")
def read_item(item_id: int, current_user = Depends(get_current_user), db = Depends(get_db)):
    cursor = db.cursor()
    cursor.execute(
        "SELECT id, name, owner_id FROM items WHERE id = %s AND owner_id = %s",
        (item_id, current_user["user_id"])
    )
    row = cursor.fetchone()
    if row is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"id": row[0], "name": row[1]}