🐍

Flask-AdminとFlask-LoginによるDB管理者画面の実装

2020/10/13に公開1

はじめに

Flaskを使ったDjangoの管理者画面のようなものを実装するには、Flask-Adminというライブラリを使うと便利です。
しかしFlask-Adminをそのまま使うだけでは、パスワードを打つことなく(ログインすることなく)管理者画面に入ることができてしまい、セキュリティ上とても脆弱です。
この記事では、Flask-AdminとFlask-Loginを使用してログイン機能のついたDB管理者画面の実装を行っていきます。

この記事は

Qiita版

Qiitaでも全く同じ内容の記事かいてます。筆者は同じです。
https://qiita.com/FPC_COMMUNITY/items/c3bad2c95a577fcc9c9d

参考元

の情報量が若干少ないので、日本語でもう少し解説の情報を増やしてみた記事です。
英語のできる方や冗長な言い回しが苦手な方は、上記のリンクから参考元サイトを見ることができます。

MVCについて

Model-View-Controllerモデルについてほんの少しで良いので知っておく必要があります。
なぜならこの記事でモデルとかコントローラーとかの単語を出すからです。
↓の記事とかが参考になると思います。
MVCモデルについて

筆者の環境

Ubuntu20.04LTS
MySQL 8.0.21
Python3.8.5

Flask==1.1.2
Flask-Admin==1.5.6
Flask-Login==0.5.0
Flask-SQLAlchemy==2.4.4
mysqlclient==2.0.1

ソースコード全体

いきなりですが、お時間がない方用に最終的なソースコード全体をお見せします。
詳しい解説はこれ以降。

from flask import Flask, abort, jsonify, render_template, request, redirect, url_for
from wtforms import form, fields, validators
import flask_admin as admin
import flask_login as login
from flask_admin.contrib import sqla
from flask_admin import helpers, expose
from flask_admin.contrib.sqla import ModelView
from werkzeug.security import generate_password_hash, check_password_hash
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = "mysql://{user}:{password}@{host}/{db_name}".format(**{
  'user': os.environ['RDS_USER'],
  'password': os.environ['RDS_PASS'],
  'host': os.environ['RDS_HOST'],
  'db_name': os.environ['RDS_DB_NAME']
})
app.config['SECRET_KEY'] = os.environ['FLASK_SECRET_KEY']
db = SQLAlchemy(app)

class AdminUser(db.Model):
  id = db.Column(db.Integer, primary_key=True)
  login = db.Column(db.String(50), unique=True)
  password = db.Column(db.String(250))

  @property
  def is_authenticated(self):
    return True

  @property
  def is_active(self):
    return True

  @property
  def is_anonymous(self):
    return False

  def get_id(self):
    return self.id

  def __unicode__(self):
    return self.username


class LoginForm(form.Form):
  login = fields.StringField(validators=[validators.required()])
  password = fields.PasswordField(validators=[validators.required()])

  def validate_login(self, field):
    user = self.get_user()

    if user is None:
      raise validators.ValidationError('ユーザー名もしくはパスワードが違います。')

    if not check_password_hash(user.password, self.password.data):
      raise validators.ValidationError('ユーザー名もしくはパスワードが違います。')

  def get_user(self):
    return db.session.query(AdminUser).filter_by(login=self.login.data).first()


class RegistrationForm(form.Form):
  login = fields.StringField(validators=[validators.required()])
  password = fields.PasswordField(validators=[validators.required()])

  def validate_login(self, field):
    if db.session.query(AdminUser).filter_by(login=self.login.data).count() > 0:
      raise validators.ValidationError('同じユーザー名が存在します。')


def init_login():
  login_manager = login.LoginManager()
  login_manager.init_app(app)

  @login_manager.user_loader
  def load_user(user_id):
    return db.session.query(AdminUser).get(user_id)


class MyModelView(sqla.ModelView):
  def is_accessible(self):
    return login.current_user.is_authenticated


class MyAdminIndexView(admin.AdminIndexView):
  @expose('/')
  def index(self):
    if not login.current_user.is_authenticated:
      return redirect(url_for('.login_view'))
    return super(MyAdminIndexView, self).index()

  @expose('/login/', methods=('GET', 'POST'))
  def login_view(self):
    form = LoginForm(request.form)
    if helpers.validate_form_on_submit(form):
      user = form.get_user()
      login.login_user(user)

    if login.current_user.is_authenticated:
      return redirect(url_for('.index'))
    link = '<p>アカウント未作成用 <a href="' + url_for('.register_view') + '">ここをクリック</a></p>'
    self._template_args['form'] = form
    self._template_args['link'] = link
    return super(MyAdminIndexView, self).index()
  
  @expose('/register/', methods=('GET', 'POST'))
  def register_view(self):
    form = RegistrationForm(request.form)
    if helpers.validate_form_on_submit(form):
      user = AdminUser()

      form.populate_obj(user)
      user.password = generate_password_hash(form.password.data)
      db.session.add(user)
      db.session.commit()
      login.login_user(user)
      return redirect(url_for('.index'))
    link = '<p>既にアカウントを持っている場合は <a href="' + url_for('.login_view') + '">ここをクリックしてログイン</a></p>'
    self._template_args['form'] = form
    self._template_args['link'] = link
    return super(MyAdminIndexView, self).index()

  @expose('/logout/')
  def logout_view(self):
    login.logout_user()
    return redirect(url_for('.index'))


init_login()
admin = admin.Admin(app, '管理者画面', index_view=MyAdminIndexView(), base_template='my_master.html')
admin.add_view(MyModelView(AdminUser, db.session))


@app.route("/", methods=['GET'])
def index():
  return "Hello, World!"

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

DBに接続する

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = "mysql://{user}:{password}@{host}/{db_name}".format(**{
  'user': os.environ['RDS_USER'],
  'password': os.environ['RDS_PASS'],
  'host': os.environ['RDS_HOST'],
  'db_name': os.environ['RDS_DB_NAME']
})
app.config['SECRET_KEY'] = os.environ['FLASK_SECRET_KEY']
db = SQLAlchemy(app)

ここの解説はインターネット上に日本語の記事も多いので省略します。

管理者アカウントのモデルを作成する

class AdminUser(db.Model):
  id = db.Column(db.Integer, primary_key=True)
  login = db.Column(db.String(50), unique=True)
  password = db.Column(db.String(250))

  @property
  def is_authenticated(self):
    return True

  @property
  def is_active(self):
    return True

  @property
  def is_anonymous(self):
    return False

  def get_id(self):
    return self.id

  def __unicode__(self):
    return self.username

loginはユーザー名のことです。
管理者画面にログインする際のユーザー名とパスワードを定義しています。
各メソッドにpropertyデコレータが付いてるのは、後にログイン処理を書いていくときにログイン済かどうかとかそういう情報を取得するためです。
propertyデコレータについて詳しく知りたい方は↓
プロパティ

コントローラーの作成


class LoginForm(form.Form):
  login = fields.StringField(validators=[validators.required()])
  password = fields.PasswordField(validators=[validators.required()])

  def validate_login(self, field):
    user = self.get_user()

    if user is None:
      raise validators.ValidationError('ユーザー名もしくはパスワードが違います。')

    if not check_password_hash(user.password, self.password.data):
      raise validators.ValidationError('ユーザー名もしくはパスワードが違います。')

  def get_user(self):
    return db.session.query(AdminUser).filter_by(login=self.login.data).first()


class RegistrationForm(form.Form):
  login = fields.StringField(validators=[validators.required()])
  password = fields.PasswordField(validators=[validators.required()])

  def validate_login(self, field):
    if db.session.query(AdminUser).filter_by(login=self.login.data).count() > 0:
      raise validators.ValidationError('同じユーザー名が存在します。')

ビュー(ログイン画面とか管理者アカウント登録画面)のフォームから入力を受けた際の処理を書いたコントローラーです。
ここで注目してほしいのは、LoginFormクラスにある

check_password_hash(user.password, self.password.data)

です。これはハッシュ化されて保存してある本当のパスワードと、ログイン画面から入力された値をハッシュ化したものを、両者比較し一致したらTrueを返しれくれる便利なやつです。

推奨されないですが、もしDBにパスワードを平文で保存している時は条件式のところを

if user.password != self.password.data:

に変更すると良いと思います。

ビューの作成


def init_login():
  login_manager = login.LoginManager()
  login_manager.init_app(app)

  @login_manager.user_loader
  def load_user(user_id):
    return db.session.query(AdminUser).get(user_id)


class MyModelView(sqla.ModelView):
  def is_accessible(self):
    return login.current_user.is_authenticated


class MyAdminIndexView(admin.AdminIndexView):
  @expose('/')
  def index(self):
    if not login.current_user.is_authenticated:
      return redirect(url_for('.login_view'))
    return super(MyAdminIndexView, self).index()

  @expose('/login/', methods=('GET', 'POST'))
  def login_view(self):
    form = LoginForm(request.form)
    if helpers.validate_form_on_submit(form):
      user = form.get_user()
      login.login_user(user)

    if login.current_user.is_authenticated:
      return redirect(url_for('.index'))
    link = '<p>アカウント未作成用 <a href="' + url_for('.register_view') + '">ここをクリック</a></p>'
    self._template_args['form'] = form
    self._template_args['link'] = link
    return super(MyAdminIndexView, self).index()
  
  @expose('/register/', methods=('GET', 'POST'))
  def register_view(self):
    form = RegistrationForm(request.form)
    if helpers.validate_form_on_submit(form):
      user = AdminUser()

      form.populate_obj(user)
      user.password = generate_password_hash(form.password.data)
      db.session.add(user)
      db.session.commit()
      login.login_user(user)
      return redirect(url_for('.index'))
    link = '<p>既にアカウントを持っている場合は <a href="' + url_for('.login_view') + '">ここをクリックしてログイン</a></p>'
    self._template_args['form'] = form
    self._template_args['link'] = link
    return super(MyAdminIndexView, self).index()

  @expose('/logout/')
  def logout_view(self):
    login.logout_user()
    return redirect(url_for('.index'))

init_login()
admin = admin.Admin(app, '管理者画面', index_view=MyAdminIndexView(), base_template='my_master.html')
admin.add_view(MyModelView(AdminUser, db.session))

普通にFlaskでやる時と若干似てる感じですね。

ここで注目してほしいのはMyModelViewクラスです。
MyModelViewはsqla.ModelViewを継承し、is_accessibleメソッドをオーバライドしています。(する必要があるのです)
is_accessibleメソッドでは、ユーザーが既にログイン済みか否かを返しています。
is_accessibleメソッドをオーバライドするだけで、後のビュークラス(ここではMyAdminIndexViewクラス)でアクセス制御ルールを定義できるようになります。

init_login()
admin = admin.Admin(app, '管理者画面', index_view=MyAdminIndexView(), base_template='my_master.html')
admin.add_view(MyModelView(AdminUser, db.session))

で実際にどのモデルにおいてどのビュークラスを使用するかなどを定義しています。

HTMLを書く

HTMLがないと意味がないですね。
プロジェクトルートディレクトリにtemplatesディレクトリを作成し、以下のような構造でファイルやディレクトリを作ります。

templates/
    admin/
        index.html
    my_master.html
    index.html

my_master.html

{% extends 'admin/base.html' %}

{% block access_control %}
{% if current_user.is_authenticated %}
<div class="btn-group pull-right">
  <a class="btn dropdown-toggle" data-toggle="dropdown" href="#">
    <i class="icon-user"></i> {{ current_user.login }} <span class="caret"></span>
  </a>
  <ul class="dropdown-menu">
    <li><a href="{{ url_for('admin.logout_view') }}">ログアウト</a></li>
  </ul>
</div>
{% endif %}
{% endblock %}

ログイン後の画面で、ユーザーIDのところを押されたらドロップダウンでログアウトボタンが出るようなやつです。

templates/index.html

<html>
  <body>
    <div>
      <a href="{{ url_for('admin.index') }}">Go to admin!</a>
    </div>
  </body>
</html>

インデックスページなんで何でも良いです。
適当に書いておきます。

templates/admin/index.html

{% extends 'admin/master.html' %}
{% block body %}
{{ super() }}
<div class="row-fluid">

    <div>
        {% if current_user.is_authenticated %}
        <h1>Civichat管理者画面</h1>
        <p class="lead">
            認証済
        </p>
        <p>
            データの管理はこちらの画面からできます。ログアウトしたい場合は/admin/logout にアクセスしてください。
        </p>
        {% else %}
        <form method="POST" action="">
            {{ form.hidden_tag() if form.hidden_tag }}
            {% for f in form if f.type != 'CSRFTokenField' %}
            <div>
            {{ f.label }}
            {{ f }}
            {% if f.errors %}
            <ul>
                {% for e in f.errors %}
                <li>{{ e }}</li>
                {% endfor %}
            </ul>
            {% endif %}
            </div>
            {% endfor %}
            <button class="btn" type="submit">完了</button>
        </form>
        {{ link | safe }}
        {% endif %}
    </div>

    <a class="btn btn-primary" href="/"><i class="icon-arrow-left icon-white"></i> 戻る</a>
</div>
{% endblock body %}

ログイン後の管理者画面のインデックスページのようなものです。

パスワード認証に加えIP制限したい

そもそもログインフォームにたどり着く前にIPアドレスでアクセス制限したい、というニーズもあると思います。
↓の記事が参考になると思います。

FlaskでIP制限する

最後に

何か間違いがあったらコメントでご指摘頂ければ幸いです。

Discussion

ペトリコールペトリコール

丁寧な記事ありがとうございます。とても参考になりました。

Python、特にFlaskは初めてなので、私の勘違いならスルーしていただければと思いますが、
'user': os.environ['RDS_USER'] であれば import os も必要ででしょうか。
また、RDS_USERなどのPythonの環境変数は、別途セットすることが前提でしょうか。
Pythonの常識がないので、常識的なことであったらすみません。