Side Channel Attack in Fastapi with Cockroachdb

A side channel attack in a Fastapi service that uses Cockroachdb exploits timing, behavioral, or incidental information leaks rather than direct SQL injection. In this stack, an attacker can infer sensitive data by observing differences in response times, error messages, or query patterns while the application interacts with Cockroachdb. Because Cockroachdb is a distributed SQL database, certain operations—such as distributed transactions or range lookups—can introduce measurable variations in latency that a client-facing Fastapi endpoint may inadvertently expose.

For example, consider a Fastapi endpoint that searches for a user by username and then fetches additional profile data from Cockroachdb. If the endpoint performs two separate queries—one to check existence and another to retrieve details—an attacker can measure response times to infer whether a username exists. Even with parameterized queries preventing injection, the timing discrepancy between a missing user (early exit or empty result) and a found user (join or index scan across nodes) becomes a side channel. Cockroachdb’s consensus and replication mechanisms can amplify these differences due to network hops and Raft group coordination, making timing variance more pronounced than in single-node databases.

Another vector arises from error handling. Fastapi may surface Cockroachdb-specific error codes or stack traces in development configurations. Distinct error paths—such as a not-null violation versus a unique constraint violation—can reveal schema details or data presence. If authentication endpoints return different status codes or messages for "user not found" versus "incorrect password," an attacker can use these signals to perform user enumeration. The distributed nature of Cockroachdb means certain failures (e.g., lease transfers or range splits) may produce variable latency or transient errors, further enriching side channel information.

Insecure use of asynchronous features in Fastapi can also contribute. If endpoints trigger background tasks that interact with Cockroachdb without proper synchronization, an attacker might infer internal state by polling and observing when side effects complete. Similarly, unauthenticated endpoints that query Cockroachdb for public data can leak access patterns; an LLM/AI Security probe from middleBrick might detect whether an endpoint is unauthenticated and observe timing as part of its active testing sequence.

Because middleBrick runs black-box scans without credentials, it can surface timing-related anomalies and error leakage for unauthenticated endpoints. While it does not fix these issues, its findings include remediation guidance to help developers harden Fastapi routes and reduce side channel exposure when working with Cockroachdb.

Remediation focuses on making operations uniform in timing and error handling, and avoiding information leaks to attackers. Use constant-time patterns, avoid branching on sensitive data, and ensure exceptions do not reveal internal details.

Example 1: Constant-time user existence check

Instead of branching based on whether a user exists, always perform a dummy query to mask timing differences.

from fastapi import Fastapi, HTTPException
from cockroachdb import connect
import asyncio

app = Fastapi()

async def get_db_connection():
    return connect("cockroachdb://user:password@host:26257/mydb")

@app.get("/users/{username}")
async def get_user(username: str):
    conn = await get_db_connection()
    try:
        # Always run the same shape of query
        cursor = await conn.execute(
            "SELECT data FROM users WHERE username = $1",
            (username,)
        )
        row = await cursor.fetchone()
        # Perform a dummy query to normalize timing when user not found
        if row is None:
            await conn.execute(
                "SELECT 1 FROM users WHERE username = $1",
                ("dummy_user_xyz",)
            )
            raise HTTPException(status_code=404, detail="Not found")
        return {"username": username, "data": row[0]}
    except Exception as e:
        # Log internally; return generic error to avoid leaking details
        raise HTTPException(status_code=500, detail="Request failed")
    finally:
        await conn.close()

Example 2: Uniform error handling with Cockroachdb transactions

Wrap operations in transactions and ensure exceptions map to consistent HTTP responses.

from fastapi import Fastapi
from cockroachdb import connect, errors as cockroach_errors
import asyncio

app = Fastapi()

async def exec_transaction(conn, queries_and_params):
    async with conn.transaction() as txn:
        for sql, params in queries_and_params:
            await txn.execute(sql, params)

@app.post("/transfer")
async def transfer(from_id: str, to_id: str, amount: int):
    conn = await get_db_connection()
    try:
        queries = [
            ("UPDATE accounts SET balance = balance - $1 WHERE id = $2", (amount, from_id)),
            ("UPDATE accounts SET balance = balance + $1 WHERE id = $2", (amount, to_id)),
        ]
        await exec_transaction(conn, queries)
        return {"status": "ok"}
    except cockroach_errors.UniqueViolation:
        # Do not reveal which constraint failed
        raise HTTPException(status_code=400, detail="Invalid request")
    except cockroach_errors.SerializationError:
        # Retries are common in distributed SQL; keep response generic
        raise HTTPException(status_code=409, detail="Conflict, please retry")
    except Exception:
        # Generic catch to avoid leaking Cockroachdb internals
        raise HTTPException(status_code=500, detail="Request failed")
    finally:
        await conn.close()

Example 3: Avoiding information leakage in logs and responses

Ensure Cockroachdb error codes are not surfaced, and use structured logging that excludes sensitive context.

import logging
from fastapi import Fastapi
from cockroachdb import connect

app = Fastapi()
logger = logging.getLogger("secure_app")

@app.get("/profile")
async def profile(user_id: str):
    conn = await get_db_connection()
    try:
        cursor = await conn.execute(
            "SELECT email, settings FROM profiles WHERE user_id = $1",
            (user_id,)
        )
        row = await cursor.fetchone()
        if row is None:
            # Same generic response and log pattern regardless of existence
            logger.info("profile_lookup", extra={"user_id": user_id, "found": False})
            raise HTTPException(status_code=404, detail="Not found")
        logger.info("profile_lookup", extra={"user_id": user_id, "found": True})
        return {"email": row[0], "settings": row[1]}
    except Exception as e:
        # Log without exposing Cockroachdb-specific messages
        logger.warning("profile_lookup_error", extra={"user_id": user_id, "error_type": type(e).__name__})
        raise HTTPException(status_code=500, detail="Request failed")
    finally:
        await conn.close()

Operational practices

• Use environment-based configuration to ensure development errors are not exposed in production.
• Apply consistent timeouts and context cancellations to reduce timing variability across distributed Cockroachdb nodes.
• Validate and normalize inputs before building queries to avoid leaking distinctions via parsing errors.

These measures reduce the attack surface for side channel observations while preserving compatibility with Cockroachdb’s distributed semantics.