網站後端.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,更加準確的引導用戶激活帳戶