Flask-WTF Passes Validation when it Should Fail

Question

I have a Flask App that passes user input validation when it should fail. I have similar code in another part of the app that works just fine. It seems like the FileAllowed() method is not being called. Or if it is, it's returning true.

This code uploads a user file to s3.

The MultipleFileField() method has a validation check for only image file extensions. However, any file passes this check. The InputRequired() method works just fine.

I've tried multiple variations of this and nothing has worked. It's not a CRSF issue because other routes with similar code work without it.

flask_wtf Form:

    class AddImgForm(FlaskForm): # should use InputRequired() not DataRequired()
        images= MultipleFileField('Upload Images', validators=[InputRequired(),FileAllowed(['jpg', 'png', 'jpeg', 'tif'])])

        submitBTN2 = SubmitField('Upload')

Route:

@users.route("/account", methods=['GET', 'POST'])
@login_required
def account():
    form = UpdateAccountForm()
    if form.validate_on_submit():

        if form.picture.data: # if a picture is provided save picture

            picture_file= save_picture(form.picture.data, 'p') # saves picture and returns dict with ['filepath'] and ['filename']

            BUCKET= os.environ['BUCKET'] # should send to 'bucket-publicaccess/uploads' bucket in production

            s3= boto3.resource("s3", 
                        region_name = "us-east-2", # had to add "us-east-2" as incorrect region was generated
                        config= boto3.session.Config(signature_version='s3v4'), # must add this to address newer security
                        aws_access_key_id = os.environ["AWS_ACCESS_KEY_ID"],
                        aws_secret_access_key = os.environ["AWS_SECRET_ACCESS_KEY"]) # AWS Generated key pairs

            s3.Bucket(BUCKET).upload_file(picture_file['filepath'], 'uploads/'+ picture_file['filename']) #upload to s3

            current_user.image_file= 'uploads/'+picture_file['filename']
            print(current_user.image_file)
            os.remove(picture_file['filepath']) # remove file from tmp directory

        current_user.username = form.username.data 
        current_user.email = form.email.data
        db.session.commit() # commit changes

        flash('Your account has been updated!', 'success')
        return redirect(url_for('users.account'))
    elif request.method == 'GET':
        form.username.data = current_user.username
        form.email.data = current_user.email
    image_file = current_user.image_file

    return render_template('account.html', title='Account',
                           image_file=image_file, form=form)

HTML:

      <form method="POST" action="" enctype="multipart/form-data" id="addImgForm">
        {{ addImgForm.hidden_tag() }}
        <fieldset class="form-group">
          <div class="form-group">
            {{ addImgForm.images.label() }}
            {{ addImgForm.images(class="form-control-file") }}
            {% if addImgForm.images.errors %}
              {% for error in addImgForm.images.errors %}
                <span class="text-danger">{{ error }}</span></br>
              {% endfor %}
            {% endif %}
          </div>
          <div class="form-group">
            <div class="modal-footer">
              <button type="button" class="btn btn-secondary" data-dismiss="modal">Close</button>
              {{ addImgForm.submitBTN2(class="btn btn-outline-info") }}
            </div>
        </fieldset>
      </form>

Any help would be appreciated as most questions are about this failing while this code always passes.

Indoaryan

Shouldn't the InputRequired() be FileRequired() ? – Comic

I don't think so because MultiFileField is a class from wtforms and so is InputRequired(). I think the issue is around combining a wtforms class and flask_wtf class. WTForms does not have a built in FileAllowed validator that I can find. So I think it just skips the method. – Indoaryan

Answer 1

The problem is with the FileAllowed validator, it is expecting a single instance of FileStorage when validating, whereas MultipleFileField passes a list of FileStorage instances to the validator. You can overcome this by writing your own validator, for example:

class MultiFileAllowed(object):

    def __init__(self, upload_set, message=None):
        self.upload_set = upload_set
        self.message = message

    def __call__(self, form, field):

        # FileAllowed only expects a single instance of FileStorage
        # if not (isinstance(field.data, FileStorage) and field.data):
        #     return

        # Check that all the items in field.data are FileStorage items
        if not (all(isinstance(item, FileStorage) for item in field.data) and field.data):
            return

        for data in field.data:
            filename = data.filename.lower()

            if isinstance(self.upload_set, Iterable):
                if any(filename.endswith('.' + x) for x in self.upload_set):
                    return

                raise StopValidation(self.message or field.gettext(
                    'File does not have an approved extension: {extensions}'
                ).format(extensions=', '.join(self.upload_set)))

            if not self.upload_set.file_allowed(field.data, filename):
                raise StopValidation(self.message or field.gettext(
                    'File does not have an approved extension.'
                ))

A simple single file example using Flask, Flask-WTF and Flask-Boostrap:

from collections import Iterable

from flask_bootstrap import Bootstrap
from flask import Flask, redirect, url_for, render_template_string
from flask_wtf import FlaskForm
from flask_wtf.file import FileAllowed
from markupsafe import Markup
from werkzeug.datastructures import FileStorage
from wtforms.fields import MultipleFileField, SubmitField
from wtforms.validators import InputRequired, StopValidation

app = Flask(__name__)
app.config['SECRET_KEY'] = '123456790'
Bootstrap(app)


class MultiFileAllowed(object):

    def __init__(self, upload_set, message=None):
        self.upload_set = upload_set
        self.message = message

    def __call__(self, form, field):

        if not (all(isinstance(item, FileStorage) for item in field.data) and field.data):
            return

        for data in field.data:
            filename = data.filename.lower()

            if isinstance(self.upload_set, Iterable):
                if any(filename.endswith('.' + x) for x in self.upload_set):
                    return

                raise StopValidation(self.message or field.gettext(
                    'File does not have an approved extension: {extensions}'
                ).format(extensions=', '.join(self.upload_set)))

            if not self.upload_set.file_allowed(field.data, filename):
                raise StopValidation(self.message or field.gettext(
                    'File does not have an approved extension.'
                ))


class ImagesForm(FlaskForm):
    images = MultipleFileField(
        'Upload Images',
        validators=[
            InputRequired(),
            MultiFileAllowed(['jpg', 'png', 'jpeg', 'tif'])
        ]
    )
    submit = SubmitField('Upload')


upload_template = '''
{% import "bootstrap/wtf.html" as wtf %}
<form method="POST" enctype="multipart/form-data">
    {{ wtf.quick_form(form) }}
</form>
'''


@app.route('/')
def index():
    return Markup("<a href='uploads'>Go to the uploads<a>")


@app.route('/uploads', methods=['GET', 'POST'])
def upload():
    form = ImagesForm()
    if form.validate_on_submit():
        if form.images:
            for image in form.images.data:
                print 'Uploaded File: {}'.format(image.filename)

        return redirect(url_for('index'))
    else:
        print form.errors

    return render_template_string(upload_template, form=form)


if __name__ == '__main__':
    app.run()

Comic

Answer 2

Thanks for the above! I had to change to

class MultiFileAllowed(object):
    def __init__(self, upload_set, message=None):
        self.upload_set = upload_set
        self.message = message

    def __call__(self, form, field):

        if not (all(isinstance(item, FileStorage) for item in field.data) and field.data):
            return

        for data in field.data:
            filename = data.filename.lower()

            if isinstance(self.upload_set, Iterable):
                print(filename, flush=True)
                print(any(filename.endswith("." + x) for x in self.upload_set), flush=True)
                if not any(filename.endswith("." + x) for x in self.upload_set):
                    raise StopValidation(
                        self.message
                        or field.gettext("File does not have an approved extension: {extensions}").format(
                            extensions=", ".join(self.upload_set)
                        )
                    )

Abdul