第8章：用户认证-Flask Web 开发(第2版)

# 示例 8-1. app/models.py: User 模型的密码哈希
from werkzeug.security import generate_password_hash, check_password_hash

class User(db.Model):
    # ...
    password_hash = db.Column(db.String(128))

    @property
    def password(self):
        raise AttributeError('密码是个只写属性')

    @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)

(venv) $ flask shell
>>> u = User()
>>> u.password = 'cat'
>>> u.password
Traceback (most recent call last):
  File "<console>", line 1, in <module>
  File "/home/flask/flasky/app/models.py", line 24, in password
    raise AttributeError('密码是个只写属性')
AttributeError: 密码是个只写属性
>>> u.password_hash
'pbkdf2:sha256:50000$moHwFH1B$ef1574909f9c549285e8547cad181c5e0213cfa44a4aba4349fa830aa1fd227f'
>>> u.verify_password('cat')
True
>>> u.verify_password('dog')
False
>>> u2 = User()
>>> u2.password = 'cat'
>>> u2.password_hash
'pbkdf2:sha256:50000$Pfz0m0KU$27be930b7f0e0119d38e8d8a62f7f5e75c0a7db61ae16709bcaa6cfd60c44b74

# 示例 8-2. tests/test_user_model.py: 测试密码安全
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='cat')
        self.assertTrue(u.password_hash != u2.password_hash)

(venv) $ flask test
test_app_exists (test_basics.BasicsTestCase) ... ok
test_app_is_testing (test_basics.BasicsTestCase) ... ok
test_no_password_getter (test_user_model.UserModelTestCase) ... ok
test_password_salts_are_random (test_user_model.UserModelTestCase) ... ok
test_password_setter (test_user_model.UserModelTestCase) ... ok
test_password_verification (test_user_model.UserModelTestCase) ... ok

.----------------------------------------------------------------------
Ran 6 tests in 0.379s

OK

# 示例 8-3. app/auth/__init__.py: 创建用户认证蓝图
from flask import Blueprint

auth = Blueprint('auth', __name__)

from . import views

# 示例 8-4. app/auth/views.py: 用户认证蓝图的路由和视图函数
from flask import render_template
from . import auth

@auth.route('/login')
def login():
    return render_template('auth/login.html')

# 示例 8-5. app/__init__.py: 注册用户认证蓝图
def create_app(config_name):
    # ...
    from .auth import auth as auth_blueprint
    app.register_blueprint(auth_blueprint, url_prefix='/auth')

    return app

(venv) $ pip install flask-login

# 示例 8-6. app/models.py: 更新User模型，以支持用户登录
from flask_login import UserMixin

class User(UserMixin, db.Model):
    __tablename__ = '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'))

# 示例 8-7. app/__init__.py: Flask-Login 初始化
from flask_login import LoginManager

login_manager = LoginManager()
login_manager.login_view = 'auth.login'

def create_app(config_name):
    # ...
    login_manager.init_app(app)
    # ...

# 示例 8-8. app/models.py: 用户加载函数
from . import login_manager

@login_manager.user_loader
def load_user(user_id):
    return User.query.get(int(user_id))

from flask_login import login_required

@app.route('/secret')
@login_required
def secret():
    return '只允许已登录用户访问!'

# 示例 8-9. app/auth/forms.py: 登录表单

from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, BooleanField, SubmitField
from wtforms.validators import DataRequired, Length, Email

class LoginForm(FlaskForm):
    email = StringField('Email', validators=[DataRequired(), Length(1, 64),
                                             Email()])
    password = PasswordField('密码', validators=[DataRequired()])
    remember_me = BooleanField('让我保持登录')
    submit = SubmitField('登录')

<!-- 示例 8-10. app/templates/base.html: 导航栏中的“登录”和“注销”两个链接-->
<ul class="nav navbar-nav navbar-right">
    {% if current_user.is_authenticated %}
    <li><a href="{{ url_for('auth.logout') }}">注销</a></li>
    {% else %}
    <li><a href="{{ url_for('auth.login') }}">登录</a></li>
    {% endif %}
</ul>

# 示例 8-11. app/auth/views.py: 用户登录路由
from flask import render_template, redirect, request, url_for, flash
from flask_login import login_user
from . import auth
from ..models import User
from .forms 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(user, form.remember_me.data)
            next = request.args.get('next')
            if next is None or not next.startswith('/'):
                next = url_for('main.index')
            return redirect(next)
        flash('错误的用户名或密码.')
    return render_template('auth/login.html', form=form)

<!-- 示例 8-12. app/templates/auth/login.html: 登录表单的模板 -->
{% extends "base.html" %}
{% import "bootstrap/wtf.html" as wtf %}

{% block title %}Flasky - 登录{% endblock %}

{% block page_content %}
<div class="page-header">
    <h1>登录</h1>
</div>
<div class="col-md-4">
    {{ wtf.quick_form(form) }}
</div>
{% endblock %}

# 示例 8-13. app/auth/views.py: 注销登录路由
from flask_login import logout_user, login_required

@auth.route('/logout')
@login_required
def logout():
    logout_user()
    flash('你已成功登出.')
    return redirect(url_for('main.index'))

<!-- 示例 8-14. app/templates/index.html: 问候已登录用户 -->
你好,
{% if current_user.is_authenticated %}
    {{ current_user.username }}
{% else %}
    陌生人
{% endif %}!

(venv) $ $ flask shell
>>> u = User(email='[email protected]', username='john', password='cat')
>>> db.session.add(u)
>>> db.session.commit()

# 示例 8-15. app/auth/forms.py: 用户注册表单
from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, BooleanField, SubmitField
from wtforms.validators import DataRequired, Length, Email, Regexp, EqualTo
from wtforms import ValidationError
from ..models import User

class RegistrationForm(FlaskForm):
    email = StringField('Email', validators=[DataRequired(), Length(1, 64),
                                             Email()])
    username = StringField('用户名', validators=[
        DataRequired(), Length(1, 64),
        Regexp('^[A-Za-z][A-Za-z0-9_.]*$', 0,
               '用户名只能包含字母、数字、点或下划线')])
    password = PasswordField('密码', validators=[
        DataRequired(), EqualTo('password2', message='两次输入的密码不一致')])
    password2 = PasswordField('重复密码', validators=[DataRequired()])
    submit = SubmitField('注册')

    def validate_email(self, field):
        if User.query.filter_by(email=field.data).first():
            raise ValidationError('Email已经被注册了')

    def validate_username(self, field):
        if User.query.filter_by(username=field.data).first():
            raise ValidationError('用户名已经被注册了')

<!-- 示例 8-16. app/templates/auth/login.html: 链接到注册页面 -->
<p>
    新用户?
    <a href="{{ url_for('auth.register') }}">
        点击这里进行注册
    </a>
</p>

# 示例 8-17. app/auth/views.py: 用户注册的路由
@auth.route('/register', methods=['GET', 'POST'])
def register():
    form = RegistrationForm()
    if form.validate_on_submit():
        user = User(email=form.email.data,
                    username=form.username.data,
                    password=form.password.data)
        db.session.add(user)
        db.session.commit()
        flash('注册成功。你现在可以登录了')
        return redirect(url_for('auth.login'))
    return render_template('auth/register.html', form=form)

(venv) $ flask shell
>>> from itsdangerous import TimedJSONWebSignatureSerializer as Serializer
>>> s = Serializer(app.config['SECRET_KEY'], expires_in=3600)
>>> token = s.dumps({ 'confirm': 23 })
>>> token
'eyJhbGciOiJIUzI1NiIsImV4cCI6MTM4MTcxODU1OCwiaWF0IjoxMzgxNzE0OTU4fQ.ey ...'
>>> data = s.loads(token)
>>> data
{'confirm': 23}

# Example 8-18. app/models.py: 用户账号确认
from itsdangerous import TimedJSONWebSignatureSerializer as Serializer
from flask import current_app
from . import db

class User(UserMixin, db.Model):
    # ...
    confirmed = db.Column(db.Boolean, default=False)

    def generate_confirmation_token(self, expiration=3600):
        s = Serializer(current_app.config['SECRET_KEY'], expiration)
        return s.dumps({'confirm': self.id}).decode('utf-8')

    def confirm(self, token):
        s = Serializer(current_app.config['SECRET_KEY'])
        try:
            data = s.loads(token.encode('utf-8'))
        except:
            return False
        if data.get('confirm') != self.id:
            return False
        self.confirmed = True
        db.session.add(self)
        return True

# 示例 8-19. app/auth/views.py: 含有发送确认邮件的用户注册路由
from ..email import send_email

@auth.route('/register', methods=['GET', 'POST'])
def register():
    form = RegistrationForm()
    if form.validate_on_submit():
        # ...
        db.session.add(user)
        db.session.commit()
        token = user.generate_confirmation_token()
        send_email(user.email, '确认你的账户',
                   'auth/email/confirm', user=user, token=token)
        flash('一封确认邮件已经发送到你的邮箱中。')
        return redirect(url_for('main.index'))
    return render_template('auth/register.html', form=form)

# 示例 8-20. app/templates/auth/email/confirm.txt: 确认邮件的正文
亲爱的 {{ user.username }},

欢迎来到 Flasky!

请点击下面的链接来确认你的账号:

{{ url_for('auth.confirm', token=token, _external=True) }}

此致,

Flasky 团队

提示: 请不要直接回复本邮件.

# 示例 8-21. app/auth/views.py: 确认用户账号
from flask_login import current_user
@auth.route('/confirm/<token>')
@login_required
def confirm(token):
    if current_user.confirmed:
        return redirect(url_for('main.index'))
    if current_user.confirm(token):
        db.session.commit()
        flash('你的账号确认成功，谢谢!')
    else:
        flash('确认链接无效或已过期.')
    return redirect(url_for('main.index'))

# 示例 8-22. app/auth/views.py: 通过 before_app_request 处理器来过滤未确认用户
@auth.before_app_request
def before_request():
    if current_user.is_authenticated \
            and not current_user.confirmed \
            and request.blueprint != '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.html')

# 示例 8-23. app/auth/views.py: 重新发送用户确认邮件
@auth.route('/confirm')
@login_required
def resend_confirmation():
    token = current_user.generate_confirmation_token()
    send_email(current_user.email, '确认你的账号',
               'auth/email/confirm', user=current_user, token=token)
    flash('一封新的确认邮件已发送到你的邮箱.')
    return redirect(url_for('main.index'))