Flask实现token认证

本文转载自查看原文 2020-11-21 19:33 4773 Django

1. 安装

pip install flask_httpauth
pip install itsdangerous

2. 创建对象

from flask import Flask, g, jsonify from flask_httpauth import HTTPBasicAuth app = Flask(__name__) auth = HTTPBasicAuth()

3. 生成token

from itsdangerous import TimedJSONWebSignatureSerializer as Serializer # 密钥，可随意修改 SECRET_KEY = 'abcdefghijklmm' # 生成token, 有效时间为600min def generate_auth_token(user_id, expiration=36000): # 第一个参数是内部私钥 # 第二个参数是有效期（秒） s = Serializer(SECRET_KEY, expires_in=expiration) return s.dumps({'user_code': user_code})

4. 解析token

from itsdangerous import TimedJSONWebSignatureSerializer as Serializer from itsdangerous import BadSignature, SignatureExpired # 密钥，可随意修改 SECRET_KEY = 'abcdefghijklmm' # 解析token def verify_auth_token(token): s = Serializer(SECRET_KEY) # token正确 try: data = s.loads(token) return data # token过期 except SignatureExpired: return None # token错误 except BadSignature: return None

5. 验证token

# 数据库配置 db = { 'host': '127.0.0.1', 'user': 'root', 'password': 'admin123', 'port': 3306, 'database': 'school', 'charset': 'utf8' } # 连接数据库操作 def connect(user_id, password): conn = mc.connect(**db) cursor = conn.cursor(dictionary=True) args = (user_id, password) verify_sql = "SELECT user_id FROM account WHERE user_id = %s AND pwd = PASSWORD(%s);" cursor.execute(verify_sql, args) user = cursor.fetchone() return user # 验证token @auth.verify_password def verify_password(username, password): # 先验证token user_id = re.sub(r'^"|"$', '', username) user_id = verify_auth_token(user_id) # 如果token不存在，验证用户id与密码是否匹配 if not user_id: user_id = connect(username, password) # 如果用户id与密码对应不上，返回False if not user_id: return False g.user_id = user_id.get('user_id') return True

6. 完整代码

配置文件config.py

# 数据库配置
db = {
        'host': '127.0.0.1',
        'user': 'root',
        'password': 'admin123',
        'port': 3306,
        'database': 'school',
        'charset': 'utf8'
}

# 密钥，可随意修改
SECRET_KEY = 'abcdefghijklmm'

主文件run.py

from flask import Flask, g, jsonify from flask_httpauth import HTTPBasicAuth from itsdangerous import TimedJSONWebSignatureSerializer as Serializer from itsdangerous import BadSignature, SignatureExpired from flask_cors import CORS import mysql.connector as mc from config import db, SECRET_KEY import re app = Flask(__name__) CORS(app, supports_credentials=True) auth = HTTPBasicAuth() # 生成token, 有效时间为600min def generate_auth_token(user_id, expiration=36000): s = Serializer(SECRET_KEY, expires_in=expiration) return s.dumps({'user_id': user_id}) # 解析token def verify_auth_token(token): s = Serializer(SECRET_KEY) # token正确 try: data = s.loads(token) return data # token过期 except SignatureExpired: return None # token错误 except BadSignature: return None # 连接数据库操作 def connect(user_id, password): conn = mc.connect(**db) cursor = conn.cursor(dictionary=True) args = (user_id, password) verify_sql = "SELECT user_id FROM account WHERE user_id = %s AND pwd = PASSWORD(%s);" cursor.execute(verify_sql, args) user = cursor.fetchone() return user # 验证token @auth.verify_password def verify_password(username, password): # 先验证token user_id = re.sub(r'^"|"$', '', username) user_id = verify_auth_token(user_id) # 如果token不存在，验证用户id与密码是否匹配 if not user_id: user_id = connect(username, password) # 如果用户id与密码对应不上，返回False if not user_id: return False g.user_id = user_id.get('user_id') return True @app.route('/login') @auth.login_required def login(): token = generate_auth_token(g.user_id) return jsonify({'token': token}) @app.route('/index') @auth.login_required def index(): return 'index' if __name__ == '__main__': app.run()