Integrity Failures in Fastapi

Integrity failures in Fastapi applications occur when attackers manipulate data to bypass authorization controls, modify records they shouldn't access, or corrupt data integrity. Fastapi's modern async design and Pydantic models create specific vulnerability patterns that differ from traditional Flask applications.

The most common Fastapi integrity failure involves improper Pydantic model validation. Consider a user profile update endpoint:

@app.put("/users/{user_id}")
async def update_user(user_id: int, user_data: UserUpdate): 
    user = await get_user(user_id)
    user.update(user_data.dict())
    await save_user(user)
    return user

This appears secure, but if UserUpdate includes fields like is_admin or account_balance, an attacker can escalate privileges or modify data they shouldn't access. Fastapi's automatic Pydantic validation doesn't prevent this—it only validates data types, not authorization.

Another Fastapi-specific pattern involves dependency injection misuse. Fastapi's dependency system can create integrity bypass paths:

async def get_current_user(token: str = Depends(oauth2_scheme)):
    return await authenticate_user(token)

If this dependency isn't properly scoped or if multiple dependencies create conflicting user contexts, integrity checks can be bypassed. Fastapi's async nature means race conditions can occur where user state changes between dependency resolution and endpoint execution.

Fastapi's OpenAPI generation also creates integrity risks. The automatic schema documentation can expose internal field names and data structures that attackers use to craft malicious payloads. A typical Fastapi application might expose:

{
  "required": ["id", "user_id", "amount", "status"],
  "properties": {
    "id": {"type": "integer"},
    "user_id": {"type": "integer"},
    "amount": {"type": "number"},
    "status": {"type": "string", "enum": ["pending", "completed", "failed"]}
  }
}

An attacker studying this schema knows exactly which fields to manipulate for integrity attacks.

Detecting integrity failures in Fastapi requires understanding its async architecture and Pydantic integration. middleBrick's black-box scanning approach is particularly effective for Fastapi applications because it tests the actual runtime behavior without needing source code access.

middleBrick scans Fastapi endpoints by sending crafted payloads that test authorization boundaries. For the user update example above, middleBrick would attempt:

POST /users/123 HTTP/1.1
Content-Type: application/json

{
  "username": "attacker",
  "is_admin": true,
  "account_balance": 999999.99
}

The scanner checks if these unauthorized fields are accepted and processed, which would indicate an integrity failure. Fastapi's Pydantic model binding means these fields might be silently accepted if not explicitly excluded.

middleBrick's OpenAPI analysis is particularly valuable for Fastapi applications. Since Fastapi auto-generates OpenAPI specs from Pydantic models, the scanner can:

• Extract all model fields and their types from the spec
• Identify fields that shouldn't be user-modifiable (IDs, status fields, permissions)
• Cross-reference these with the actual endpoint behavior
• Detect discrepancies between declared models and runtime behavior

For Fastapi's dependency injection system, middleBrick tests for context manipulation by:

• Calling endpoints with different authentication states
• Measuring response times to detect async race conditions
• Testing if user context changes mid-request affect data access

The scanner also tests Fastapi's async database operations for integrity issues. Since Fastapi commonly uses async databases like PostgreSQL with asyncpg or MongoDB, middleBrick sends concurrent requests to detect race conditions where data integrity can be compromised.

Fastapi provides several native mechanisms to prevent integrity failures. The most effective approach uses Pydantic's exclude and include parameters combined with Fastapi's dependency injection for proper authorization.

First, create separate Pydantic models for different operations:

class UserRead(PydanticBase):
    id: int
    username: str
    email: str
    created_at: datetime

class UserUpdate(PydanticBase):
    username: Optional[str] = None
    email: Optional[str] = None
    bio: Optional[str] = None

class UserAdminUpdate(PydanticBase):
    username: Optional[str] = None
    email: Optional[str] = None
    is_admin: Optional[bool] = None
    account_balance: Optional[float] = None

Then use these models with proper authorization checks:

@app.put("/users/{user_id}")
async def update_user(
    user_id: int,
    user_data: UserUpdate,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db)
):
    target_user = await get_user_by_id(db, user_id)
    
    if current_user.id != target_user.id and not current_user.is_admin:
        raise HTTPException(status_code=403, detail="Insufficient permissions")
    
    target_user.username = user_data.username or target_user.username
    target_user.email = user_data.email or target_user.email
    target_user.bio = user_data.bio or target_user.bio
    
    await db.commit()
    return target_user

For admin operations, create a separate endpoint with stricter controls:

@app.put("/admin/users/{user_id}")
async def admin_update_user(
    user_id: int,
    admin_data: UserAdminUpdate,
    current_user: User = Depends(get_current_admin_user)
):
    if not current_user.is_admin:
        raise HTTPException(status_code=403, detail="Admin privileges required")
    
    target_user = await get_user_by_id(user_id)
    target_user.is_admin = admin_data.is_admin or target_user.is_admin
    target_user.account_balance = admin_data.account_balance or target_user.account_balance
    
    await save_user(target_user)
    return target_user

Fastapi's Depends system can also enforce integrity through context validation:

async def validate_user_context(user_id: int = Path(),
                              current_user: User = Depends(get_current_user)):
    if current_user.id != user_id and not current_user.is_admin:
        raise HTTPException(status_code=403, detail="Context mismatch")
    return current_user

This dependency can be added to any endpoint that requires context validation, ensuring the authenticated user matches the resource being accessed.