生成token 與解密 token 和 攔截器

#學(xué)習(xí)交流 訪問
# https://v.iiar.cn

import jwt
import datetime
from models import XUser
from flask import request, jsonify
from functools import wraps

SECRET_KEY = 'XPay'


# 創(chuàng)建token
def generate_token(user_id):
    try:
        payload = {
            'exp': datetime.datetime.utcnow() + datetime.timedelta(days=7),
            'iat': datetime.datetime.utcnow(),
            'sub': user_id
        }
        return jwt.encode(
            payload,
            SECRET_KEY,  # 替換為你的密鑰
            algorithm='HS256'
        )
    except Exception as e:
        return e


# 解析token
def decode_token(token):
    """ 解碼Token并處理異常 """
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=['HS256'])
        return payload['sub']  # 返回成功標(biāo)志和用戶ID
    except jwt.ExpiredSignatureError as e:
        return False
    except jwt.InvalidTokenError as e:
        return False


# 查詢用戶狀態(tài)
def check_token_and_user_status(token):
    user_id = decode_token(token)
    print(user_id)
    token_in_user = XUser.query.filter_by(token=token).first()

    # 根據(jù)是否開啟多端登陸判斷token是否有效
    multi_device_login = True
    if not multi_device_login and not token_in_user:
        return False, 'token過期'

    if user_id:
        user = XUser.query.get(user_id)
        if user:
            if user.state == '正常':
                return True, user
            else:
                return False, '該用戶狀態(tài)異常，請聯(lián)系客服'
        else:
            return False, '用戶不存在'
    else:
        return False, 'token無效'


def user_token_required(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        token = request.headers.get('token')
        if not token:
            return jsonify({
                'code': 401,
                'data': '',
                'msg': 'token不存在',
                'time': datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            })

        user_state, user_info = check_token_and_user_status(token)
        if not user_state:
            return jsonify({
                'code': 401,
                'data': '',
                'msg': user_info,
                'time': datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            })
        return f(*args, **kwargs, user_info=user_info)

    return decorated_function

詳細(xì)解釋

這段代碼提供了一個使用 Python 和 Flask 結(jié)合 JWT (JSON Web Tokens) 進(jìn)行用戶認(rèn)證的簡單框架。它包括了生成 token、解碼 token、檢查用戶狀態(tài)和一個裝飾器函數(shù)，用于保護(hù)需要認(rèn)證的路由。下面是對代碼的逐部分解釋：

1. generate_token(user_id) 函數(shù)

• 這個函數(shù)用于為指定的用戶 ID 生成一個 JWT token。Token 包含三個重要的信息（稱為payload）：exp（過期時間），iat（發(fā)行時間），和sub（主題，這里用作用戶 ID）。
• Token 使用 HS256 算法進(jìn)行簽名，SECRET_KEY作為簽名密鑰。
• Token 默認(rèn)有效期為 7 天。

2. decode_token(token) 函數(shù)

• 用于解碼和驗(yàn)證接收到的 JWT token。如果 token 有效且未過期，這個函數(shù)返回保存在 token 的sub字段中的用戶 ID。
• 如果 token 過期（ExpiredSignatureError）或無效（InvalidTokenError），函數(shù)將返回 False。

3. check_token_and_user_status(token) 函數(shù)

• 首先解碼 token 來獲取用戶 ID，然后查詢數(shù)據(jù)庫中是否存在對應(yīng)的用戶和 token。
• 如果設(shè)置了不允許多端登錄（multi_device_login = False），它還會檢查數(shù)據(jù)庫中的 token 是否與提供的 token 匹配。
• 根據(jù)用戶的狀態(tài)（如是否正常）和 token 的有效性返回相應(yīng)的狀態(tài)和信息。

4. user_token_required(f) 裝飾器

• 一個 Flask 路由裝飾器，用于在執(zhí)行實(shí)際路由函數(shù)之前進(jìn)行用戶認(rèn)證。
• 從請求頭中獲取 token，如果不存在 token 或 token 無效，則返回錯誤信息。
• 如果用戶認(rèn)證成功，路由函數(shù)將正常執(zhí)行。裝飾器會將user_info（從check_token_and_user_status返回的用戶信息）作為關(guān)鍵字參數(shù)傳遞給路由函數(shù)。

使用場景

• 在需要對用戶身份進(jìn)行驗(yàn)證的 Flask 路由上，可以使用@user_token_required裝飾器來確保只有攜帶有效 token 的請求才能訪問。
• 這套系統(tǒng)使得管理用戶會話和訪問控制變得簡單，特別是在構(gòu)建 RESTful API 時非常有用。

注意事項(xiàng)

• 實(shí)際部署時，SECRET_KEY應(yīng)當(dāng)是一個安全的值，且不應(yīng)該硬編碼在代碼中。
• 在處理用戶狀態(tài)和多端登錄邏輯時，需要根據(jù)實(shí)際業(yè)務(wù)需求進(jìn)行調(diào)整。
• XUser模型和數(shù)據(jù)庫查詢邏輯需要根據(jù)實(shí)際的數(shù)據(jù)庫設(shè)計來實(shí)現(xiàn)。這里假設(shè)XUser是一個模型類，代表用戶數(shù)據(jù)，且有一個state屬性和一個token屬性。

模型部分

class XUser(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    mid = db.Column(db.Integer, comment='推薦人id')
    phone = db.Column(db.String(80), comment='手機(jī)號')
    user_name = db.Column(db.String(80), comment='用戶昵稱')
    openid = db.Column(db.String(120), comment='openid')
    password = db.Column(db.String(120), comment='密碼')
    registration_time = db.Column(db.DateTime, default=datetime.now, comment='注冊時間')
    token = db.Column(db.String(255), comment='Token')
    account_balance = db.Column(db.Numeric(20, 4), default=0.00, comment="賬戶余額")
    state = db.Column(db.String(120), default='正常', comment='用戶狀態(tài)')
    fee_percentage = db.Column(db.Integer, default=4, comment='手續(xù)費(fèi)比例')

    def to_dict(self):
        return {
            'id': self.id,
            'mid': self.mid,
            'phone': self.phone,
            'user_name': self.user_name,
            'openid': self.openid,
            'registration_time': self.registration_time.strftime(
                '%Y-%m-%d %H:%M:%S') if self.registration_time else None,
            'account_balance': float(self.account_balance) if self.account_balance is not None else None,
            'state': self.state,
            'fee_percentage': self.fee_percentage
        }

模型解釋

這個XUser類是一個模型定義，使用 SQLAlchemy ORM （一個 Python SQL 工具包和對象關(guān)系映射器）用于定義和操作數(shù)據(jù)庫中的用戶表。每個屬性（使用db.Column聲明）對應(yīng)用戶表中的一個字段。以下是各個字段的解釋：

字段解釋

• id: 用戶的唯一標(biāo)識符，整數(shù)類型，設(shè)為主鍵。
• mid: 推薦人的ID，整數(shù)類型，用來表示這個用戶是被哪個已存在的用戶推薦的。
• phone: 用戶的手機(jī)號，字符串類型，最長80字符。
• user_name: 用戶昵稱，字符串類型，最長80字符。
• openid: 用戶的微信OpenID，字符串類型，最長120字符。這是微信平臺特有的標(biāo)識用戶的ID，用于識別微信用戶。
• password: 用戶密碼，字符串類型，最長120字符。在實(shí)際應(yīng)用中，密碼應(yīng)該經(jīng)過加密存儲。
• registration_time: 用戶注冊時間，DateTime類型，默認(rèn)值為當(dāng)前時間。這里使用了datetime.now來自動設(shè)置這個字段的值。
• token: 用于認(rèn)證的Token，字符串類型，最長255字符。這是在用戶登錄時生成的，用于后續(xù)請求的身份驗(yàn)證。
• account_balance: 用戶賬戶余額，數(shù)值類型，最多20位數(shù)字，小數(shù)點(diǎn)后最多4位。默認(rèn)值為0.00。
• state: 用戶狀態(tài)，字符串類型，最長120字符，默認(rèn)值為’正?！＿@個字段可以用來表示用戶是否被禁用或其他狀態(tài)。
• fee_percentage: 手續(xù)費(fèi)比例，整數(shù)類型。表示在進(jìn)行某些交易時需要收取的手續(xù)費(fèi)百分比，默認(rèn)值為4%。

to_dict 方法

to_dict方法是一個實(shí)例方法，用于將XUser對象的屬性轉(zhuǎn)換成一個字典，這在進(jìn)行JSON序列化時非常有用，例如在 RESTful API 響應(yīng)中返回用戶信息。這個方法特別處理了registration_time和account_balance字段，確保它們以適當(dāng)?shù)母袷捷敵觯?code>registration_time格式化為字符串，account_balance確保為浮點(diǎn)數(shù)或None）。

總結(jié)

XUser類不僅定義了數(shù)據(jù)庫表結(jié)構(gòu)，還提供了方便的方法來操作和轉(zhuǎn)換用戶數(shù)據(jù)。通過使用 SQLAlchemy ORM，可以簡化數(shù)據(jù)庫操作，提高代碼的可讀性和可維護(hù)性。

用戶注冊/登陸/獲取用戶基本信息

from datetime import datetime
from models import db, XUser
from flask import request, jsonify, Blueprint

from tools.common_method import check_password, set_password
from tools.user_token import generate_token
from tools.user_token import user_token_required

user_api = Blueprint('user_api', __name__)


# 登陸
@user_api.route('/api/user/login', methods=['POST'])
def user_login():
    data = request.json
    phone = data.get('phone')
    password = data.get('password')
    user = XUser.query.filter_by(phone=phone).first()
    print(phone)
    if not user:
        return jsonify({
            'code': 403,
            'data': '',
            'msg': '用戶賬號或密碼錯誤',
            'time': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        })
    else:
        if check_password(user.password, password) and user.state == '正常':
            token = generate_token(user.id)
            user.token = token
            db.session.commit()
            return jsonify({
                'code': 200,
                'token': token,
                'msg': '登陸成功',
                'time': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            })
        else:
            return jsonify({
                'code': 403,
                'data': '',
                'msg': '該用戶狀態(tài)不允許登陸',
                'time': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            })


# 注冊
@user_api.route('/api/user/register', methods=['POST'])
def user_register():
    data = request.json
    phone = data.get('phone')
    password = data.get('password')
    user_name = data.get('user_name')
    password_len = len(password)
    phone_len = len(phone)
    if phone_len != 11:
        return jsonify({
            'code': 403,
            'data': '',
            'msg': '注冊失敗，請輸入11位手機(jī)號',
            'time': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        })
    if password_len < 6 or password_len > 32:
        return jsonify({
            'code': 403,
            'data': '',
            'msg': '注冊失敗，密碼長度請大于6位，小于32位',
            'time': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        })
    user = XUser.query.filter_by(phone=phone).first()
    if user:
        return jsonify({
            'code': 403,
            'data': '',
            'msg': '注冊失敗，該手機(jī)已經(jīng)注冊，請直接登陸或更換手機(jī)后注冊',
            'time': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        })
    else:
        new_password = set_password(password)
        new_user = XUser(
            phone=phone,
            password=new_password,
            user_name=user_name
        )
        db.session.add(new_user)
        db.session.commit()
        token = generate_token(new_user.id)
        new_user.token = token
        db.session.commit()
        return jsonify({
            'code': 200,
            'token': token,
            'msg': '注冊成功',
            'time': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        })


@user_api.route('/api/user/user_info', methods=['GET'])
@user_token_required
def get_user_info(user_info):
    return reg_func(200, user_info.to_dict(), '獲取信息成功')


def reg_func(code, data, msg):
    return jsonify({
        'code': code,
        'data': data,
        'msg': msg,
        'time': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    }), code

注冊/登陸/獲取用戶信息代碼詳細(xì)解釋

這段代碼是一個使用 Flask 框架構(gòu)建的簡易用戶管理系統(tǒng)的一部分，包括用戶的登錄、注冊以及獲取用戶信息的功能。它利用了 Flask 的 Blueprint 功能來組織和注冊相關(guān)的路由，以及 SQLAlchemy ORM 來處理數(shù)據(jù)庫操作。下面是對代碼主要部分的解釋：

登錄 (user_login)

• 接收客戶端發(fā)送的 JSON 數(shù)據(jù)，包含用戶的phone和password。
• 通過手機(jī)號查詢數(shù)據(jù)庫中的用戶。
• 如果用戶不存在，返回403錯誤和消息"用戶賬號或密碼錯誤"。
• 如果用戶存在，調(diào)用check_password函數(shù)驗(yàn)證密碼是否正確，并檢查用戶狀態(tài)是否為"正常"。
• 如果驗(yàn)證通過，為用戶生成一個新的 token（使用generate_token函數(shù)），更新用戶的 token 字段，并提交到數(shù)據(jù)庫。
• 返回包含新 token 和成功消息的 JSON 響應(yīng)。

注冊 (user_register)

• 同樣接收包含phone、password和user_name的 JSON 數(shù)據(jù)。
• 驗(yàn)證手機(jī)號是否為11位，密碼長度是否在6到32位之間。
• 如果手機(jī)號或密碼格式不符合要求，返回403錯誤和相應(yīng)的失敗消息。
• 如果手機(jī)號已經(jīng)被注冊，返回403錯誤和消息"注冊失敗，該手機(jī)已經(jīng)注冊，請直接登陸或更換手機(jī)后注冊"。
• 如果驗(yàn)證通過，使用set_password函數(shù)加密密碼，然后創(chuàng)建一個新的XUser實(shí)例，并將其添加到數(shù)據(jù)庫。
• 為新用戶生成 token，更新用戶的 token 字段，并提交到數(shù)據(jù)庫。
• 返回包含新 token 和成功消息的 JSON 響應(yīng)。

獲取用戶信息 (get_user_info)

• 這個路由使用@user_token_required裝飾器，要求請求必須包含有效的 token。
• 如果認(rèn)證通過，裝飾器會傳遞user_info參數(shù)（用戶信息）給get_user_info函數(shù)。
• 函數(shù)調(diào)用reg_func生成標(biāo)準(zhǔn)的 JSON 響應(yīng)，包含用戶信息和成功消息。

輔助函數(shù) (reg_func)

• reg_func是一個幫助函數(shù)，用于生成標(biāo)準(zhǔn)化的 JSON 響應(yīng)。它接收狀態(tài)碼、數(shù)據(jù)和消息作為參數(shù)，并返回一個 Flask jsonify 響應(yīng)。

關(guān)鍵工具和方法

• Blueprint：用于創(chuàng)建一組相關(guān)的路由和視圖函數(shù)。
• request.json：用于獲取 JSON 格式的請求體數(shù)據(jù)。
• jsonify：將數(shù)據(jù)轉(zhuǎn)換為 JSON 響應(yīng)。
• db.session：用于數(shù)據(jù)庫操作，如添加新記錄和提交更改。
• generate_token和user_token_required：自定義函數(shù)和裝飾器，用于處理 JWT token 的生成和驗(yàn)證。

整個代碼展示了如何在 Flask 應(yīng)用中實(shí)現(xiàn)用戶認(rèn)證和管理的基本流程，使用 JWT tokens 提供安全的用戶狀態(tài)管理和接口保護(hù)。文章來源地址http://www.zghlxwxcb.cn/news/detail-836698.html

到了這里，關(guān)于快速實(shí)現(xiàn)用戶認(rèn)證：使用Python和Flask配合PyJWT生成與解密Token的教程及示例代碼的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！