Type Confusion in Fastapi with Cockroachdb

Type confusion in a FastAPI service that uses CockroachDB typically arises when application code or an OpenAPI specification does not enforce strict types for request, response, or path parameters, and those values are passed directly to database operations. CockroachDB, like other SQL databases, does not understand Python types; it receives values via drivers that rely on correct parameterization. If FastAPI allows a route to accept loosely typed input (for example, a path parameter that could be either an integer ID or a string slug) and that input is used in constructing dynamic queries or ORM filters, the runtime behavior can diverge from the expected schema. This mismatch can lead to information leaks, unintended record access, or errors that reveal stack traces or internal details.

Consider an endpoint designed to fetch a user by ID, where the path parameter is declared as int in the OpenAPI spec but is not validated or coerced before use in an SQL string or a raw query. An attacker could send a string such as 1 OR 1=1 if the framework does not strictly enforce types, potentially causing the query to return unexpected rows. Even when using an ORM like SQLAlchemy with CockroachDB, if the model field types are not aligned with the input expectations, a parameter meant to be a UUID could be interpreted as a string, bypassing intended filters and enabling horizontal privilege escalation across tenant boundaries (BOLA/IDOR).

Another scenario involves object-level authorization where a FastAPI dependency resolves a resource by a numeric primary key, but the underlying CockroachDB table uses composite keys or encoded identifiers. If the dependency does not validate that the provided key matches the expected type and structure, an attacker may substitute a different type (e.g., a JSON object serialized as a string) that passes initial checks but leads to unauthorized data retrieval. The OpenAPI/Swagger spec analysis performed by middleBrick would highlight mismatches between declared parameter schemas and runtime behavior, especially when $ref definitions are not consistently resolved across paths and components.

With LLM endpoints exposed in the same service, type confusion can be compounded if model inputs are not strictly validated. An LLM endpoint that accepts free-form text could receive crafted input designed to inject type-bypass patterns into downstream database calls. While middleBrick’s LLM/AI Security checks do not fix these issues, they can surface related risks such as system prompt leakage or unauthorized tool usage when type confusion affects prompt handling logic.

Because middleBrick scans the unauthenticated attack surface and includes OpenAPI/Swagger spec analysis with full $ref resolution, it can identify inconsistencies between declared types in the spec and the observed runtime behavior. Findings are mapped to frameworks such as OWASP API Top 10 and include severity and remediation guidance, though the scanner does not fix or block requests.

To prevent type confusion when FastAPI interacts with CockroachDB, enforce strict type validation at the edge of the application and use parameterized queries or an ORM with typed models. Below are concrete, working examples that demonstrate secure patterns.

1. Path parameter strict typing with Pydantic models

Define request and path models with explicit types and use them in your route. This ensures that FastAPI validates and coerces input before it reaches business logic or database code.

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel, Field
from typing import Optional
import asyncpg

app = FastAPI()

class UserPath(BaseModel):
    user_id: int = Field(..., gt=0, description="Must be a positive integer")

class UserResponse(BaseModel):
    id: int
    username: str
    email: str

async def get_db_connection():
    # asyncpg is compatible with CockroachDB
    return await asyncpg.connect("postgresql://user:password@host:26257/dbname")

@app.get("/users/{user_id}", response_model=UserResponse)
async def read_user(path: UserPath, conn: asyncpg.Connection = Depends(get_db_connection)):
    try:
        row = await conn.fetchrow("SELECT id, username, email FROM users WHERE id = $1", path.user_id)
        if row is None:
            raise HTTPException(status_code=404, detail="User not found")
        return UserResponse(id=row["id"], username=row["username"], email=row["email"])
    finally:
        await conn.close()

2. Parameterized queries with SQLAlchemy ORM and typed models

When using SQLAlchemy with CockroachDB, define models with explicit types and use the ORM query interface to avoid string concatenation.

from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from fastapi import FastAPI, HTTPException, Depends

Base = declarative_base()

class UserORM(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    email = Column(String)

engine = create_engine("cockroachdb://user:password@host:26257/dbname?sslmode=require")
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
app = FastAPI()

@app.get("/users/{user_id}")
def read_user(user_id: int, db = SessionLocal()):
    try:
        user = db.query(UserORM).filter(UserORM.id == user_id).first()
        if user is None:
            raise HTTPException(status_code=404, detail="User not found")
        return {"id": user.id, "username": user.username, "email": user.email}
    finally:
        db.close()

3. Avoid dynamic SQL and raw string interpolation

Never build queries by formatting strings with user input. Use bound parameters or the ORM’s filter methods to ensure types are handled safely by the driver and database.

# Unsafe — do not do this
# query = f"SELECT * FROM users WHERE id = {user_input}"

# Safe — parameterized query
await conn.fetchrow("SELECT * FROM users WHERE id = $1", user_id)

# Safe — SQLAlchemy filter
user = db.query(UserORM).filter(UserORM.id == user_input).first()

4. Align OpenAPI spec with runtime behavior

Ensure that path, query, and body parameters declared in your OpenAPI/Swagger definitions match the Python types used in your route handlers. Use middleBrick’s OpenAPI/Swagger spec analysis (with full $ref resolution) to detect mismatches that could lead to type confusion in production.

# Example OpenAPI snippet (YAML)
# paths:
#   /users/{user_id}:
#     get:
#       parameters:
#         - name: user_id
#           in: path
#           required: true
#           schema:
#             type: integer
#             minimum: 1

By combining strict typing in FastAPI, parameterized database interactions, and continuous spec validation, you reduce the risk of type confusion when working with CockroachDB.