explanation of the token-based password-reset functionality in flask-security

def uia_username_mapper(identity):
   # we allow pretty much anything - but we bleach it.
return bleach.clean(identity, strip = True)

[{
   "email": {
      "mapper": uia_email_mapper,
      "case_insensitive": True
   }
}, ]

[{
      "email": {
         "mapper": uia_email_mapper,
         "case_insensitive": True
      }
   },
   {
      "us_phone_number": {
         "mapper": uia_phone_mapper
      }
   },
]

from passlib import totp
"{1: <result of totp.generate_secret()>}"

SmsSenderFactory.senders[<service-name>] = <service-class>

{
   "us_phone_number": {
      "mapper": uia_phone_mapper
   }
},

# ourapp / forms.py

from flask_wtf
import Form
from wtforms
import StringField, PasswordField
from wtforms.validators
import DataRequired, Email

class EmailForm(Form):
   email = TextField('Email', validators = [DataRequired(), Email()])

class PasswordForm(Form):
   password = PasswordField('Email', validators = [DataRequired()])

# ourapp / views.py

from flask
import redirect, url_for, render_template

from.import app
from.forms
import EmailForm
from.models
import User
from.util
import send_email, ts

@app.route('/reset', methods = ["GET", "POST"])
def reset():
   form = EmailForm()
if form.validate_on_submit()
user = User.query.filter_by(email = form.email.data).first_or_404()

subject = "Password reset requested"

# Here we use the URLSafeTimedSerializer we created in `util`
at the
# beginning of the chapter
token = ts.dumps(self.email, salt = 'recover-key')

recover_url = url_for(
   'reset_with_token',
   token = token,
   _external = True)

html = render_template(
   'email/recover.html',
   recover_url = recover_url)

# Let 's assume that send_email was defined in myapp/util.py
send_email(user.email, subject, html)

return redirect(url_for('index'))
return render_template('reset.html', form = form)

# ourapp/views.py

from flask import redirect, url_for, render_template

from . import app, db
from .forms import PasswordForm
from .models import User
from .util import ts

@app.route('/reset/<token>', methods=["GET", "POST"])
   def reset_with_token(token):
   try:
   email = ts.loads(token, salt="recover-key", max_age=86400)
   except:
   abort(404)

   form = PasswordForm()

   if form.validate_on_submit():
   user = User.query.filter_by(email=email).first_or_404()

   user.password = form.password.data

   db.session.add(user)
   db.session.commit()

   return redirect(url_for('signin'))

   return render_template('reset_with_token.html', form=form, token=token)

{# ourapp/templates/reset_with_token.html #}

{% extends "layout.html" %}

{% block body %}
<form action="{{ url_for('reset_with_token', token=token) }}" method="POST">
    {{ form.password.label }}: {{ form.password }}<br>
    {{ form.csrf_token }}
    <input type="submit" value="Change my password" />
</form>
{% endblock %}

$ mkvirtualenv <your-app-name>
   $ pip install flask-security flask-sqlalchemy

from flask
import Flask, render_template
from flask_sqlalchemy
import SQLAlchemy
from flask_security
import Security, SQLAlchemyUserDatastore, \

UserMixin, RoleMixin, login_required
# Create app
app = Flask(__name__)
app.config['DEBUG'] = True
app.config['SECRET_KEY'] = 'super-secret'
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite://'
# Create database connection object
db = SQLAlchemy(app)
# Define models
roles_users = db.Table('roles_users',

   db.Column('user_id', db.Integer(), db.ForeignKey('user.id')),

   db.Column('role_id', db.Integer(), db.ForeignKey('role.id')))
class Role(db.Model, RoleMixin):

   id = db.Column(db.Integer(), primary_key = True)

name = db.Column(db.String(80), unique = True)

description = db.Column(db.String(255))
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 = db.relationship('Role', secondary = roles_users,

   backref = db.backref('users', lazy = 'dynamic'))
# Setup Flask - Security
user_datastore = SQLAlchemyUserDatastore(db, User, Role)
security = Security(app, user_datastore)
# Create a user to test with
@app.before_first_request
def create_user():

   db.create_all()

user_datastore.create_user(email = 'matt@nobien.net', password = 'password')

db.session.commit()
# Views
@app.route('/')
@login_required
def home():

   return render_template('index.html')
if __name__ == '__main__':

   app.run()

$ mkvirtualenv <your-app-name>
   $ pip install flask-security sqlalchemy

$ mkvirtualenv <your-app-name>
   $ pip install flask-security flask-mongoengine

from flask
import Flask, render_template
from flask_mongoengine
import MongoEngine
from flask_security
import Security, MongoEngineUserDatastore, \

UserMixin, RoleMixin, login_required
# Create app
app = Flask(__name__)
app.config['DEBUG'] = True
app.config['SECRET_KEY'] = 'super-secret'
# MongoDB Config
app.config['MONGODB_DB'] = 'mydatabase'
app.config['MONGODB_HOST'] = 'localhost'
app.config['MONGODB_PORT'] = 27017
# Create database connection object
db = MongoEngine(app)
class Role(db.Document, RoleMixin):

   name = db.StringField(max_length = 80, unique = True)

description = db.StringField(max_length = 255)
class User(db.Document, UserMixin):

   email = db.StringField(max_length = 255)

password = db.StringField(max_length = 255)

active = db.BooleanField(
   default = True)

confirmed_at = db.DateTimeField()

roles = db.ListField(db.ReferenceField(Role),
   default = [])
# Setup Flask - Security
user_datastore = MongoEngineUserDatastore(db, User, Role)
security = Security(app, user_datastore)
# Create a user to test with
@app.before_first_request
def create_user():

   user_datastore.create_user(email = 'matt@nobien.net', password = 'password')
# Views
@app.route('/')
@login_required
def home():

   return render_template('index.html')
if __name__ == '__main__':

   app.run()

$ mkvirtualenv <your-app-name>
   $ pip install flask-security flask-peewee

pipenv install flask - mail

#~/movie-bag/app.py

from flask
import Flask
from flask_bcrypt
import Bcrypt
from flask_jwt_extended
import JWTManager
   +
   from flask_mail
import Mail

   ...

   api = Api(app, errors = errors)
bcrypt = Bcrypt(app)
jwt = JWTManager(app) +
   mail = Mail(app)

app.config['MONGODB_SETTINGS'] = {
      'host': 'mongodb://localhost/movie-bag'
         ...

mkdir services
cd services
touch mail_service.py

#~/movie-bag/services / mail_service.py

from threading
import Thread
from flask_mail
import Message

from app
import app
from app
import mail

def send_async_email(app, msg):
   with app.app_context():
   try:
   mail.send(msg)
except ConnectionRefusedError:
   raise InternalServerError("[MAIL SERVER] not working")

def send_email(subject, sender, recipients, text_body, html_body):
   msg = Message(subject, sender = sender, recipients = recipients)
msg.body = text_body
msg.html = html_body
Thread(target = send_async_email, args = (app, msg)).start()

#~/movie-bag/resources / errors.py

class UnauthorizedError(Exception):
   pass

   +
   class EmailDoesnotExistsError(Exception):
   +pass +
   +class BadTokenError(Exception):
   +pass +
   errors = {
      "InternalServerError": {
         "message": "Something went wrong",
         @ @ - 54,
         5 + 60,
         13 @ @ errors = {
            "UnauthorizedError": {
               "message": "Invalid username or password",
               "status": 401 +
            },
            +"EmailDoesnotExistsError": {
               +"message": "Couldn't find the user with given email address",
               +"status": 400 +
            },
            +"BadTokenError": {
               +"message": "Invalid token",
               +"status": 403
            }
         }

Hi Miguel,
Thanks
for your post.
I have several questions.
1. Why not to
catch specific exception in verify_reset_password_token method of User model, e.g.jwt.ExpiredSignatureError and so on ?
   2. If we talk about real system is it worth to use some kind of Task Queue to process mail - related tasks ?

@Vitalii: 1. Because really any error that is triggered during token validation leads to rejecting the user.You can look in the jwt package
for all the possible exceptions that can be raised and only
catch those
if that makes more sense to you.

If you have high - volume of emails, using a separate process or pool or processes to send those emails might be a good idea.For low volume a background thread is fine.

Wonderful!
   What about logging in from other social sites like facebook, twitter, or linkedin ? Is it recommended
for serious apps that need high security like a banking app ?

@James: I have written about FB / Twitter authentication in a Flask app here: https: //blog.miguelgrinberg.com/post/oauth-authentication-with-flask. I would not use this for a high security application where you need to proof your identity (i.e. banking), but it is a secure method to authenticate users regardless.

Miguel, what 's the explanation for the "_external" parameter passed to render_template() to produce the email bodies...?

Dear Miguel,
thank you very much
for this excellent tutorial!I was looking
for something like this
for a long time...
   Regarding the load times of the password reset feature, I am still noticing a small difference in speed after implementing the asynchronous send - method.This opens up our young web app
for timing attacks, even though we only
return a single message saying "check your email".
An easy suggestion of how to fix this would be to start a new thread in both cases, when the email exists among our users and when it doesn 't and let the thread figure out whether to send the mail or not.