使用Flask-Login实现token验证和超时失效使用体会

环境信息

Flask 0.12.2

Flask-Login 0.4.0

itsdangerous 0.24

OS:  Deepin 15.4.1

使用Flask-Login实现token验证和超时失效的原理
       1.Flask-Login采用session机制来实现用户的校验，即当首次登录成功后产生token，浏览器在后续的访问中会将token发送到服务端，服务端根据校验token解析出来的信息来判断后续的访问是否是已经登录成功过的有效用户。

       2.token的计时使用了SimpleCache的自动失效机制。

代码解析
1.model的get_id()返回token

def get_id(self, life_time=None):
print("------get_id,life_time=", life_time)
key = current_app.config.get("SECRET_KEY", "The securet key by C~C!")
s = URLSafeSerializer(key)
browser_id = create_browser_id()
if not life_time:
life_time = current_app.config.get("TOKEN_LIFETIME")
token = s.dumps((self.id, self.user_name, self.password, browser_id, life_time))
return token

2.views中解析并校验token的有效性

@login_manager.user_loader
def user_loader(token):
print("----------user_loader, token=", token)
return load_token(token)

def load_token(token):
# 通过loads()方法来解析浏览器发送过来的token，从而进行初步的验证
key = current_app.config.get("SECRET_KEY", "The securet key by C~C!")
try:
s = URLSafeSerializer(key)
id, name, password, browser_id, life_time = s.loads(token)
except BadData:
print("token had been modified!")
return None

# 判断浏览器信息是否改变
bi = create_browser_id()
if not constant_time_compare(str(bi), str(browser_id)):
print("the user environment had changed, so token has been expired!")
return None

# 校验密码
user = User.query.get(id)
if user:
# 能loads出id，name等信息，说明已经成功登录过，那么cache中就应该有token的缓存
token_cache = simple_cache.get(token)
if not token_cache:  # 此处找不到有2个原因：1.cache中因为超时失效（属于正常情况）；2.cache机制出错（属于异常情况）。
print("the token is not found in cache.")
return None
if str(password) != str(user.password):
print("the password in token is not matched!")
simple_cache.delete(token)
return None
else:
simple_cache.set(token, 1, timeout=life_time)
else:
print('the user is not found, the token is invalid!')
return None
return user

3.在login()中通过调用Login_user()来产生触发session机制

@authentication.route('/login/', methods=['GET', 'POST'])
def login():
if request.method == "POST":
user_name = request.form.get('username')
password = request.form.get('password')
user = User.query.filter_by(user_name=user_name, password=password).first()
next = request.args.get('next')
if not user:
flash("The user name or password is not matched. C~C")
return render_template('login.html')
else:
login_user(user)
g.user = user
# 完成登录后将token存到缓存中并设置过期时间，后面校验时如果缓存中不存在，则报错
life_time = current_app.config.get("TOKEN_LIFETIME")
token = user.get_id(life_time)   # 这里调用get_id()产生token，Flask-Login的token也是调用get_id()产生的，从而保证的二者的一致。
print("-------- login, token=", token)
simple_cache.set(token, 1, life_time)  # 设置cache的失效时间，在login()就可以通过判断cache中是否存在token来知道是否超时了。
return redirect(next or url_for('Archive.archive_list'))
else:
return render_template('login.html')

4.token超时失效的回调

@login_manager.unauthorized_handler
def unauthorized():
return redirect(url_for('Authentication.login'))

方法调用流程说明
A.未登录成功过的情形
    get_id()--->user_loader()--->log_in()
B.已登录成功过的情形
    user_loader()--->对应url的方法

详细代码
utils.py

# coding: utf-8
# author: chenchong

from hashlib import sha512
from flask import request

def get_remote_addr():
"""获取客户端IP地址"""
address = request.headers.get('X-Forwarded-For', request.remote_addr)
if not address:
address = address.encode('utf-8').split(b',')[0].strip()
return address

def create_browser_id():
agent = request.headers.get('User-Agent')
if not agent:
agent = str(agent).encode('utf-8')
base_str = "%s|%s" % (get_remote_addr(), agent)
h = sha512()
h.update(base_str.encode('utf8'))
return h.hexdigest()

model.py

# coding: utf-8
# author: chenchong
from flask import current_app
from itsdangerous import URLSafeSerializer
from app import db
from app.utils import create_browser_id

class User(db.Model):
__tablename__ = "t_user"
id = db.Column(db.Integer, primary_key=True)
user_name = db.Column(db.String(32), unique=True, nullable=False)
password = db.Column(db.String(32), nullable=False)
status = db.Column(db.SmallInteger, nullable=False, default=1, doc="用户状态，0-禁用，1-启动")
create_datetime = db.Column(db.DateTime, server_default=db.text("CURRENT_TIMESTAMP"), doc="创建时间")
update_datetime = db.Column(db.DateTime, nullable=False, server_default=db.text("CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP"), doc="更新时间")
archive_list = db.relationship("Archive", backref="User", lazy="dynamic")
def __init__(self, **kwargs):
super(User, self).__init__(**kwargs)
def __repr__(self):
return "User<name:%r>" % self.user_name

def is_authenticated(self):
print("------is_authenticated")
if not self.status:
return True
        return False
    def is_active(self):
print("------is_active")
if not self.status:
return True
        return False
    def get_id(self, life_time=None):
print("------get_id,life_time=", life_time)
key = current_app.config.get("SECRET_KEY", "The securet key by C~C!")
s = URLSafeSerializer(key)
browser_id = create_browser_id()
if not life_time:
life_time = current_app.config.get("TOKEN_LIFETIME")
token = s.dumps((self.id, self.user_name, self.password, browser_id, life_time))
return token

def is_anonymous(self):
print("----------is_anonymous")
return False

views.py

# coding: utf-8
# author: chenchong

import os.path
from flask import render_template, Blueprint, redirect, request, url_for, flash, g, current_app
from itsdangerous import URLSafeSerializer, BadData, constant_time_compare
from flask_login import login_user, login_required, logout_user, current_user
from app.Authentication.model import User
from app import db, login_manager
from app import simple_cache
from app.utils import create_browser_id

authentication = Blueprint("Authentication", __name__, template_folder=os.path.join(os.path.dirname(__file__), "templates"))
@login_manager.unauthorized_handler
def unauthorized():
return redirect(url_for('Authentication.login'))
@login_manager.user_loader
def user_loader(token):
"""
这里的入参就是get_id()的返回值
"""
print("----------user_loader, token=", token)
return load_token(token)
# browser的url中不会携带TOKENID这参数来访问本website，所以注释这个方法。
# @login_manager.request_loader
# def request_loader(request):
#     token = request.args.get('TOKENID', 'token_id')
#     if not token:
#         return None
#     print('------ request_loader, token=', token)
#     load_token(token)

def load_token(token):
# 通过loads()方法来解析浏览器发送过来的token，从而进行初步的验证
key = current_app.config.get("SECRET_KEY", "The securet key by C~C!")
try:
s = URLSafeSerializer(key)
id, name, password, browser_id, life_time = s.loads(token)
except BadData:
print("token had been modified!")
return None
# 判断浏览器信息是否改变
bi = create_browser_id()
if not constant_time_compare(str(bi), str(browser_id)):
print("the user environment had changed, so token has been expired!")
return None
# 校验密码
user = User.query.get(id)
if user:
# 能loads出id，name等信息，说明已经成功登录过，那么cache中就应该有token的缓存
token_cache = simple_cache.get(token)
if not token_cache:  # 此处找不到有2个原因：1.cache中因为超时失效（属于正常情况）；2.cache机制出错（属于异常情况）。
print("the token is not found in cache.")
return None
        if str(password) != str(user.password):
print("the password in token is not matched!")
simple_cache.delete(token)
return None
        else:
simple_cache.set(token, 1, timeout=life_time)  # 刷新超时时长
else:
print('the user is not found, the token is invalid!')
return None
    return user

@authentication.route('/', methods=['GET'])
def hello_world():
return redirect(url_for('Authentication.login'))
@authentication.errorhandler(404)
def page_not_found(error):
return render_template('404.html'), 404
@authentication.route('/login/', methods=['GET', 'POST'])
def login():
if request.method == "POST":
user_name = request.form.get('username')
password = request.form.get('password')
user = User.query.filter_by(user_name=user_name, password=password).first()
next = request.args.get('next')
if not user:
flash("The user name or password is not matched. C~C")
return render_template('login.html')
else:
login_user(user)  # 触发session机制，通过user.get_id()就可以获取到token
g.user = user
# 完成登录后将token存到缓存中并设置过期时间，后面校验时如果缓存中不存在，则报错
life_time = current_app.config.get("TOKEN_LIFETIME")
token = user.get_id(life_time)
print("-------- login, token=", token)
simple_cache.set(token, 1, life_time)  # 成功登录之后将token放到cache并设置失效时长，后面的验证阶段，如果从cache取不到token就说明超时了
return redirect(next or url_for('Archive.archive_list'))
else:
return render_template('login.html')
@authentication.route("/logout/")
@login_required
def logout():
if hasattr(g, 'user'):
logout_user(g.user)
logout_user()
return "Bye~"