Mass Assignment in Flask

Mass assignment vulnerabilities in Flask applications typically arise when request data is automatically mapped to model objects without proper field filtering. Flask's flexibility with request handling creates several attack vectors that developers must understand.

The most common pattern involves using Flask's request.json or request.form directly with SQLAlchemy models. Consider this vulnerable endpoint:

from flask import Flask, request, jsonify
from models import User

app = Flask(__name__)

@app.route('/users', methods=['POST'])
def create_user():
    user_data = request.json
    user = User(**user_data)  # Vulnerable: direct unpacking
    db.session.add(user)
    db.session.commit()
    return jsonify(user.to_dict()), 201

An attacker can exploit this by sending additional fields that map to sensitive model attributes:

{
  "username": "attacker",
  "password": "password123",
  "is_admin": true,
  "account_balance": 10000
}

Flask-SQLAlchemy's dynamic model creation means any matching field name gets populated, potentially escalating privileges or modifying protected data.

Another Flask-specific manifestation occurs with Marshmallow schemas used for deserialization:

from flask import Blueprint, request
from schemas import UserSchema
from models import User

user_bp = Blueprint('user', __name__)

@user_bp.route('/users', methods=['POST'])
def create():
    schema = UserSchema()
    user_data = schema.load(request.json)  # Vulnerable if schema is too permissive
    user = User(**user_data)
    return schema.jsonify(user), 201

The vulnerability here depends on the schema definition. If the schema allows unknown fields or doesn't explicitly exclude sensitive fields, mass assignment succeeds.

Flask-RESTful resources present another attack surface:

from flask_restful import Resource
from models import User

class UserResource(Resource):
    def post(self):
        parser = reqparse.RequestParser()
        parser.add_argument('username', type=str)
        parser.add_argument('password', type=str)
        parser.add_argument('is_admin', type=bool)  # Should be excluded
        
        args = parser.parse_args()
        user = User(**args)  # Vulnerable if admin field is included
        db.session.add(user)
        db.session.commit()
        return {'id': user.id}, 201

The Flask-specific pattern here is the use of RequestParser, which can inadvertently include sensitive fields if not carefully configured.

Detecting mass assignment vulnerabilities in Flask requires both static analysis and runtime scanning. middleBrick's Flask-specific detection examines your application's request handling patterns and model definitions.

middleBrick scans for these Flask-specific indicators:

• Direct request.json unpacking into model constructors
• Marshmallow schemas with unknown=INCLUDE or missing exclude parameters
• Flask-RESTful RequestParser configurations that include sensitive fields
• Dynamic model field assignment patterns

Run middleBrick against your Flask API with:

middlebrick scan https://yourapi.com --output json

For local development, use the CLI to scan your running Flask app:

middlebrick scan http://localhost:5000 --output html

middleBrick's scanner specifically looks for Flask's request handling patterns and identifies where request data flows directly into model constructors without validation. The scanner checks for:

• Unfiltered request.json or request.form usage
• Marshmallow schemas that don't use exclude or only parameters
• Flask-RESTful resources with permissive RequestParser configurations

The scanner also tests for actual exploitation by attempting to set sensitive fields through API endpoints, providing concrete evidence of vulnerabilities.

Securing Flask applications against mass assignment requires explicit field filtering and proper data validation. Here are Flask-specific remediation patterns:

Explicit Field Filtering:

from flask import request
from models import User

ALLOWED_FIELDS = {'username', 'email', 'first_name', 'last_name'}

@app.route('/users', methods=['POST'])
def create_user():
    user_data = {k: v for k, v in request.json.items() 
                 if k in ALLOWED_FIELDS}
    user = User(**user_data)
    db.session.add(user)
    db.session.commit()
    return jsonify(user.to_dict()), 201

Marshmallow Schema Whitelisting:

from marshmallow import Schema, fields

class UserSchema(Schema):
    username = fields.Str(required=True)
    email = fields.Email(required=True)
    first_name = fields.Str()
    last_name = fields.Str()
    
    class Meta:
        unknown = EXCLUDE  # Block unknown fields
        exclude = ('is_admin', 'account_balance', 'password_hash')

@app.route('/users', methods=['POST'])
def create_user():
    schema = UserSchema()
    user_data = schema.load(request.json)
    user = User(**user_data)
    db.session.add(user)
    db.session.commit()
    return schema.jsonify(user), 201

Flask-RESTful with Field Restrictions:

from flask_restful import Resource, reqparse
from models import User

class UserResource(Resource):
    def __init__(self):
        self.parser = reqparse.RequestParser()
        self.parser.add_argument('username', type=str, required=True)
        self.parser.add_argument('email', type=str, required=True)
        self.parser.add_argument('first_name', type=str)
        self.parser.add_argument('last_name', type=str)
        # Explicitly exclude sensitive fields
        
    def post(self):
        args = self.parser.parse_args()
        user = User(**args)
        db.session.add(user)
        db.session.commit()
        return {'id': user.id}, 201

Model-Level Protection:

from sqlalchemy.ext.declarative import DeclarativeMeta

class BaseModel:
    @classmethod
    def create_from_dict(cls, data):
        # Only allow fields defined in the model
        allowed = {k: v for k, v in data.items() 
                   if k in cls.__table__.columns.keys()}
        return cls(**allowed)

class User(BaseModel, db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)
    is_admin = db.Column(db.Boolean, default=False)
    
@app.route('/users', methods=['POST'])
def create_user():
    user = User.create_from_dict(request.json)
    db.session.add(user)
    db.session.commit()
    return jsonify(user.to_dict()), 201