Flask-Security with "confirm change of email" functionality

_flask_security_confirm_change_email.md

Flask-Security with "confirm change of email" functionality

A simple Flask app to demonstrate Flask-Security with "confirm change of email" functionality added.

To try it out locally:

pip install -r requirements.txt
python app.py

Then navigate to http://localhost:5000/ .

app.py

"""Flask-Security with 'confirm change of email' functionality"""
import os

from flask import Flask, render_template
from flask_mail import Mail
from flask_security import (
    Security, SQLAlchemyUserDatastore, UserMixin)
from flask_sqlalchemy import SQLAlchemy

from change_email import change_email, confirm_change_email
from send_mail import send_mail


tmpl_dir = os.path.dirname(os.path.abspath(__file__))

app = Flask(__name__, template_folder=tmpl_dir)

app.config.update(
    DEBUG=True,
    SQLALCHEMY_DATABASE_URI='sqlite:///dev.db',
    SECRET_KEY='1a2b3c4d5e6f',
    SECURITY_CONFIRMABLE=True,
    SECURITY_REGISTERABLE=True,
    SECURITY_RECOVERABLE=True,
    SECURITY_CHANGEABLE=True,
    SQLALCHEMY_TRACK_MODIFICATIONS=False,
    SECURITY_PASSWORD_HASH='bcrypt',
    SECURITY_PASSWORD_SALT='a1b2c3d4e5f6',
)


Mail(app)
db = SQLAlchemy(app)


class User(db.Model, UserMixin):
    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(255), unique=True)
    password = db.Column(db.String(255))
    active = db.Column(db.Boolean())
    confirmed_at = db.Column(db.DateTime())

    roles = []


user_datastore = SQLAlchemyUserDatastore(db, User, None)
security_ctx = Security(app, user_datastore)


@security_ctx.send_mail_task
def security_send_mail(msg):
    """Send Flask-Security emails via custom function."""
    send_mail(
        sender=msg.sender, recipients=msg.recipients,
        subject=msg.subject, body=msg.body)


@app.before_first_request
def setup_db():
    db.create_all()


@app.route('/')
def home():
    return render_template('home.html')


app.add_url_rule(
    '/confirm-change-email/<token>',
    endpoint=None,
    view_func=confirm_change_email)


app.add_url_rule(
    '/change-email',
    endpoint=None,
    view_func=change_email,
    methods=['GET', 'POST'])


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

change_email.html

{% from "security/_macros.html" import render_field_with_errors, render_field %}
{% include "security/_messages.html" %}
<h1>Change email</h1>
<form action="{{ url_for('change_email') }}" method="POST" name="change_email_form">
  {{ form.hidden_tag() }}
  {{ render_field_with_errors(form.email) }}
  {{ render_field_with_errors(form.new_email) }}
  {{ render_field_with_errors(form.new_email_confirm) }}
  {{ render_field(form.submit) }}
</form>

change_email.py

from flask import current_app as app
from flask import (
    abort, after_this_request, flash, redirect, render_template,
    url_for)
from flask_login import current_user
from flask_security.forms import email_required, email_validator
from flask_security.utils import (
    config_value, do_flash, get_message, get_token_status, hash_data,
    login_user, logout_user, verify_hash)
from flask_wtf import FlaskForm
from werkzeug.local import LocalProxy
from wtforms import StringField, SubmitField
from wtforms.validators import EqualTo

from send_mail import send_mail


_security = LocalProxy(lambda: app.extensions['security'])
_datastore = LocalProxy(lambda: _security.datastore)


def _commit(response=None):
    _datastore.commit()
    return response


class ChangeEmailForm(FlaskForm):
    email = StringField(
        'Email',
        validators=[email_required])
    new_email = StringField(
        'New email',
        validators=[email_required, email_validator])
    new_email_confirm = StringField(
        'Retype email',
        validators=[EqualTo('new_email',
                            message='Email does not match')])
    submit = SubmitField('Change email')

    def validate(self):
        if not super(ChangeEmailForm, self).validate():
            return False

        if self.email.data != current_user.email:
            self.email.errors.append('Invalid email')
            return False
        if self.email.data.strip() == self.new_email.data.strip():
            self.email.errors.append(
                'Your new email must be different than your '
                'previous email')
            return False
        return True


def confirm_change_email_token_status(token):
    """Returns the expired status, invalid status, user, and new email
    of a confirmation token. For example::

        expired, invalid, user, new_email = (
            confirm_change_email_token_status('...'))

    Based on confirm_email_token_status in Flask-Security.

    :param token: The confirmation token
    """
    expired, invalid, user, token_data = get_token_status(
        token, 'confirm', 'CONFIRM_EMAIL', return_data=True)
    new_email = None

    if not invalid and user:
        user_id, token_email_hash, new_email = token_data
        invalid = not verify_hash(token_email_hash, user.email)

    return expired, invalid, user, new_email


def generate_change_email_confirmation_link(user, new_email):
    """Based on generate_confirmation_token in Flask-Security."""
    token = generate_change_email_confirmation_token(user, new_email)
    return (
        url_for('confirm_change_email', token=token, _external=True),
        token)


def generate_change_email_confirmation_token(user, new_email):
    """Generates a unique confirmation token for the specified user.

    Based on generate_confirmation_token in Flask-Security.

    :param user: The user to work with
    :param new_email: The user's new email address
    """
    data = [str(user.id), hash_data(user.email), new_email]
    return _security.confirm_serializer.dumps(data)


def send_change_email_confirmation_instructions(user, new_email):
    """Sends the confirmation instructions email for the specified user.

    Based on send_confirmation_instructions in Flask-Security.

    :param user: The user to send the instructions to
    :param new_email: The user's new email address
    """

    confirmation_link, token = generate_change_email_confirmation_link(
        user, new_email)

    subject = 'Please confim your change of email'
    msg_body = render_template(
        'confirm_change_email.txt', user=current_user,
        new_email=new_email, confirmation_link=confirmation_link)
    send_mail(
        sender=_security.email_sender,
        recipients=[new_email],
        subject=subject, body=msg_body)


def change_user_email(user, new_email):
    """Changes the email for the specified user

    Based on confirm_user in Flask-Security.

    :param user: The user to confirm
    :param new_email: The user's new email address
    """
    if user.email == new_email:
        return False

    user.email = new_email
    _datastore.put(user)

    return True


def confirm_change_email(token):
    """View function which handles a change email confirmation request.

    Based on confirm_email in Flask-Security."""
    expired, invalid, user, new_email = (
        confirm_change_email_token_status(token))

    if not user or invalid:
        invalid = True
        do_flash(*get_message('INVALID_CONFIRMATION_TOKEN'))
    if expired:
        send_change_email_confirmation_instructions(user, new_email)
        do_flash(*(
            (
                'You did not confirm your change of email within {0}. '
                'New instructions to confirm your change of email have '
                'been sent to {1}.').format(
                    _security.confirm_email_within, new_email),
            'error'))
    if invalid or expired:
        return redirect(url_for('home'))

    if user != current_user:
        logout_user()
        login_user(user)

    if change_user_email(user, new_email):
        after_this_request(_commit)
        msg = (
            'Thank you. Your change of email has been confirmed.',
            'success')
    else:
        msg = (
            'Your change of email has already been confirmed.'
            'info')

    do_flash(*msg)
    return redirect(url_for('home'))


def change_email():
    """Change email page."""
    if not _security.confirmable:
        abort(404)

    form = ChangeEmailForm()

    if form.validate_on_submit():
        new_email = form.new_email.data
        confirmation_link, token = (
            generate_change_email_confirmation_link(
                current_user, new_email))
        flash(
            (
                'Thank you. Confirmation instructions for changing '
                'your email have been sent to {0}.').format(new_email),
            'success')

        if config_value('SEND_REGISTER_EMAIL'):
            subject = 'Confirm change of email instructions'
            msg_body = render_template(
                'confirm_change_email.txt', user=current_user,
                new_email=new_email, confirmation_link=confirmation_link)
            send_mail(
                sender=_security.email_sender,
                recipients=[new_email],
                subject=subject, body=msg_body)

        return redirect(url_for('home'))

    return render_template('change_email.html', form=form)

confirm_change_email.txt

Hi {{ new_email }} ,

You recently changed your email address.

Please confirm your new email through the link below:

{{ confirmation_link }}

home.html

{% include "security/_messages.html" %}

<h1>Flask-Security "confirm change of email" functionality</h1>

{% if current_user.is_authenticated %}

<p>Logged in as {{ current_user.email }}</p>

<h2>{{ _('Menu') }}</h2>
<ul>
  {% if security.confirmable %}
  <li><a href="{{ url_for('change_email') }}">Change email</a></li>
  {% endif %}
  {% if security.changeable %}
  <li><a href="{{ url_for_security('change_password') }}">Change password</a><br/></li>
  {% endif %}
  <li><a href="{{ url_for_security('logout') }}">Logout</a></li>
</ul>

{% else %}{# current_user.is_authenticated #}

{% include "security/_menu.html" %}

{% endif %}{# current_user.is_authenticated #}

requirements.txt

Flask
Flask-Security
Flask-SQLAlchemy
bcrypt

send_mail.py

from flask import current_app as app
from flask_mail import Message


def send_mail(
        sender=None, recipients=None, subject=None, body=None,
        is_log_msg=True):
    """Send mail using Flask-Mail and log the sent message."""
    if is_log_msg:
        log_msg = 'Email sent by site\n'
        log_msg += 'From: <{0}>\n'.format(sender)
        log_msg += 'To: {0}\n'.format(recipients)
        log_msg += 'Subject: {0}\n'.format(subject)
        log_msg += body

        app.logger.info(log_msg)

    if app.debug:
        return

    msg = Message(
        subject,
        sender=sender,
        recipients=recipients)
    msg.body = body

    app.mail.send(msg)

Looks very good. It would be nice to include also signal before/after the email address is changed.