Python

Python・FlaskでMySQLログイン認証を作る方法―MVC構成、登録確認メール、パスワード再設定まで詳しく解説

PythonでWebアプリを作り始めると、早い段階で必要になるのがログイン認証です。単にユーザー名とパスワードを照合するだけなら短いコードでも作れますが、実際のサービスではアカウント作成、入力値の確認、メールアドレスの有効性確認、ログアウト、パスワードを忘れた場合の再設定まで必要です。

この記事では、PythonのWebフレームワーク「Flask」とMySQLを使い、認証機能をMVC形式で整理して作る手順を解説します。認証情報は「ユーザー名・メールアドレス・パスワード」の3項目です。パスワードは平文で保存せず、ハッシュ化して安全に管理します。

この記事で作る機能

  • ユーザー名、メールアドレス、パスワードによるアカウント作成
  • フォーム入力値のバリデーション
  • 確認メールによるメールアドレスの有効性確認
  • ログイン、ログアウト
  • ログインが必要なページの保護
  • パスワード紛失時の再設定メール送信
  • 期限付きトークンによるパスワード再設定
  • MySQLへのユーザー情報保存

使用する構成

今回は次のライブラリを使います。

  • Flask:Pythonの軽量Webフレームワーク
  • Flask-SQLAlchemy:MySQLのデータをPythonのクラスとして扱う
  • Flask-Migrate:テーブル構造の変更を管理する
  • Flask-Login:ログイン状態を管理する
  • Flask-WTF / WTForms:フォームとバリデーション、CSRF対策
  • Flask-Mail:確認メールや再設定メールを送信する
  • itsdangerous:有効期限付きの安全なトークンを作る
  • PyMySQL:PythonからMySQLへ接続する
  • python-dotenv:環境変数を読み込む

MVC形式の考え方

Flaskにはフォルダ構成を強制する決まりがありません。そのため、すべてを1つの app.py に書くこともできます。しかし、機能が増えると修正箇所が分かりにくくなります。今回は役割ごとに分離します。

  • Model:MySQLのテーブルとデータ操作。今回は models/user.py
  • View:利用者に表示するHTML。今回は templates/
  • Controller:URLごとの処理、入力確認、画面遷移。今回は controllers/auth_controller.py

FlaskではControllerをBlueprintで分けると、認証機能を他の機能から切り離しやすくなります。

完成時のファイル構成

python-example/
├── run.py
├── requirements.txt
├── .env
├── .gitignore
├── migrations/
└── app/
    ├── __init__.py
    ├── config.py
    ├── extensions.py
    ├── models/
    │   ├── __init__.py
    │   └── user.py
    ├── forms/
    │   ├── __init__.py
    │   └── auth_form.py
    ├── controllers/
    │   ├── __init__.py
    │   └── auth_controller.py
    ├── services/
    │   ├── __init__.py
    │   ├── mail_service.py
    │   └── token_service.py
    └── templates/
        ├── base.html
        ├── index.html
        └── auth/
            ├── register.html
            ├── login.html
            ├── forgot_password.html
            └── reset_password.html

1. Pythonの仮想環境を作る

mkdir python-example
cd python-example
python3 -m venv venv
source venv/bin/activate

python3 -m venv venv は、プロジェクト専用のPython環境を作るコマンドです。ほかのPythonアプリとライブラリのバージョンが混ざるのを防ぎます。Windowsの場合は venv\Scripts\activate で有効化します。

2. 必要なライブラリをインストールする

pip install Flask Flask-SQLAlchemy Flask-Migrate Flask-Login Flask-WTF Flask-Mail email-validator PyMySQL python-dotenv
pip freeze > requirements.txt

email-validator は、メールアドレスの形式をWTFormsで確認するために必要です。インストール後の状態を requirements.txt に記録しておけば、別のサーバーでも pip install -r requirements.txt で同じ環境を作れます。

3. MySQLにデータベースと専用ユーザーを作る

MySQLへ管理権限のあるユーザーでログインします。

mysql -u root -p

続いて、データベースとアプリ専用ユーザーを作成します。

CREATE DATABASE python_example
  CHARACTER SET utf8mb4
  COLLATE utf8mb4_unicode_ci;

CREATE USER 'python_user'@'localhost'
  IDENTIFIED BY '十分に長く推測されにくいパスワード';

GRANT SELECT, INSERT, UPDATE, DELETE, CREATE, ALTER, INDEX, DROP
  ON python_example.* TO 'python_user'@'localhost';

FLUSH PRIVILEGES;
EXIT;

utf8mb4 を指定すると、日本語だけでなく絵文字も保存できます。本番環境ではrootユーザーをアプリから使わず、対象データベースだけを操作できる専用ユーザーを使います。

4. 環境変数を設定する

パスワードやメールサーバーの認証情報をソースコードへ直接書かないように、プロジェクト直下へ .env を作ります。

SECRET_KEY=ここに長いランダム文字列
DATABASE_URL=mysql+pymysql://python_user:DBパスワード@localhost/python_example?charset=utf8mb4

MAIL_SERVER=smtp.example.com
MAIL_PORT=587
MAIL_USE_TLS=true
MAIL_USERNAME=no-reply@example.com
MAIL_PASSWORD=メール用パスワード
MAIL_DEFAULT_SENDER=no-reply@example.com

SECRET_KEY はCSRF対策やトークン署名に使います。次のコマンドで生成できます。

python -c "import secrets; print(secrets.token_hex(32))"

.gitignore には必ず次を記載します。

venv/
.env
__pycache__/
*.pyc

5. 設定ファイルを作る

app/config.py を作成します。

import os
from dotenv import load_dotenv

load_dotenv()

class Config:
    SECRET_KEY = os.environ["SECRET_KEY"]
    SQLALCHEMY_DATABASE_URI = os.environ["DATABASE_URL"]
    SQLALCHEMY_TRACK_MODIFICATIONS = False

    MAIL_SERVER = os.environ["MAIL_SERVER"]
    MAIL_PORT = int(os.getenv("MAIL_PORT", "587"))
    MAIL_USE_TLS = os.getenv("MAIL_USE_TLS", "true").lower() == "true"
    MAIL_USERNAME = os.environ["MAIL_USERNAME"]
    MAIL_PASSWORD = os.environ["MAIL_PASSWORD"]
    MAIL_DEFAULT_SENDER = os.environ["MAIL_DEFAULT_SENDER"]

os.environ["名前"] は、設定がない場合にエラーとなります。重要な設定の不足に気づきやすいため、本番用の認証情報にはこの書き方が向いています。

6. 拡張機能をまとめる

app/extensions.py を作ります。

from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_login import LoginManager
from flask_mail import Mail

db = SQLAlchemy()
migrate = Migrate()
login_manager = LoginManager()
mail = Mail()

login_manager.login_view = "auth.login"
login_manager.login_message = "ログインしてください。"
login_manager.login_message_category = "warning"

ここでは拡張機能の本体だけを作り、アプリとの接続は後ほど create_app() の中で行います。この形式はApplication Factoryパターンと呼ばれ、テストや設定の切り替えがしやすくなります。

7. Userモデルを作る

app/models/user.py を作成します。

from datetime import datetime, timezone
from flask_login import UserMixin
from werkzeug.security import generate_password_hash, check_password_hash
from app.extensions import db, login_manager

class User(UserMixin, db.Model):
    __tablename__ = "users"

    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(50), unique=True, nullable=False, index=True)
    email = db.Column(db.String(255), unique=True, nullable=False, index=True)
    password_hash = db.Column(db.String(255), nullable=False)
    email_verified = db.Column(db.Boolean, nullable=False, default=False)
    created_at = db.Column(
        db.DateTime,
        nullable=False,
        default=lambda: datetime.now(timezone.utc)
    )

    def set_password(self, password):
        self.password_hash = generate_password_hash(password)

    def check_password(self, password):
        return check_password_hash(self.password_hash, password)

@login_manager.user_loader
def load_user(user_id):
    return db.session.get(User, int(user_id))

データベースにはパスワードそのものではなく password_hash を保存します。generate_password_hash() は同じパスワードでも毎回異なるハッシュ値を作ります。ログイン時は check_password() で照合します。

email_verified はメール確認が完了したかを示す項目です。登録直後は False、確認用URLを開いたあとに True へ変更します。

8. フォームとバリデーションを作る

app/forms/auth_form.py を作成します。

from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, SubmitField
from wtforms.validators import (
    DataRequired, Email, EqualTo, Length, Regexp, ValidationError
)
from app.models.user import User

class RegisterForm(FlaskForm):
    username = StringField(
        "ユーザー名",
        validators=[
            DataRequired(),
            Length(min=3, max=50),
            Regexp(
                r"^[A-Za-z0-9_]+$",
                message="半角英数字とアンダースコアのみ使用できます。"
            )
        ]
    )
    email = StringField(
        "メールアドレス",
        validators=[DataRequired(), Email(), Length(max=255)]
    )
    password = PasswordField(
        "パスワード",
        validators=[
            DataRequired(),
            Length(min=12, max=128),
            Regexp(r"^(?=.*[a-z])(?=.*[A-Z])(?=.*\d).+$",
                   message="英大文字・英小文字・数字を含めてください。")
        ]
    )
    password_confirm = PasswordField(
        "パスワード(確認)",
        validators=[DataRequired(), EqualTo("password")]
    )
    submit = SubmitField("アカウントを作成")

    def validate_username(self, field):
        if User.query.filter_by(username=field.data.strip()).first():
            raise ValidationError("このユーザー名はすでに使われています。")

    def validate_email(self, field):
        email = field.data.strip().lower()
        if User.query.filter_by(email=email).first():
            raise ValidationError("このメールアドレスはすでに登録されています。")

class LoginForm(FlaskForm):
    login_id = StringField(
        "ユーザー名またはメールアドレス",
        validators=[DataRequired(), Length(max=255)]
    )
    password = PasswordField("パスワード", validators=[DataRequired()])
    submit = SubmitField("ログイン")

class ForgotPasswordForm(FlaskForm):
    email = StringField(
        "メールアドレス",
        validators=[DataRequired(), Email(), Length(max=255)]
    )
    submit = SubmitField("再設定メールを送信")

class ResetPasswordForm(FlaskForm):
    password = PasswordField(
        "新しいパスワード",
        validators=[
            DataRequired(),
            Length(min=12, max=128),
            Regexp(r"^(?=.*[a-z])(?=.*[A-Z])(?=.*\d).+$",
                   message="英大文字・英小文字・数字を含めてください。")
        ]
    )
    password_confirm = PasswordField(
        "新しいパスワード(確認)",
        validators=[DataRequired(), EqualTo("password")]
    )
    submit = SubmitField("パスワードを変更")

ブラウザ側の required だけでは不十分です。利用者はHTTPリクエストを直接送ることもできるため、必ずPython側でも確認します。主な確認内容は、必須入力、長さ、メール形式、使用可能文字、パスワード一致、重複登録です。

9. 有効期限付きトークンを作る

app/services/token_service.py を作ります。

from flask import current_app
from itsdangerous import URLSafeTimedSerializer, BadSignature, SignatureExpired

def _serializer():
    return URLSafeTimedSerializer(current_app.config["SECRET_KEY"])

def create_token(user_id, purpose):
    return _serializer().dumps({"user_id": user_id}, salt=purpose)

def verify_token(token, purpose, max_age=3600):
    try:
        data = _serializer().loads(token, salt=purpose, max_age=max_age)
        return data.get("user_id")
    except (SignatureExpired, BadSignature):
        return None

purpose には、メール確認なら email-confirm、パスワード再設定なら password-reset を指定します。用途ごとにsaltを分けることで、確認用トークンをパスワード再設定へ流用できないようにします。max_age=3600 は有効期限1時間です。

10. メール送信処理を作る

app/services/mail_service.py を作成します。

from flask import url_for
from flask_mail import Message
from app.extensions import mail
from app.services.token_service import create_token

def send_verification_email(user):
    token = create_token(user.id, "email-confirm")
    verify_url = url_for(
        "auth.verify_email", token=token, _external=True
    )
    message = Message(
        subject="メールアドレスの確認",
        recipients=[user.email],
        body=f"""アカウント作成ありがとうございます。
次のURLを1時間以内に開いてください。

{verify_url}

心当たりがない場合は、このメールを破棄してください。
"""
    )
    mail.send(message)

def send_password_reset_email(user):
    token = create_token(user.id, "password-reset")
    reset_url = url_for(
        "auth.reset_password", token=token, _external=True
    )
    message = Message(
        subject="パスワード再設定",
        recipients=[user.email],
        body=f"""次のURLを1時間以内に開き、新しいパスワードを設定してください。

{reset_url}

依頼していない場合は、このメールを破棄してください。
"""
    )
    mail.send(message)

_external=True を指定すると、メール本文で利用できる絶対URLが作られます。本番環境ではHTTPSで公開し、正しいドメインが生成されるようにWebサーバー側のHost設定も確認します。

11. Controllerを作る

app/controllers/auth_controller.py を作ります。

from flask import Blueprint, render_template, redirect, url_for, flash, request
from flask_login import login_user, logout_user, login_required, current_user
from sqlalchemy import or_
from app.extensions import db
from app.models.user import User
from app.forms.auth_form import (
    RegisterForm, LoginForm, ForgotPasswordForm, ResetPasswordForm
)
from app.services.mail_service import (
    send_verification_email, send_password_reset_email
)
from app.services.token_service import verify_token

auth_bp = Blueprint("auth", __name__, url_prefix="/auth")

@auth_bp.route("/register", methods=["GET", "POST"])
def register():
    if current_user.is_authenticated:
        return redirect(url_for("index"))

    form = RegisterForm()
    if form.validate_on_submit():
        user = User(
            username=form.username.data.strip(),
            email=form.email.data.strip().lower()
        )
        user.set_password(form.password.data)
        db.session.add(user)
        db.session.commit()

        send_verification_email(user)
        flash("確認メールを送信しました。メール内のURLを開いてください。", "success")
        return redirect(url_for("auth.login"))

    return render_template("auth/register.html", form=form)

@auth_bp.route("/verify-email/<token>")
def verify_email(token):
    user_id = verify_token(token, "email-confirm", max_age=3600)
    user = db.session.get(User, user_id) if user_id else None

    if not user:
        flash("確認URLが無効、または有効期限が切れています。", "danger")
        return redirect(url_for("auth.login"))

    if not user.email_verified:
        user.email_verified = True
        db.session.commit()

    flash("メールアドレスを確認しました。ログインできます。", "success")
    return redirect(url_for("auth.login"))

@auth_bp.route("/login", methods=["GET", "POST"])
def login():
    if current_user.is_authenticated:
        return redirect(url_for("index"))

    form = LoginForm()
    if form.validate_on_submit():
        login_id = form.login_id.data.strip()
        user = User.query.filter(
            or_(User.username == login_id, User.email == login_id.lower())
        ).first()

        if not user or not user.check_password(form.password.data):
            flash("ユーザー名、メールアドレス、またはパスワードが違います。", "danger")
            return render_template("auth/login.html", form=form)

        if not user.email_verified:
            flash("メールアドレスの確認が完了していません。", "warning")
            return render_template("auth/login.html", form=form)

        login_user(user)
        next_url = request.args.get("next")
        if next_url and next_url.startswith("/"):
            return redirect(next_url)
        return redirect(url_for("index"))

    return render_template("auth/login.html", form=form)

@auth_bp.route("/logout", methods=["POST"])
@login_required
def logout():
    logout_user()
    flash("ログアウトしました。", "success")
    return redirect(url_for("auth.login"))

@auth_bp.route("/forgot-password", methods=["GET", "POST"])
def forgot_password():
    form = ForgotPasswordForm()
    if form.validate_on_submit():
        user = User.query.filter_by(
            email=form.email.data.strip().lower()
        ).first()

        if user:
            send_password_reset_email(user)

        flash("登録済みの場合は、再設定メールを送信しました。", "success")
        return redirect(url_for("auth.login"))

    return render_template("auth/forgot_password.html", form=form)

@auth_bp.route("/reset-password/<token>", methods=["GET", "POST"])
def reset_password(token):
    user_id = verify_token(token, "password-reset", max_age=3600)
    user = db.session.get(User, user_id) if user_id else None

    if not user:
        flash("再設定URLが無効、または有効期限が切れています。", "danger")
        return redirect(url_for("auth.forgot_password"))

    form = ResetPasswordForm()
    if form.validate_on_submit():
        user.set_password(form.password.data)
        db.session.commit()
        flash("パスワードを変更しました。", "success")
        return redirect(url_for("auth.login"))

    return render_template("auth/reset_password.html", form=form)

パスワード再設定では、メールアドレスが未登録の場合でも同じ案内を表示しています。「そのメールアドレスが登録されているか」を第三者に知られにくくするためです。また、ログアウトをGETではなくPOSTにしているのは、外部サイトのリンクを開いただけでログアウトさせられる動作を防ぐためです。

12. Flaskアプリを初期化する

app/__init__.py を作成します。

from flask import Flask, render_template
from flask_login import login_required
from app.config import Config
from app.extensions import db, migrate, login_manager, mail

def create_app():
    app = Flask(__name__)
    app.config.from_object(Config)

    db.init_app(app)
    migrate.init_app(app, db)
    login_manager.init_app(app)
    mail.init_app(app)

    from app.controllers.auth_controller import auth_bp
    app.register_blueprint(auth_bp)

    @app.route("/")
    def index():
        return render_template("index.html")

    @app.route("/mypage")
    @login_required
    def mypage():
        return "ログイン済みのユーザーだけが表示できます。"

    return app

run.py は次の内容です。

from app import create_app

app = create_app()

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

13. HTMLテンプレートを作る

app/templates/base.html では、CSRFトークンを含むログアウトフォームとメッセージ表示を用意します。

<!doctype html>
<html lang="ja">
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <title>{% block title %}Python Example{% endblock %}</title>
</head>
<body>
  <nav>
    <a href="{{ url_for('index') }}">ホーム</a>
    {% if current_user.is_authenticated %}
      <form method="post" action="{{ url_for('auth.logout') }}">
        <input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
        <button type="submit">ログアウト</button>
      </form>
    {% else %}
      <a href="{{ url_for('auth.login') }}">ログイン</a>
      <a href="{{ url_for('auth.register') }}">アカウント作成</a>
    {% endif %}
  </nav>

  {% with messages = get_flashed_messages(with_categories=true) %}
    {% for category, message in messages %}
      <p class="{{ category }}">{{ message }}</p>
    {% endfor %}
  {% endwith %}

  {% block content %}{% endblock %}
</body>
</html>

フォームは共通して、ラベル、入力欄、バリデーションエラーを表示します。例として app/templates/auth/register.html を示します。

{% extends "base.html" %}
{% block title %}アカウント作成{% endblock %}
{% block content %}
<h1>アカウント作成</h1>
<form method="post" novalidate>
  {{ form.hidden_tag() }}

  {{ form.username.label }}
  {{ form.username(autocomplete="username") }}
  {% for error in form.username.errors %}
    <p class="error">{{ error }}</p>
  {% endfor %}

  {{ form.email.label }}
  {{ form.email(autocomplete="email") }}
  {% for error in form.email.errors %}
    <p class="error">{{ error }}</p>
  {% endfor %}

  {{ form.password.label }}
  {{ form.password(autocomplete="new-password") }}
  {% for error in form.password.errors %}
    <p class="error">{{ error }}</p>
  {% endfor %}

  {{ form.password_confirm.label }}
  {{ form.password_confirm(autocomplete="new-password") }}
  {% for error in form.password_confirm.errors %}
    <p class="error">{{ error }}</p>
  {% endfor %}

  {{ form.submit() }}
</form>
{% endblock %}

form.hidden_tag() にはCSRFトークンが含まれます。第三者サイトから勝手にフォームを送信される攻撃を防ぐため、省略しないようにします。ログイン画面にはユーザー名またはメールアドレス、パスワード、ログインボタンと、パスワード再設定画面へのリンクを置きます。

14. MySQLへテーブルを作成する

モデルを読み込める状態にしてからマイグレーションを実行します。

export FLASK_APP=run.py
flask db init
flask db migrate -m "create users table"
flask db upgrade

flask db init は初回だけ実行します。flask db migrate はモデルとの差分から変更ファイルを作り、flask db upgrade が実際にMySQLへ反映します。本番環境では、生成された変更内容を確認してから適用してください。

テーブルを確認する場合は次のようにします。

mysql -u python_user -p python_example
SHOW TABLES;
DESCRIBE users;

15. 動作を確認する

flask --app run.py run --debug

ブラウザで http://127.0.0.1:5000/auth/register を開き、次の順番で確認します。

  1. 未入力や不正なメール形式でエラーが表示される
  2. 短いパスワードや条件を満たさないパスワードが拒否される
  3. 正常に登録するとMySQLへユーザーが追加される
  4. 確認メールが届き、URLを開くと email_verified が有効になる
  5. 確認前はログインできず、確認後はログインできる
  6. 同じユーザー名やメールアドレスを重複登録できない
  7. パスワード再設定メールが届き、新しいパスワードに変更できる
  8. 期限切れ、改ざん済みトークンが拒否される
  9. 未ログインで /mypage を開くとログイン画面へ移動する

本番公開前に追加したい対策

ここまでで基本的な認証は動きますが、インターネットへ公開する場合は次の対策も重要です。

  • 必ずHTTPSを使用する
  • ログインや再設定メール送信へ回数制限を入れる
  • 連続失敗を記録し、総当たり攻撃を検知する
  • セッションCookieへ SecureHttpOnly、適切な SameSite を設定する
  • 確認メールの再送機能を用意し、再送にも回数制限を入れる
  • メール送信はCeleryなどのバックグラウンド処理へ分離する
  • パスワード変更後に既存セッションを無効化する仕組みを検討する
  • ユーザー入力や認証結果をログへ残す際、パスワードやトークンを書かない
  • MySQLとアプリのバックアップ、復元手順を用意する

Cookie設定の一例は次のとおりです。

SESSION_COOKIE_SECURE = True
SESSION_COOKIE_HTTPONLY = True
SESSION_COOKIE_SAMESITE = "Lax"
REMEMBER_COOKIE_SECURE = True
REMEMBER_COOKIE_HTTPONLY = True

よくあるエラー

Access denied for user

MySQLのユーザー名、パスワード、接続元ホスト、GRANTの対象を確認します。'python_user'@'localhost''python_user'@'127.0.0.1' はMySQLでは別の接続元として扱われる場合があります。

No module named PyMySQL

仮想環境が有効か確認し、pip install PyMySQL を実行します。Gunicornやsystemdから起動する場合は、仮想環境内の実行ファイルを指定します。

確認メールのURLがlocalhostになる

本番ドメインやリバースプロキシの設定が正しくFlaskへ渡っていない可能性があります。ApacheやNginxのHostヘッダー、HTTPS判定、必要に応じてProxyFixの設定を確認します。

CSRF token is missing

テンプレート内のフォームに {{ form.hidden_tag() }} があるか確認します。通常のHTMLフォームを使う場合も、CSRFトークンのhidden項目が必要です。

まとめ

PythonとFlaskでログイン認証を作る場合は、ログイン処理だけでなく、登録時の入力確認、メールアドレス確認、パスワードのハッシュ化、再設定用トークンの期限管理までを一つの流れとして考えることが大切です。

MVC形式でModel、View、Controllerを分け、メール処理とトークン処理をServiceへ分離しておけば、機能追加やテストもしやすくなります。まずはローカル環境で登録から再設定まで一通り確認し、その後にHTTPS、回数制限、ログ管理など本番向けの安全対策を追加すると進めやすいでしょう。

コメント

タイトルとURLをコピーしました