Mass Assignment in Fastapi

Mass assignment vulnerabilities in Fastapi occur when user-controlled data is automatically mapped to model attributes without proper filtering. Fastapi's Pydantic models and dependency injection system create several attack vectors that developers must understand.

The most common scenario involves Pydantic models with orm_mode=True that expose all model fields to client input. Consider this vulnerable Fastapi endpoint:

from fastapi import FastAPI
from pydantic import BaseModel
from typing import Optional

app = FastAPI()

class User(BaseModel):
    id: int
    name: str
    email: str
    is_admin: bool = False
    last_login: Optional[datetime] = None

@app.put('/users/{user_id}')
async def update_user(user_id: int, user_data: User):
    # Vulnerable: accepts all fields including is_admin
    user = await get_user_from_db(user_id)
    user.__dict__.update(user_data.__dict__)
    await save_user(user)
    return user

An attacker can escalate privileges by sending:

{
  "id": 123,
  "name": "Evil Hacker",
  "email": "[email protected]",
  "is_admin": true,
  "last_login": "2024-01-01T00:00:00"
}

Fastapi's dependency injection system can also introduce mass assignment through automatic model binding. When using Depends() with Pydantic models, all fields become writable unless explicitly restricted:

from fastapi import Depends

async def get_user_data() -> User:
    return User.parse_obj(await request.json())

@app.put('/users/{user_id}')
async def update_user(user_id: int, user_data: User = Depends(get_user_data)):
    # All fields from request are bound to user_data
    user = await get_user_from_db(user_id)
    user.__dict__.update(user_data.__dict__)
    await save_user(user)
    return user

Fastapi's automatic serialization/deserialization with Pydantic models means that any field defined in the model becomes a potential attack vector. This is particularly dangerous when models include sensitive fields like is_active, role, permissions, or database-specific fields like created_at and updated_at that should never be user-modifiable.

Detecting mass assignment vulnerabilities in Fastapi requires both static analysis and runtime scanning. middleBrick's Fastapi-specific scanner examines Pydantic models, endpoint definitions, and request handling patterns to identify potential mass assignment issues.

The scanner analyzes your OpenAPI specification to identify endpoints that accept model objects with writeable fields. It specifically looks for:

• PUT and PATCH endpoints that accept full model objects
• Pydantic models with orm_mode=True that expose ORM model fields
• Endpoints using Depends() with Pydantic models
• Database model fields that should be read-only

Here's how you can scan your Fastapi application with middleBrick:

# Install middleBrick CLI
npm install -g middlebrick

# Scan your Fastapi API
middlebrick scan https://your-fastapi-app.com/openapi.json

# Or scan from your Fastapi app
middlebrick scan http://localhost:8000/docs

middleBrick's Fastapi scanner specifically checks for these vulnerable patterns:

# Vulnerable pattern detected
class UserModel(BaseModel):
    id: int  # Should be read-only
    username: str
    password: str  # Should never be settable via update
    is_admin: bool = False  # Privilege escalation risk
    created_at: datetime  # Should be immutable

@app.put('/users/{user_id}')
async def update_user(user_id: int, user: UserModel):
    # Scanner flags this as high risk
    user = await get_user(user_id)
    user.__dict__.update(user.__dict__)  # Mass assignment
    await save_user(user)
    return user

The scanner also examines your Fastapi application's dependency injection patterns, looking for Depends() usage that might automatically bind request data to model objects without field filtering.

Fastapi provides several native mechanisms to prevent mass assignment vulnerabilities. The most effective approach is using Pydantic's exclude and include parameters to control field exposure.

Here's a secure implementation using field exclusion:

from pydantic import BaseModel, EmailStr
from typing import Optional

class UserUpdate(BaseModel):
    name: Optional[str] = None
    email: Optional[EmailStr] = None
    phone: Optional[str] = None

class UserRead(BaseModel):
    id: int
    name: str
    email: EmailStr
    is_admin: bool
    created_at: datetime

@app.put('/users/{user_id}')
async def update_user(user_id: int, update_data: UserUpdate):
    user = await get_user_from_db(user_id)
    
    # Only update allowed fields
    if update_data.name:
        user.name = update_data.name
    if update_data.email:
        user.email = update_data.email
    if update_data.phone:
        user.phone = update_data.phone
    
    await save_user(user)
    return UserRead.from_orm(user)

Another Fastapi-specific approach uses Pydantic's Config with schema_extra to document field restrictions:

class UserUpdate(BaseModel):
    name: Optional[str] = None
    email: Optional[EmailStr] = None
    
    class Config:
        schema_extra = {
            'example': {
                'name': 'John Doe',
                'email': '[email protected]'
            }
        }

@app.put('/users/{user_id}')
async def update_user(user_id: int, update_data: UserUpdate):
    user = await get_user_from_db(user_id)
    
    # Fastapi's Pydantic will only accept the defined fields
    # Any extra fields in the request will cause a 422 error
    update_dict = update_data.dict(exclude_unset=True)
    
    for key, value in update_dict.items():
        setattr(user, key, value)
    
    await save_user(user)
    return UserRead.from_orm(user)

For database models, Fastapi integrates well with SQLAlchemy's hybrid approach. Use separate Pydantic models for input and output:

# Input model - only fields users can set
class UserCreate(BaseModel):
    username: str
    email: EmailStr
    password: str

# Output model - all fields users can see
class UserResponse(BaseModel):
    id: int
    username: str
    email: EmailStr
    is_admin: bool
    created_at: datetime
    
    class Config:
        from_attributes = True

@app.post('/users/')
async def create_user(user_data: UserCreate):
    # Only the fields in UserCreate are accepted
    user = User(
        username=user_data.username,
        email=user_data.email,
        password=hash_password(user_data.password)
    )
    await save_user(user)
    return UserResponse.from_orm(user)