Out Of Bounds Write in Fastapi with Cockroachdb

An Out Of Bounds Write occurs when an API writes data to a memory location outside the intended buffer. In a Fastapi service that uses Cockroachdb, the risk typically arises at the application layer rather than in Cockroachdb itself, but the database interaction can amplify impact if user-controlled input is used to index into buffers or to determine iteration ranges for bulk operations.

Fastapi does not manage memory directly, but Python’s underlying buffers (e.g., bytearray, array.array, or third-party C extensions) can be exposed through endpoint logic. If an endpoint constructs a buffer based on a request parameter and writes payload bytes into it, a missing bounds check can allow writes beyond the allocated size.

When combined with Cockroachdb, an Out Of Bounds Write may surface in data handling patterns where request parameters influence SQL operations indirectly. For example, using a user-supplied integer to control batch sizes or to compute offsets for slicing bytea columns can lead to unsafe iteration or slicing. If the API builds SQL queries by concatenating such values, an attacker can craft inputs that cause the application to write data beyond expected structures, potentially corrupting in-memory representations before sending statements to Cockroachdb.

Cockroachdb’s wire protocol and SQL interface do not introduce buffer overflow vulnerabilities, but the database can reflect unsafe application behavior. For instance, if an endpoint deserializes JSON into a fixed-size structure and then forwards data to Cockroachdb using dynamic SQL, an oversized field can trigger an out of bounds condition during processing. This may lead to unpredictable behavior, including data exposure or crashes, before the query reaches Cockroachdb.

Consider an endpoint that accepts an index and a byte payload to update a specific segment of a bytea column. If the index is not validated against the length of the existing data, an attacker can specify an index that points outside the current bounds, causing the application to write into adjacent memory when reconstructing the buffer. The resulting malformed data is then sent in an UPDATE statement to Cockroachdb, which may accept it syntactically but propagate corruption at the application level.

Real-world attack patterns include manipulation of Content-Range headers, file upload processing, or streaming endpoints where chunk indices are not properly constrained. The OWASP API Top 10 category ‘2023:1 – Broken Object Level Authorization’ can intersect with this vulnerability when bounds issues enable unauthorized data modification across resources.

To detect such issues, middleBrick runs parallel checks that include Input Validation and Unsafe Consumption tests. These scans identify endpoints where user input influences buffer construction or SQL parameterization without proper length verification, providing findings mapped to OWASP and highlighting the exact location of risky parameter usage.

Remediation focuses on strict input validation, safe buffer handling, and robust SQL construction. Below are concrete Fastapi examples that prevent out of bounds writes when interacting with Cockroachdb.

1. Validate indices before buffer manipulation

Ensure any index derived from request data is within the legal range of the target buffer or data structure.

from fastapi import FastAPI, HTTPException, Query
import psycopg2
from psycopg2.extensions import connection

app = FastAPI()

def get_db() -> connection:
    return psycopg2.connect(
        host="localhost",
        database="mydb",
        user="app",
        password="securepass",
    )

@app.put("/segments/{seg_id}")
async def update_segment(
    seg_id: int,
    index: int = Query(..., ge=0),
    payload: bytes = Query(...)
):
    if index < 0 or len(payload) == 0:
        raise HTTPException(status_code=400, detail="Invalid index or empty payload")
    # Fetch current segment to determine valid bounds
    conn = get_db()
    try:
        with conn.cursor() as cur:
            cur.execute("SELECT data FROM segments WHERE id = %s", (seg_id,))
            row = cur.fetchone()
            if not row:
                raise HTTPException(status_code=404, detail="Segment not found")
            current = row[0]  # type: bytes
            if index + len(payload) > len(current):
                raise HTTPException(status_code=400, detail="Write would exceed segment bounds")
            # Safe: index and payload length are within bounds
            new_data = current[:index] + payload + current[index + len(payload):]
            cur.execute("UPDATE segments SET data = %s WHERE id = %s", (new_data, seg_id))
            conn.commit()
    finally:
        conn.close()
    return {"status": "updated"}

2. Use parameterized queries to avoid injection and malformed data

Never construct SQL strings with user input. Use placeholders and let the driver handle encoding.

from fastapi import FastAPI, Depends
import psycopg2

app = FastAPI()

def get_db():
    return psycopg2.connect(
        host="localhost",
        database="mydb",
        user="app",
        password="securepass",
    )

@app.post("/data/{record_id}")
async def store_data(
    record_id: int,
    chunk_index: int,
    chunk_data: bytes,
    conn: psycopg2.connection = Depends(get_db)
):
    if chunk_index < 0:
        raise HTTPException(status_code=400, detail="chunk_index must be non-negative")
    with conn.cursor() as cur:
        # Use parameterized query to prevent injection and ensure type safety
        cur.execute(
            "INSERT INTO chunks (record_id, idx, data) VALUES (%s, %s, %s) "
            "ON CONFLICT (record_id, idx) DO UPDATE SET data = EXCLUDED.data",
            (record_id, chunk_index, chunk_data)
        )
        conn.commit()
    return {"record_id": record_id, "index": chunk_index}

3. Leverage Pydantic models for structured input

Define strict schemas to enforce length and type constraints before data reaches SQL logic.

from fastapi import FastAPI
from pydantic import BaseModel, Field
import psycopg2

app = FastAPI()

class SegmentUpdate(BaseModel):
    index: int = Field(ge=0, description="Zero-based start index")
    payload: bytes = Field(min_length=1, max_length=4096, description="Payload must fit within segment")

@app.patch("/segments/{seg_id}/data")
async def patch_data(seg_id: int, update: SegmentUpdate, conn: psycopg2.connection = Depends(get_db)):
    with conn.cursor() as cur:
        cur.execute("SELECT data FROM segments WHERE id = %s FOR UPDATE", (seg_id,))
        row = cur.fetchone()
        if not row:
            raise HTTPException(status_code=404, detail="Segment not found")
        current = row[0]
        if update.index + len(update.payload) > len(current):
            raise HTTPException(status_code=400, detail="Payload exceeds segment bounds")
        new_data = current[:update.index] + update.payload + current[update.index + len(update.payload):]
        cur.execute("UPDATE segments SET data = %s WHERE id = %s", (new_data, seg_id))
        conn.commit()
    return {"result": "ok"}

4. Apply principle of least privilege and schema constraints

Ensure the Cockroachdb user used by Fastapi has only necessary permissions. Combine database constraints (e.g., CHECK) with application-level validation to reduce risk.

-- Example Cockroachdb schema constraint (run once)
ALTER TABLE segments ADD CONSTRAINT valid_data_length CHECK (octet_length(data) = 65536);

These measures align with secure coding practices and help ensure that user input never drives unsafe memory operations or SQL behavior.