重点说明:根据作者的经验,从第八章开始,若只把书上示例的代码敲上去,是很难重复实现书中效果的,基本上都会出错,原因有(1)书中省略了许多代码(2)由于扩展版本的升级,书中部分代码已不再适用。所以,强烈强烈强烈建议边看书边对照着Git上对应标签的版本阅读源代码,以免遇到巨坑又不知道哪里出错又无法debug。此时也提醒一下自己,要把代码实现一遍后再来做笔记,以免自己坑自己白费功夫。

8.1 Flask的认证扩展

  1. Werkzeug:计算密码的散列值并进行核对(将密码生成散列值,并验证密码是否正确)。
  2. Flask-Login:管理已登录用户的用户会话。
  3. itsdangerous:生成并核对加密安全令牌(例如注册账号验证邮箱时生成验证链接时用)。

8.2 密码的安全性

若想保证存储在数据库中的用户密码的安全性,那么就不能存储密码本身,而要存储密码的散列值。当需要核对验证密码是否正确时,核对输入的密码生成的散列值是否与数据库中存储的散列值一致即可。因为计算散列值的函数是可复现的:只要输入一样,结果就一样。

使用Werkzeug实现密码散列

Werkzeug中的security模块可以实现散列值的计算。这一功能需要用到两个函数(generate_password_hash()check_password_hash()),它们分别用在注册用户阶段和验证用户阶段。

  • generate_password_hash(password, method=pbkdf2:sha1, salt_length=8):这个函数将原始密码作为输入,以字符串形式输出密码的散列值。methodsalt_length的默认值可满足大多数需求。

  • check_password_hash(hash, password):这个函数的参数分别是从数据库中读取的密码散列值,和用户输入的密码。密码正确则返回True

1. 使app/models.py中的User模型支持密码散列:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
from werkzeug.security import generate_password_hash, check_password_hash
# ...
class User(db.Models):
# 为User模型增加password_hash字段
password_hash = db.Column(db.String(128))
# 当试图读取password的值时,返回错误
@property
def password(self):
raise AttributeError('password is not a readable attribute')
# 计算密码散列值并赋值给passwo_hash
@password.setter
def password(self, password):
self.password_hash = generate_password_hash(password)
# 验证密码是否正确
def verify_password(self, password):
return check_password_hash(self.password_hash, password)

注意:User模型中的password属性为只写,不能读取。当给password赋值时,会调用generate_password_hash()函数计算散列值,并赋值给password_hash字段。可见下例。

2. 在shell中测试:

1
2
3
4
5
6
7
8
9
10
11
12
13
(venv) $ python manage.py shell
>>> u = User()
>>> u.password = 'cat' # 根据密码`'cat'`计算散列值并赋值给password_hash字段
>>> u.password_hash
'pbkdf2:sha1:1000$duxMk......928bed'
>>> u.verify_password('cat')
True
>>> u.verify_password('dog')
False
>>> u2 = User()
>>> u2.password = 'cat'
>>> u2.password_hash
'pbkdf2:sha1:1000$UjvnGe......75ee89'

注意:即使用户u和用户u2都使用了相同的密码,但他们的密码散列值是不一样的。

3. tests/test_user_model.py支持密码散列测试:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
import unittest
from app.models import User
class UserModelTestCase(unittest.TestCase):
def test_password_setter(self):
u = User(password = 'cat')
self.assertTrue(u.password_hash is not None)
def test_no_password_getter(self):
u = User(password='cat')
with self.assertRaises(AttributeError):
u.password
def test_password_verification(self):
u = User(password='cat')
self.assertTrue(u.verify_password('cat'))
self.assertFalse(u.verify_password('dog')
def test_password_salts_are_random(self):
u = User(password='cat')
u2 = User(password='dog')
self.assertTrue(u.passwors_hahs != u2.password_hash)

8.3 创建认证蓝本

将与用户认证系统相关的路由定义在auth蓝本中。

1. 在app/auth/__init__.py中创建蓝本:

1
2
3
4
5
from flask import Blueprint
auth = Blueprint('auth', __name__)
from app.auth import views

2. 在app/auth/views.py中定义蓝本中的路由和视图函数:

1
2
3
4
5
6
from flask import render_template
from app.auth import auth
@auth.route('/login')
def login():
return render_template('auth/login.html')

注意:如果配置了多个模板文件夹,render_tempalte()函数首先会搜索程序配置的模板文件夹,然后再搜索蓝本配置的模板文件夹。

3. 在app/__init__.py中注册蓝本:

1
2
3
4
5
6
7
8
# ...
def create_app(config_name):
# ...
from app.auth import auth as auth_blueprint
app.register_blueprint(auth_blueprint, url_prefix='/auth')
return app

注意:注册蓝本时,url_prefix参数是可选参数,使用这个参数后,蓝本中定义的理由都会加上指定的前缀。如该例中,/login路由会注册成/auth/login,完整的URL就变成http://127.0.0.1:5000/auth/login

8.4 使用Flask-Login认证用户

用户登录程序后,他们的认证状态要被记录下来,这样浏览不同的页面时才能记住这个状态。

8.4.1 准备用于登录的用户模型

要想使用Flask-Login扩展,程序的User模型必须实现以下几个方法:
表8-1 Flask-Login要求实现的用户方法
|方法 | 说明 |
|———|———-|
|is_authenticated | 如果用户已经登录,必须返回True,否则返回False|
|is_active | 如果允许用户登录,必须返回True,否则返回False。如果禁用账户,可以返回False|
|is_anonymous | 对普通用户必须返回False|
|get_id() | 必须返回用户的唯一标识符,使用Unicode编码字符串|

Flask-Login提供了一个UserMixin类,其中包含这些方法的默认实现,且能满足大多数需求。

1. 在app/models.py中修改User模型,支持用户登录:

1
2
3
4
5
6
7
8
9
10
11
from flask_login import UserMixin
# ...
# 继承UserMixin类
class User(UserMixin, db.models):
__tbalename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
email = db.Column(db.String(64), unique=True, index=True)
username = db.Column(db.String(64), unique=True, index=True)
password_hash = db.Column(db.String(128))
role_id = db.Column(db.Integer, db.ForeignKey('roles.id')

2. 在app/__init__.py工厂函数中初始化Flask-Login:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from flask_login import LoginManager
# ...
login_manager = LoginManager()
login_manager.session_protection = 'strong'
login_manager.login_view = 'auth.login'
def create_app(config_name):
# ...
login_manager.init_app(app)
# ...
return app
  • LoginManager对象的session_protection属性可以设为None'basic''strong',以提供不同安全等级防止用户会话遭篡改。设为'strong'时,Flask-Login会记录客户端IP地址和浏览器的用户代理信息,如果发现异动,就登出用户。
  • login_view属性设置登录页面的端点。

3. Flask-Login要求程序实现一个回调函数,使用指定的标识符加载用户。在app/models.py中定义加载用户的回调函数:

1
2
3
4
5
6
from app import login_manager
# ...
@login_manager.user_loader
def load_user(user_id):
return User.query.get(int(user_id))

加载用户的回调函数接收以Unicode字符串形式表示的用户标识符。如果能找到用户,则返回用户对象,否则返回None。

8.4.2 保护路由

1. 可以在路由中使用Flask-Login提供的login_required修饰器,使得路由只让认证用户访问(即访问被注册的路由时要先登录用户),如:

1
2
3
4
5
6
from flask_login import login_required
@app.route('/secret')
@login_required
def secret():
return 'Only authenticated users are allowed!'

或者见8.6.2 发送确认邮件中的第三点。

8.4.3 定义登录表单

该表单包括一个用于输入电子邮件地址的文本字段、一个密码字段、一个“记住我”复选框、提交按钮。

1. 在app/auth/forms.py中定义登录表单:

1
2
3
4
5
6
7
8
9
from flask_wtf import Form
from wtforms import StringField, PasswordField, BooleanField, SubmitField
from wtforms.validators import Required, Length, Email
class LoginForm(Form):
email = StringField('Email', validators=[Required(), Length(1,64), Email()])
password = PasswordField('Password', validators=[Required()])
remember_me = BooleanField('Keep me logged in')
submit = SubmitField('Log In')

接下来在模板中渲染表单即可。

2. 在app/templates/base.html导航条中使用Jinja2条件语句,根据当前用户的登录状态显示“Sign In”或“Sign Out”:

1
2
3
4
5
6
7
8
9
10
11
<ul class="nav navbar-nav navbar-right">
{% if current_user.is_authenticated %}
<li>
<a href="{{ url_for('auth.logout') }}">Sign Out</a>
</li>
{% else %}
<li>
<a href="{{url_for('auth.login') }}">Sign In</a>
</li>
{% endif %}
</ul>

重点注意:变量current_user是由Flask-Login定义的,而且在视图函数和模板中自动可用。

8.4.4 登录用户

1. 在app/auth/views.py中定义登录路由:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from flask import render_template, redirect, request, url_for, flash
from flask_login import login_user # 导入登录用户函数(由Flask-Login提供)
from app.auth import auth
from app.models import User
from app.auth.froms import LoginForm
@auth.route('/login', methods=['GET', 'POST'])
def login():
form = LoginForm()
if form.validate_on_submit():
user = User.query.filter_by(email=form.email.data).first()
# 判断用户是否存在、密码是否正确
if user is not None and user.verify_password(form.password.data):
# login_user()函数由Flask-Login提供
# 若用户存在以及密码正确,则调用login_user()函数,
# 在用户会话中把用户标记为已登录状态
login_user(user, form.remember_me.data)
# 返回原先的地址或者主页index
return redirect(request.args.get('next') or url_for('main.index')
flash('Invalid usernane or password')
return render_template('auth/login.html', form=form)

重定向的URL有两种可能:(1)是用户访问未授权的URL时会显示登录表单,Flask-Login会把原本的地址保存在request.args字典的next键中,可用get()方法获取。(2)是如果next的值为空,则重定向到主页。

2. 在app/templates/auth/login.htm中渲染登录表单:

1
2
3
4
5
6
7
8
9
10
11
12
{% extends "base.html" %}
{% import "bootstrap/wtf.html" as wtf %}
{% block title %}Flasky - Login{% endblock %}
{% block page_content %}
<div class="page_header">
<h1>Login</h1>
</div>
<div class="col-md-4">
{% wtf.quick_form(form) }}
</div>
{% endblock %}

8.4.5 登出用户

1. 在app/auth/views.py中定义登出路由:

1
2
3
4
5
6
7
8
9
from flask_login import logout_user, login_required
# ...
@auth.route('/logout')
@login_required
def logout():
logout_user()
flash('You have been logged out.')
return redirect(url_for('main.index')

Flask-Login提供的logout_user()函数:删除并重设用户会话。

8.4.6 测试登录(在shell中测试)

1. 在app/templates/index.html中为已登录的用户显示一个欢迎消息:

1
2
3
4
5
6
Hello,
{% if current_user.is_authenticates %}
{{ current_user.username }}
{% else %}
Stranger
{% endif %}!

2. 在shell中注册新用户(因为未创建用户注册功能,所以可在shell创建新用户):

1
2
3
4
(venv)$ python manage.py shell
>>> u = User(email='123456@qq.com', username='john', password='cat')
>>> db.session.add(u)
>>> db.session.commit()

3. 打开登录页面进行登录

登录后即可显示欢迎消息。

8.5 注册新用户

8.5.1 添加用户注册表单

注册页面使用的表单要求用户输入电子邮件地址、用户名和密码。

1. 在app/auth/forms.py中定义用户注册表单:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from falsk_wtf import Form
from wtforms import StringField, PasswordField, BooleanField, SubmitField
from wtforms.validators import Required, Length, Email, Regexp, EqulTo
from wtforms import ValidationError
from app.models import User
class RegistrationForm(Form):
email = StringField('Email', validators=[Required(), Length(1,64), Email()])
username = StringField('Username', validators=[REquired(), Length(1,64),
Regexp('^[A-Za-z][A-Za-z0-9_.]*$', 0, 'Usernames must have only letters, numbers, dots or underscores')])
password = PasswordField('Password', validators=[Required(), EqualTo('password2', message='Password must match')])
password2 = PasswordField('Confirm password', validators=[Required()])
submit = SubmitField('Register')
def validate_email(self, field):
if User.query.filter_by(email=field.data).first():
raise ValidationError('Email already registered')
def validate_username(self, field):
if User.query.filter_by(usernanme=field.data).first():
raise ValidationError('Username already in use')
  • Regexp()验证函数确username字段只包含字母、数字、下划线、点号。该验证函数正则后面的两个参数分别是正则的旗标和验证失败时显示的错误信息。
  • EqualTo()验证函数确保两个密码字段一致。这个验证函数要附属到两个密码字段中的一个,另一个则作为参数传入。
  • 该表单还自定义了两个验证函数,以方法的形式实现。如果表单类中定义了以validate_开头且后面跟着字段名的方法,这个方法就和常规的验证函数一起调用

2. 渲染template/auth/register.html表单:

和之前渲染表单一样,使用wtf.quick_form(form)渲染。

3. 在template/auth/login.html中添加注册页面链接:

1
2
3
4
5
6
7
# ...
<p>
New User?
<a href="{{ url_for('auth.register') }}">
Click here to register
</a>
</p>

8.5.2 注册新用户(定义注册新用户的路由)

1. 在app/auth/views.py中定义用户注册路由:

1
2
3
4
5
6
7
8
9
10
11
12
13
# ...
@auth.route('/register', methods=['GET', 'POST'])
def register():
form = RegistationForm()
if form.validate_on_submit():
user = User(email=form.email.data,
username=form.username.data,
password=form.password.data)
db.session.add(user)
flash('You can login now.')
return redirect(url_for('auth.login'))
return render_template('auth/register.html', form=form)

8.6 确认账户(验证邮箱)

用户验证邮箱之前,新账户先被标记成待确认状态。账户确认过程,一般会要求用户点击一个包含确认令牌的特殊URL链接。

8.6.1 使用itsdangerous生成确认令牌

思路:确认邮件最简单的URL链接是http://www.example.com/auth/confirm/<id>这种形式,其中id是数据库分配给用户的数字id。用户点击链接后,处理这个路由的视图函数将id作为参数进行确认,然后将用户账户状态更新为已确认。

存在问题:不安全。只要用户能判断确认链接的格式,就可以随便指定URL中的id,从而验证确认随意账户。

解决方法:把URL中的id换成将相同信息安全加密后得到的令牌

解决工具:使用itsdangerous包中的TiemdJSONWebSignatureSerializer类提供的dumps()方法和load()方法。

1. 在app/models.py中的User模型中添加验证用户功能:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
from itsdangerous import TimedJSONWebSignatureSerializer as Serializer
from falsk import current_app
from app import db
class User(db.models):
# ...
confirmed = db.Column(db.Boolean, default=False)
# 使用dumps()生成加密令牌
def generate_confirmation_token(self, expiration=3600):
s = Serializer(current_app.config['SECRET_KEY'], expires_in=expiration)
return s.dumps({'confirm': self.id})
# 解码加密令牌并验证原始数据是否与存储中current_app中已登录用户数据一致
def confirm(self, token)
s = Serializer(current_app.config['SECRET_KEY'])
try:
data = s.loads(token) # 解码令牌返回原始数据
except:
return False
# 判断原始数据是否跟已登录用户id一致,防止恶意验证
if data.get('confirm') != self.id:
return False
self.confirmed = True
# 更新confirmed字段
db.session.add(self)
return True
  • TimedJSONWebSignatureSerializer类生成具有国旗时间的JSON Web签名。这个类的构造函数可接受两个参数:密匙(可用Flask中的SECRET_KEY密匙)和expires_in(设置过期时间,单位秒)。
  • dumps()方法为指定的数据生成一个加密签名,然后再对数据和签名进行序列化,生成令牌字符串
  • loads()方法的唯一参数是令牌字符串。这个方法会检验签名和过期时间,如果通过则返回原始数据,否则抛出异常。

8.6.2 发送确认邮件

1. 在app/auth/views.py中使register路由支持邮件发送:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from app.email import send_email
# ...
@auth.route('/register', methods=['GET', 'POST'])
def register():
form = RegistrationForm()
if form.validate_on_submit():
# ...
db.session.add(user)
# 在请求结束前先提交User实例,因为提交后才能得到新用户的id,
# 从而向generate_confirmation_token方法传参得到token
db.session.commit()
token = user.generate_confirmation_token()
send_mail(user.email, 'Confirm Your Account',
'auth/email/confirm', user=user, token=token)
flash('A confirmation email has been sent to you by email.')
return redirect(url_for('main.index'))
return render_template('auth/register.html', form=form)

重点注意:即便通过设置SQLALCHEMY_COMMIT_ON_TEARDOWN=True,程序可以在请求末尾自动提交数据库变化,但是这里也要调用db.session.commit()提交变化。因为提交数据库后才能得到新用户的id值,从而将id值传参给generate_confirmation_token()方法生成令牌。

2. 在app/template/auth/email/confirm.txt中编写确认邮件文本:

1
2
3
4
5
6
7
8
9
10
11
Welcome to Flasky!
To confirm your account please click on the following link:
{{ url_for('auth/confirm', token=token, _external=True) }}
Sincerely,
The Flasky Team
Note: replies to this email address ate not monitored.

注意:默认情况下,url_for()生成的是相对URL,如url_for('auth.confirm', token='abc'),返回的是/auth/confirm/abc',显然,这不是能够在电子邮件中发送正确的URL。所以需要使用_external=True参数,从而生成完整的URL。

3. 在app/auth/views.py中定义确认用户的路由:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from flask_login import current_user
# ...
# 路由中的token将会被作为参数传入视图函数中
@auth.route('/confirm/<token>')
@login_required
def confirm(token):
# 判断user中是否已经确认过
if current_user.confirmed:
return redirect(url_for('main.index'))
# 判断调用confirm()方法返回的是True还是False,从而验证用户
if current_user.confirm(token):
# 提交数据库变化(很重要)
db.session.commit()
flash('You have confirm your account. Thanks!')
else:
flash('The confirmation link is invalid or has expired.')
return redirect(url_for('main.index')

Flask-Login提供的login_required修饰器会保护这个路由:用户点击确认邮件中的链接后,要先登录,然后才能执行这个视图函数。

4. 在app/auth/views.py中在before_app_request处理程序中过滤未确认的账户:

每个程序都可以决定用户在确认账户之前可以有哪些操作,如允许未确认的用户登录,并显示一些内容,但更进一步的看更多内容需要先确认账户。这一步可以使用Flask提供的before_request钩子完成。对于蓝本来说,before_request钩子只能应用到属于蓝本的请求,若想在蓝本中使用针对程序全局请求的钩子,需要使用before_app_request修饰器。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# ...
@auth.before_app_request
def before_request():
if current_user.is_authenticated \ # 判断用户是否已登录
and not current_user.confirmed \ # 判断用户账户是否已验证
and request.endpoint[:5] != 'auth.' \ # 判断请求的端点是否不再认证蓝本中
and request.endpoint != 'static':
return redirect(url_for('auth.unconfirmed'))
@auth.route('/unconfirmed')
def unconfirmed():
if current_user.is_anonymous or current_user.confirmed:
return redirect(url_for('main.index'))
return render_template('auth/unconfirmed.htm')

当满足一下三个条件时,before_app_request处理程序会拦截请求,将请求重定向到/auth/unconfirmed路由,显示一个验证账户相关信息的页面:
(1)用户已登录
(2)用户的账户还未验证
(3)请求的端点(使用request.endpoint获取)不在认证蓝本(/auth)中。

5. 在app/auth/views.py中支持重新发送验证邮件:

1
2
3
4
5
6
7
8
9
10
# ...
@auth.route('/confirm')
@login_required
def resend_confirmation():
token = current_user.generate_confirmation_token()
send_email(current_user.email, 'Confirm Your Account',
'auth/email/confirm', user=current_user, token=token)
flash('A new confirmation email has been sent to you by email.')
return redirect(url_for('main.index'))

这个路由也用login_required保护(要先登录才能执行视图函数),确保程序知道再次发送验证邮件的是哪个用户。

重点注意:凡是修改了数据库模型,为了使新模型能够应用到新程序,要使用Flask-Migrate进行数据库迁移,从而实现更新数据库的效果。

8.7 管理账户

修改密码、重设密码、修改电子邮件地址,具体代码实现可查看git仓库。