Integrity Failures in Fastapi with Cockroachdb

Integrity failures occur when an API allows unauthorized or invalid changes to data, violating correctness and trust constraints. The combination of FastAPI and CockroachDB can expose integrity risks when application-layer validation is insufficient and database-side constraints are not enforced consistently. CockroachDB provides strong ACID guarantees and serializable isolation, but these guarantees only protect integrity when the application uses them correctly.

One common pattern is building endpoints that perform read-modify-write cycles without explicit transaction boundaries or optimistic concurrency control. For example, consider an endpoint that reads a bank balance, computes a new balance, and writes it back. If two requests execute concurrently, the second write can overwrite the first, resulting in a lost update. This is a BOLA/IDOR-style integrity failure: the API fails to ensure that state transitions are applied to the intended, current state. CockroachDB’s serializable isolation will raise a transaction retry error when write skew occurs, but if the FastAPI route does not implement retry logic, the client receives a 500 error and the operation is lost, revealing an integrity weakness in the application layer.

Another scenario involves missing or inconsistent checks on identifiers and ownership. Suppose an endpoint like /accounts/{account_id}/transfer uses a path parameter account_id to locate a resource but does not verify that the authenticated subject has permission or that the provided account_id matches the user’s allowed set. Because CockroachDB enforces foreign-key and uniqueness constraints, omitting these checks in the application can lead to privilege escalation or illegal state changes that violate invariants. For example, if a uniqueness constraint on a natural key is not enforced at the schema level or validated before insert, clients might create duplicate records that corrupt data integrity.

Schema evolution and type mismatches between FastAPI models and CockroachDB columns can also introduce integrity failures. If a Pydantic model omits a non-nullable field or applies incorrect coercion, the application may send invalid or partial data to the database. CockroachDB will reject the write with a constraint violation, but without proper error handling in FastAPI, this surfaces as a generic server error rather than a clear validation failure. This makes it harder to distinguish integrity violations from transient faults and can mask the root cause. Additionally, using unvalidated inputs in dynamic SQL or ORM queries may enable logical corruption, such as updating the wrong row due to concatenated identifiers or missing filters.

These issues are detectable by the 12 security checks in middleBrick, including BOLA/IDOR, Property Authorization, and Input Validation. In a black-box scan, middleBrick submits unauthenticated requests to the FastAPI endpoints, observes responses, and cross-references findings with OpenAPI/Swagger specs and CockroachDB schema definitions (including $ref resolution). The scanner can identify missing ownership checks, lack of transaction usage for multi-step updates, and absence of schema-level constraints that should be enforced by the database. Findings include severity, remediation guidance, and mappings to frameworks such as OWASP API Top 10 and PCI-DSS, helping teams prioritize integrity-related fixes.

Remediation centers on using explicit transactions with retry logic, enforcing ownership and constraints in both the application and the database, and validating inputs before they reach CockroachDB. Below are concrete, realistic FastAPI code examples that demonstrate these practices.

1) Transactions with retry for integrity-sensitive updates

Use CockroachDB’s transactional guarantees to ensure read-modify-write cycles are atomic. Implement retries on serialization failures, which are expected under serializable isolation.

from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker, Session
import tenacity

DATABASE_URL = "cockroachdb://root@localhost:26257/defaultdb?sslmode=require"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

app = FastAPI()

@tenacity.retry(
    stop=tenacity.stop_after_attempt(3),
    wait=tenacity.wait_exponential(multiplier=0.5, min=1, max=10),
    retry=tenacity.retry_if_exception_type(InvalidRequestError),
)
def update_balance(session: Session, account_id: int, delta: int) -> None:
    result = session.execute(text(
        "SELECT balance FROM accounts WHERE id = :id FOR UPDATE"), {"id": account_id}
    )
    row = result.fetchone()
    if row is None:
        raise HTTPException(status_code=404, detail="Account not found")
    new_balance = row[0] + delta
    if new_balance < 0:
        raise HTTPException(status_code=400, detail="Insufficient funds")
    session.execute(text(
        "UPDATE accounts SET balance = :bal WHERE id = :id"),
        {"bal": new_balance, "id": account_id}
    )
    session.commit()

@app.post("/accounts/{account_id}/deposit")
def deposit(account_id: int, amount: int, session: Session = Depends(lambda: SessionLocal())):
    try:
        update_balance(session, account_id, amount)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

2) Enforce ownership and constraints in queries

Always include tenant or user context in WHERE clauses and rely on database constraints to reject invalid states.

from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class UserAccount(Base):
    __tablename__ = "user_accounts"
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, nullable=False)
    account_id = Column(Integer, nullable=False)
    balance = Column(Integer, default=0)
    __table_args__ = (
        UniqueConstraint("user_id", "account_id", name="uq_user_account"),
    )

# In a route, include user_id in the filter
def get_user_account(session: Session, user_id: int, account_id: int) -> Optional[UserAccount]:
    return session.query(UserAccount).filter(
        UserAccount.user_id == user_id,
        UserAccount.account_id == account_id
    ).first()

3) Input validation and schema-level constraints

Define non-nullable columns and unique constraints in CockroachDB, and validate payloads in FastAPI with Pydantic before sending to the database.

from pydantic import BaseModel, Field

class TransferRequest(BaseModel):
    from_account_id: int = Field(..., gt=0)
    to_account_id: int = Field(..., gt=0)
    amount: int = Field(..., gt=0)

@app.post("/transfer")
def transfer(req: TransferRequest, session: Session = Depends(lambda: SessionLocal())):
    # Ownership and existence checks
    src = get_user_account(session, current_user.id, req.from_account_id)
    dst = get_user_account(session, current_user.id, req.to_account_id)
    if not src or not dst:
        raise HTTPException(status_code=403, detail="Forbidden")
    try:
        session.execute(text(
            "UPDATE accounts SET balance = balance - :amt WHERE id = :id"), {"amt": req.amount, "id": req.from_account_id}
        )
        session.execute(text(
            "UPDATE accounts SET balance = balance + :amt WHERE id = :id"), {"amt": req.amount, "id": req.to_account_id}
        )
        session.commit()
    except Exception as e:
        session.rollback()
        raise HTTPException(status_code=500, detail=str(e))

By combining explicit transactions, ownership checks, and schema constraints, the FastAPI + CockroachDB stack preserves data integrity. middleBrick can validate these patterns in scans, highlighting missing retries, absent ownership filters, and schema definitions that do not enforce critical constraints.