本篇博客介绍使用HTTPTokenAuth来实现用户认证。

pip install flask-httpauth

2.相关代码

在 extensions.py 中初始化 HTTPTokenAuth 的扩展实例，并添加相关的认证代码和自定义错误返回码：

from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_bootstrap import Bootstrap
from flask_login import LoginManager
from flask_httpauth import HTTPTokenAuth
from itsdangerous import TimedJSONWebSignatureSerializer as Serializer, BadSignature, SignatureExpired
from flask import current_app, g, make_response, jsonify


db = SQLAlchemy()
migrate = Migrate()
bootstrap = Bootstrap()
login_manager = LoginManager()
auth = HTTPTokenAuth(scheme='JWT')


@login_manager.user_loader
def load_user(user_id):
    from myblog.models import Admin
    user = Admin.query.get(int(user_id))
    return user

login_manager.login_view = 'admin.login'
login_manager.login_message = '你必须登陆后才能访问该页面！'
login_manager.login_message_category = "info"


@auth.verify_token
def verify_token(token):
    g.token_error = False
    g.token_timeout = False
    s = Serializer(current_app.config['SECRET_KEY'])
    try:
        data = s.loads(token)
    except BadSignature:
        g.token_error = True
        return False
    except SignatureExpired:
        g.token_timeout = True
        return False
    return True

@auth.error_handler
def unauthorized():
    if g.token_error:
        return make_response(jsonify({'code':402,'error': 'Token Error'}), 402)

    if g.token_timeout:
        return make_response(jsonify({'code':401,'error': 'Token Expired'}), 401)

在 models.py 中的 Admin 类添加生成 token 的方法：

from datetime import datetime
from werkzeug.security import generate_password_hash, check_password_hash
from flask_login import UserMixin

from myblog.extensions import db
from itsdangerous import TimedJSONWebSignatureSerializer as Serializer
from flask import current_app


class Admin(db.Model, UserMixin):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(20))
    password_hash = db.Column(db.String(128))

    def set_password(self, password):
        self.password_hash = generate_password_hash(password)

    def validate_password(self, password):
        return check_password_hash(self.password_hash, password)

    def create_token(self, user_id):
        s = Serializer(current_app.config['SECRET_KEY'], expires_in=current_app.config['TOKEN_EXPIRATION'])
        token = s.dumps({"id": user_id}).decode('ascii')
        return token

在 config.py 中新增一项 token 过期时间的配置：

class Config(object):
    SQLALCHEMY_DATABASE_URI = 'mysql+cymysql://root:root@localhost:3306/myflask?charset=utf8'
    SQLALCHEMY_TRACK_MODIFICATIONS = True

    SECRET_KEY = "you will never known it."
    TOKEN_EXPIRATION = 10800

在 admin.py 中新增一个auth_login视图，并将 add 视图改为用 token 验证：

from flask import request, Blueprint, render_template, redirect, url_for, flash, jsonify
from flask_login import login_user, login_required, current_user, logout_user

from myblog.models import Article, Category, Comment, Admin
from myblog.forms import LoginForm, AddForm
from myblog.extensions import db, auth


admin_bp = Blueprint('admin', __name__)


@admin_bp.route('/auth_login', methods=['POST'])
def auth_login():
    username = request.form['username']
    password = request.form['password']
    admin = Admin.query.first()
    if username == admin.username and admin.validate_password(password):
        z_token = admin.create_token(admin.id)
 
    return jsonify(token=z_token)


@admin_bp.route('/add', methods=['POST', 'GET'])
@auth.login_required  
def add():
    form = AddForm()
    if form.validate_on_submit():
        title = form.title.data
        body = form.body.data
        category = form.category.data

        article = Article(title=title, body=body, category_id=category)
        db.session.add(article)
        db.session.commit()
        flash('新建博客成功', category='info')
        return redirect(url_for('admin.edit'))

    return render_template('admin/add.html', form=form)

3.相关验证

我们发现在登录的情况下，已经无法打开 add 视图：

我们在 postman 中模拟请求auth_login，拿到token：

然后在 postman 中添加 Authorization 请求头（注意前面要加上"JWT "），模拟请求add，发现已经能正常请求了：