网站后端.Flask.实战-社交博客开发-邮件确认?

时间 2019-11-13

原文原文链接

生成确认令牌

1.为防止泛滥注册,有时须要邮箱确认,用户注册后,当即发送一封确认邮件,新帐户先被标记为未确认状态,帐户确认过程当中,每每会要求用户点击一个包含确认token的特殊的URL连接html

2.确认邮件中经常使用相似/auth/confirm/<id>形式的url,id为数据库分配给用户的id,用户点击此url后发送GET请求到确认的视图函数,视图函数判断id有效性,若是有效经过id找到对应的用户对象改变其状态,但此方法不安全,只要攻击者知道url连接则能够激活任意用户python

3.itsdangerous包不只能够签名实现会话加密保护用户会话防止篡改,还能够经过TimedJSONWebSignatureSerializer(secret_key, expires_in=3600)类生成包含用户id的指定时间有效安全令牌,很是适合token使用数据库

FlaskWeb/app/models.pyflask

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
#
# Authors: limanman
# OsChina: http://my.oschina.net/pydevops/
# Purpose:
#
"""
from . import db, loginmanager
from flask import current_app
from flask_login import UserMixin
from itsdangerous import TimedJSONWebSignatureSerializer as Serializer
from werkzeug.security import generate_password_hash, check_password_hash


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

class Role(db.Model):
    __tablename__ = 'roles'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True, nullable=False, index=True)
    users = db.relationship('User', backref='role', lazy='dynamic')

class User(UserMixin, db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(64), unique=True, nullable=False, index=True)
    username = db.Column(db.String(64), unique=True, nullable=False, index=True)
    password_hash = db.Column(db.String(128), nullable=False)
    is_confirmed = db.Column(db.Boolean, default=False)
    role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))

    @property
    def password(self):
        raise AttributeError(u'password 不容许读取.')

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

    def generate_confirm_token(self, expires_in=3600):
        s = Serializer(current_app.config['SECRET_KEY'], expires_in=expires_in)
        data = s.dumps({'confirm_id': self.id})
        return data

    def verify_confirm_token(self, token):
        s = Serializer(current_app.config['SECRET_KEY'])
        try:
            data = s.loads(token)
        except BaseException, e:
            return False
        if data.get('confirm_id') != self.id:
            return False
        self.is_confirmed = True
        db.session.add(self)
        db.session.commit()
        return True

说明:在用户模型中加入is_confirmed字段,自定义的generate_confirm_token是经过TimedJSONWebSignatureSerializer计算带时间序列的JsonWeb签名,而后用此签名dumps生成加密token(加密的是用户对象的id字段),一样能够用此签名loads解密token来验证token中的用户id,这样即便攻击者知道如何生成签名令牌,也没法确认别人的帐户浏览器

发送确认邮件

1.注册用户在将用户添加入数据库,而后重定向到/index,在重定向以前须要发送确认邮件安全

FlaskWeb/app/auth/views.pysession

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
#
# Authors: limanman
# OsChina: http://my.oschina.net/pydevops/
# Purpose:
#
"""
from .. import db
from . import auth
from ..models import User
from ..email import send_mail
from .forms import LoginForm, RegisterForm
from flask import render_template, flash, url_for, redirect
from flask_login import login_required, fresh_login_required, login_user, logout_user, current_user

@auth.route('/')
@fresh_login_required
def index():
    pass

@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 and user.verify_password(form.password.data):
            flash(u'已成功登陆', 'success')
            login_user(user, form.remeber_me.data)
            return redirect(url_for('main.index'))
        flash(u'用户名或密码错误', 'danger')
        return redirect(url_for('auth.login'))
    return render_template('auth/login.html', form=form)

@auth.route('/logout', methods=['GET', 'POST'])
@fresh_login_required
def logout():
    logout_user()
    flash(u'已退出登陆', 'success')
    return redirect(url_for('main.index'))

@auth.route('/register', methods=['GET', 'POST'])
def register():
    form = RegisterForm()
    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()
        token = user.generate_confirm_token(expires_in=3600)
        flash(u'确认已发送至你的邮箱,点击激活邮件', 'success')
        send_mail(form.email.data,
                  u'Flasky - 注册确认邮件',
                  'auth/email/confirm',
                  user=user, token=token)
        return redirect(url_for('main.index'))
    return render_template('auth/register.html', form=form)

说明:经过配置可实如今请求结尾自动提交数据库变化,这里须要添加db.session.commit(),因为确认token中须要用到id,因此必须在请求结束以前生成token以前提交数据库变化app

FlaskWeb/app/templates/auth/email/confirm.txt函数

尊敬的 {{ user.username }}:

您好!欢迎您使用 Flasky!

您只需点击下方的连接，便可验证您的电子邮件地址并完成
注册：
{{ url_for('auth.confirm', token=token, _external=True ) }}

若是以上连接无效，请复制此网址，并将其粘贴到
新的浏览器窗口中.

若是您对账户存在疑问.则能够随时访问 Flasky 账户
帮助中心：https://support.flasky.com/accounts/

感谢您使用 Flasky 账户,祝您使用愉快!
这只是一封公告邮件.咱们并不监控或回答对此邮件的回复.

说明:认证蓝图邮件模版放在FlaskWeb/app/templates/auth/email文件夹中,这里为了邮件能够渲染纯文本和富文本,分别加入txt/html文件,url_for利用auth.confirm视图生成一个带有/auth/confirm/<token>的路径,因为设置_external=True,则url地址为完整的urlui

FlaskWeb/app/auth/views.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
#
# Authors: limanman
# OsChina: http://my.oschina.net/pydevops/
# Purpose:
#
"""
from .. import db
from . import auth
from ..models import User
from ..email import send_mail
from .forms import LoginForm, RegisterForm
from flask import render_template, flash, url_for, redirect
from flask_login import login_required, fresh_login_required, login_user, logout_user, current_user

@auth.route('/')
@fresh_login_required
def index():
    pass

@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 and user.verify_password(form.password.data):
            flash(u'已成功登陆', 'success')
            login_user(user, form.remeber_me.data)
            return redirect(url_for('main.index'))
        flash(u'用户名或密码错误', 'danger')
        return redirect(url_for('auth.login'))
    return render_template('auth/login.html', form=form)

@auth.route('/logout', methods=['GET', 'POST'])
@fresh_login_required
def logout():
    logout_user()
    flash(u'已退出登陆', 'success')
    return redirect(url_for('main.index'))

@auth.route('/register', methods=['GET', 'POST'])
def register():
    form = RegisterForm()
    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()
        token = user.generate_confirm_token(expires_in=3600)
        flash(u'确认已发送至你的邮箱,点击激活邮件', 'success')
        send_mail(form.email.data,
                  u'Flasky - 注册确认邮件',
                  'auth/email/confirm',
                  user=user, token=token)
        return redirect(url_for('main.index'))
    return render_template('auth/register.html', form=form)

@auth.route('/confirm/<token>')
@login_required
def confirm(token):
    if current_user.verify_confirm_token(token):
        flash(u'邮箱验证经过', 'success')
        return redirect(url_for('main.index'))
    flash(u'确认链接已失效', 'danger')
    return redirect(url_for('main.index'))

说明:在confirm视图函数上附加了login_required,会自动加载session['id']还原用户对象,若是还原失败自动重定向到loginmanager.login_view登陆视图,不然开始验证current_user当前用户对象token是否正确,正确就改变数据库is_comfirmed值为1,邮件确认状态

2.程序应容许未确认的用户登陆但只能显示一个页面,这个页面要求用户在获取权限以前先确认帐户

FlaskWeb/app/auth/views.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
#
# Authors: limanman
# OsChina: http://my.oschina.net/pydevops/
# Purpose:
#
"""
from .. import db
from . import auth
from ..models import User
from ..email import send_mail
from .forms import LoginForm, RegisterForm
from flask import render_template, flash, url_for, redirect, request
from flask_login import login_required, fresh_login_required, login_user, logout_user, current_user

@auth.before_app_request
def before_app_request():
    if current_user.is_authenticated \
            and not current_user.is_confirmed \
            and request.endpoint \
            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.is_confirmed:
        return redirect(url_for('main.index'))
    return render_template('auth/unconfirmed.html')

@auth.route('/')
@fresh_login_required
def index():
    pass

@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 and user.verify_password(form.password.data):
            flash(u'已成功登陆', 'success')
            login_user(user, form.remeber_me.data)
            return redirect(url_for('main.index'))
        flash(u'用户名或密码错误', 'danger')
        return redirect(url_for('auth.login'))
    return render_template('auth/login.html', form=form)

@auth.route('/logout', methods=['GET', 'POST'])
@fresh_login_required
def logout():
    logout_user()
    flash(u'已退出登陆', 'success')
    return redirect(url_for('main.index'))

@auth.route('/register', methods=['GET', 'POST'])
def register():
    form = RegisterForm()
    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()
        token = user.generate_confirm_token(expires_in=3600)
        flash(u'确认已发送至你的邮箱,点击激活邮件', 'success')
        send_mail(form.email.data,
                  u'Flasky - 注册确认邮件',
                  'auth/email/confirm',
                  user=user, token=token)
        return redirect(url_for('main.index'))
    return render_template('auth/register.html', form=form)

@auth.route('/confirm')
@login_required
def resend_confirmation():
    token = current_user.generate_confirm_token()
    send_mail(current_user.email,
              u'Flasky - 注册确认邮件',
              'auth/email/confirm',
              user=current_user, token=token)
    flash(u'一封新的确认邮件已经发送至你的邮箱')
    return redirect(url_for('main.index'))

@auth.route('/confirm/<token>')
@login_required
def confirm(token):
    if current_user.verify_confirm_token(token):
        flash(u'邮箱验证经过', 'success')
        return redirect(url_for('main.index'))
    flash(u'确认链接已失效', 'danger')
    return redirect(url_for('main.index'))

说明:before_request钩子只能应用到当前蓝图请求上,若是要设置全局请求钩子则须要使用before_app_request,要想实现引导未确认用户自助激活状态,则须要同时知足用户已登陆(current_user.is_authenticated),用户的帐户还未确认(current_user.is_confirmed ),请求的端点request.endpoint不在认证蓝本中

注意:有可能会出现request.endpoint为None的时候,因此强烈建议多加一个判断request.endpoint是否为None,更加准确的引导用户激活帐户