Bola Idor in Fastapi with Cockroachdb

Broken Object Level Authorization (BOLA) occurs when an API exposes direct object references (IDs) without verifying that the requesting actor has permission to access the targeted resource. In a FastAPI application backed by CockroachDB, the vulnerability typically arises at the intersection of predictable identifiers, missing ownership checks, and overly permissive query patterns.

Consider a typical endpoint that retrieves a user profile using a path parameter such as /users/{user_id}. If the route uses a raw user_id to construct a CockroachDB SQL query without confirming the caller owns that record, an attacker can change the ID to another valid integer or UUID and access arbitrary user data. Because CockroachDB uses a SQL wire protocol and supports standard SELECT statements, an attacker does not need to exploit a database misconfiguration—only the lack of authorization logic at the API layer.

FastAPI’s dependency injection and path parameter parsing make it easy to bind user_id directly into a query. Developers may write code like session.execute(select(User).filter(User.id == user_id)) without validating that the authenticated subject has rights to that specific user_id. This pattern is especially risky when IDs are sequential integers or UUIDs that can be enumerated. The API’s unauthenticated attack surface, tested by middleBrick in its black-box scanning, can expose endpoints where object-level authorization is absent, leading to unauthorized data disclosure.

Another common pattern involves tenant or organization scoping. Suppose a tenant_id is stored in the database but the API does not include it in the filter predicate. An authenticated user in tenant A could supply a record_id belonging to tenant B, and the query would return data because the backend does not enforce tenant_id equality. Cockroachdb’s strong consistency and SQL semantics mean that the query will succeed and return data if the row exists, regardless of tenant boundaries. middleBrick’s checks for BOLA/IDOR highlight such cases by correlating spec definitions with runtime behavior, showing where path or query parameters lack contextual authorization.

Additionally, HTTP method confusion can compound BOLA. For example, an endpoint accepting PUT /records/{id} might inadvertently allow a user to modify another user’s record when the identifier is manipulated. Because CockroachDB does not inherently enforce object-level permissions, the responsibility falls on the application to ensure that each mutation validates the subject’s relationship to the object. Without explicit checks, even read-only endpoints can become vectors for horizontal privilege escalation.

To mitigate BOLA in FastAPI with CockroachDB, enforce authorization checks before constructing SQL queries. Always scope queries by the subject’s tenant or user identifier and use parameterized statements to avoid injection while ensuring predicate correctness. Below are concrete, working examples that you can adopt to close the gap.

1. User-scoped query with authenticated subject

Assume you have an authenticated subject available via a dependency that provides a user ID. The endpoint should include the subject ID in the filter, not rely solely on the path parameter.

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy import select, create_engine
from sqlalchemy.orm import Session, declarative_base, sessionmaker

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

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    username = Column(String)
    tenant_id = Column(Integer)

def get_current_user():
    # Replace with real auth logic returning subject identity
    return {"user_id": 1, "tenant_id": 100}

@app.get("/users/{user_id}")
def read_user(user_id: int, db: Session = Depends(get_db), current_user: dict = Depends(get_current_user)):
    stmt = select(User).filter(User.id == user_id, User.tenant_id == current_user["tenant_id"])
    result = db.execute(stmt)
    user = result.scalar_one_or_none()
    if user is None:
        raise HTTPException(status_code=404, detail="Not found or access denied")
    return {"id": user.id, "username": user.username}

2. Explicit tenant check for related objects

When accessing child records, verify that the parent tenant matches the subject’s tenant. This prevents cross-tenant IDOR via associated resources.

@app.get("/users/{user_id}/posts/{post_id}")
def read_post(user_id: int, post_id: int, db: Session = Depends(get_db), current_user: dict = Depends(get_current_user)):
    stmt = select(Post).join(User).filter(
        Post.id == post_id,
        User.id == user_id,
        User.tenant_id == current_user["tenant_id"]
    )
    result = db.execute(stmt)
    post = result.scalar_one_or_none()
    if post is None:
        raise HTTPException(status_code=404, detail="Not found or access denied")
    return {"post_id": post.id, "title": post.title}

3. Using parameterized queries with CockroachDB placeholders

Always use bound parameters rather than string formatting. The following pattern works with CockroachDB’s PostgreSQL-compatible wire protocol and ensures that filters are applied correctly.

from sqlalchemy import text

@app.get("/records/{record_id}")
def read_record(record_id: str, db: Session = Depends(get_db), current_user: dict = Depends(get_current_user)):
    sql = text("SELECT id, data, tenant_id FROM records WHERE id = :rid AND tenant_id = :tid")
    result = db.execute(sql, {"rid": record_id, "tid": current_user["tenant_id"]})
    row = result.fetchone()
    if row is None:
        raise HTTPException(status_code=404, detail="Not found or access denied")
    return {"id": row.id, "data": row.data}

4. Avoiding ID enumeration by using opaque references

Where possible, use UUIDs or hash-based identifiers that do not leak enumeration information, and still enforce tenant and ownership checks.

import uuid
from sqlalchemy import Column, String

class Document(Base):
    __tablename__ = "documents"
    id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
    tenant_id = Column(Integer)
    owner_id = Column(Integer)

@app.get("/documents/{doc_id}")
def get_document(doc_id: str, db: Session = Depends(get_db), current_user: dict = Depends(get_current_user)):
    stmt = select(Document).filter(Document.id == doc_id, Document.owner_id == current_user["user_id"])
    result = db.execute(stmt)
    doc = result.scalar_one_or_none()
    if doc is None:
        raise HTTPException(status_code=404, detail="Not found or access denied")
    return {"id": doc.id, "tenant_id": doc.tenant_id}

5. Middleware to validate tenant context early

Use middleware to ensure tenant context is available for downstream dependencies, reducing the chance of omitting scoping in individual routes.

from fastapi import Request

@app.middleware("http")
async def tenant_middleware(request: Request, call_next):
    # Example: extract tenant from header and attach to state or context
    tenant_id = request.headers.get("X-Tenant-ID")
    if request.url.path.startswith("/api") and not tenant_id:
        raise HTTPException(status_code=400, detail="Missing tenant identifier")
    response = await call_next(request)
    return response

These patterns ensure that object-level authorization is enforced before queries reach CockroachDB. By combining authenticated subject checks, tenant scoping, and parameterized SQL, you reduce the attack surface for BOLA and related IDOR issues in FastAPI services backed by CockroachDB.