本篇博客介绍使用HTTPTokenAuth来实现用户认证。
1.相关库安装
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，发现已经能正常请求了：
