前回の続き

Flask で EC サイトの機能を作ってみる①の続きで、ここではフォームと CSRF について考えてみる。

Flask でアプリケーションを作るときに POST を行うような処理が必要な場合は普通、form を html に設けることになる。 これはこれから話すフォーム(python 側でフォームを組み立てて html に表示する)を必ずしも必要とはしませんが、バリデーションなどの機能がついてくるので特に理由がないなら採用して良いと思います。

フォームは Flask WTForm と呼ばれるライブラリで構築します。

Flask WTForm でフォームを構築する

まず requirements.txt に Flask-WTF を追加する。 あと、メールのバリデーションもしたいので email-validator も入れておく。

Flask-WTF>=1.2,<2.0
email-validator>=2.1,<3.0

前回の記事で作成した main.py で FlaskForm 等を import してサインアップフォームを作成してみると以下のような感じになる。 ここでは認証については考えないので、単純なユーザーデータ作成を行うサインアップとして考える。

from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField
from wtforms.validators import InputRequired, Length, Email
from flask_wtf.csrf import CSRFProtect

app = Flask(__name__)

# 省略

csrf = CSRFProtect()
csrf.init_app(app) # CSRF 機能を設定

class SignUpForm(FlaskForm):
    name = StringField('name', validators=[InputRequired(), Length(min=1, max=20)])
    email = StringField('email', validators=[InputRequired(), Email()])

そういえば、前回の記事で作成した users テーブルには email カラムを設けていなかったので、User クラスを修正して再度マイグレーションをかけておく。 email カラムはユニークにして null が入らないようにした。 name もユニークにしている。

class User(db.Model):
    __tablename__ = 'users'
    id: Mapped[int] = mapped_column(db.Integer, primary_key=True)
    name: Mapped[str] = mapped_column(db.String(20), unique=True, nullable=False)
    email: Mapped[str] = mapped_column(db.String(100), unique=True, nullable=False)
    purchase_transactions: Mapped[List['PurchaseTransaction']] = db.relationship(back_populates='user', cascade='all, delete-orphan')
    created: Mapped[datetime] = mapped_column(db.DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
    updated: Mapped[datetime] = mapped_column(db.DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
    def __repr__(self) -> str:
        return f'User(id={self.id!r}, name={self.name!r}, created={self.created!r}, updated={self.updated!r})'

そしたらコンテナに入るなりして flask db migrate と flask db upgrade を実行する。

root@d4d07ffc9c48:/app# flask db migrate -m 'change user'
INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
INFO  [alembic.ddl.postgresql] Detected sequence named 'purchase_transactions_id_seq' as owned by integer column 'purchase_transactions(id)', assuming SERIAL and omitting
INFO  [alembic.autogenerate.compare] Detected added column 'users.email'
INFO  [alembic.autogenerate.compare] Detected added unique constraint None on '('email',)'
INFO  [alembic.autogenerate.compare] Detected added unique constraint None on '('name',)'
  Generating /app/migrations/versions/1070adbc40d0_change_user.py ...  done
root@13582844164c:/app# flask db upgrade

テーブルも作れたので、フォームを html に組み込んでみよう。 sign_up.html を作成する。 sign_up.html ではフォームの原型を定義する _formhelpers.html から render_field をインポートしている。 あとはこの render_field を form 内に配置してあげるだけで良い。

<!-- sign_up.html -->
{% extends 'base.html' %}
{% from "_formhelpers.html" import render_field %}

{% block header %}
  <h1>{% block title %}SignUp{% endblock %}</h1>
{% endblock %}

{% block content %}
  <section>
    <form method="post">
      {{ form.csrf_token }}
      <dl>
        {{ render_field(form.name, placeholder="name") }}
        {{ render_field(form.email, placeholder="email") }}
      </dl>
      <input type="submit" value="サインアップ">
    </form>
  </section>
{% endblock %}

こっちは _formhelpers.html

<!-- _formhelpers.html -->
{% macro render_field(field) %}
  <dt>{{ field.label }}
  <dd>{{ field(**kwargs)|safe }}
  {% if field.errors %}
    <ul class=errors>
    {% for error in field.errors %}
      <li>{{ error }}</li>
    {% endfor %}
    </ul>
  {% endif %}
  </dd>
{% endmacro %}

そもそも sign_up() API を作ってなかったので追加しておくのも忘れずに。 こいつは GET と POST の両方で使うので methods で2つを許容しておく。

上で触れた SignUpForm に request.form を突っ込み、そのバリデーションを行った上で問題なければ、User を作成し、作成後は商品一覧ページ(/products)へリダイレクトしている。 バリデーションに問題があれば sign_up.html に戻ってくるようになっている。

@app.route("/sign_up", methods=('GET', 'POST'))
def sign_up():
    form = SignUpForm(request.form)
    if form.validate_on_submit():
        name = form.name.data
        email = form.email.data
        user = User(name=name, email=email)
        db.session.add(user)
        db.session.commit()
        return redirect(url_for('products'))
    return render_template('sign_up.html', form=form)

たったこれだけなのだが、以上でバリデーションと CSRF 対策ができた。 試しにメールアドレスの項目に出鱈目なフォーマットで入力してみると、Invalid email address. と表示されるだろう。

CSRF については、main.py に記載していた secret_key を基に生成されるトークンが form に埋め込まれるので、これが POST 時に検証され、検証に成功すると sign_up() が正常に実行されるようになる。

CSRF トークンは同一オリジンなら cookie の設定や Origin ヘッダーの検証などで置き換えることもできるかと思うが、特に実装に苦労することでもないので黙ってやれば良いと思う。 脆弱性とかっていつどんな形で見つかるかわからないので、CSRF トークンにしか防げないような問題が見つかる可能性もないわけではないと思う（蓋然性はどうあれ）。

cookie と Origin ヘッダーの検証については次回の認証の記事で触れようと思う。 まだそもそも認証の機能がなく CSRF は起こり得ないため、ちょっとここでは説明しづらいというのもある。

を仕舞い

フォームについては以上、次は認証について書こうと思う。