Flask-WTF 在它应该失败时通过验证

Question

我有一个 Flask 应用程序，当它应该失败时通过用户输入验证。我在应用程序的另一部分有类似的代码，效果很好。似乎没有调用 FileAllowed() 方法。或者，如果是，它会返回 true。

此代码将用户文件上传到 s3。

MultipleFileField() 方法仅对图像文件扩展名进行验证检查。但是，任何文件都会通过此检查。 InputRequired() 方法工作得很好。

我已经尝试了多种变体，但没有任何效果。这不是 CRSF 问题，因为其他具有类似代码的路由没有它也可以工作。

flask_wtf 形式：

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

        submitBTN2 = SubmitField('Upload')

路线：

@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>

任何帮助将不胜感激，因为大多数问题都与此失败有关，而此代码始终通过。

Answer 1

问题出在FileAllowed 验证器上，它在验证时期望FileStorage 的单个实例，而MultipleFileField 将FileStorage 实例列表传递给验证器。您可以通过编写自己的验证器来克服这个问题，例如：

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.'
                ))

使用 Flask、Flask-WTF 和 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()

Answer 2

感谢以上！我不得不改成

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)
                        )
                    )