Flask JWT登录认证异常及数据验证、密码校验等技术疑问

核心问题分析与解决 #

1. 登录恒生成Token的原因 #

你的登录接口SQL查询写了WHERE user_name = user_name and password = password，这是恒真条件——不管输入什么都会匹配到数据，所以永远会生成Token。必须改成带正确参数绑定的查询逻辑。

2. 登录数据验证与哈希密码校验 #

• 登录数据验证：用Marshmallow写专门的登录Schema，确保用户名和密码必填且符合格式要求。
• 哈希密码校验：注册时用generate_password_hash生成哈希值存储，登录时不能直接对比明文和哈希值，必须用check_password_hash(数据库存储的哈希密码, 用户输入的明文密码)来验证合法性。

3. SQLAlchemy Core下的JWT库选择 #

推荐用Flask-JWT-Extended，理由：

• 不依赖ORM，完美适配SQLAlchemy Core的原生SQL操作
• 功能全面：支持access token、refresh token、token过期、权限控制等
• 配置简单，文档清晰，社区活跃度高

修正后的完整代码 #

from dataclasses import fields
import datetime
import json
from typing_extensions import Required
from sqlalchemy import DateTime
from email.policy import default, strict
from flask import Flask, jsonify, request, make_response
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from sqlalchemy.sql import text, func
from flask_marshmallow import Marshmallow
from marshmallow import Schema, fields, validates, ValidationError, validate
from sqlalchemy.sql import select, insert, bindparam
from sqlalchemy.dialects.postgresql import UUID
import uuid
from werkzeug.security import generate_password_hash, check_password_hash
from sqlalchemy import create_engine
import logging
from flask_jwt_extended import JWTManager, jwt_required, create_access_token
from functools import wraps
import jwt

engine = create_engine('mysql://root:admin@172.17.0.2:3306/flask')

app = Flask(__name__)

app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:admin@172.17.0.2:3306/flask'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SECRET_KEY'] = 'thisissecret'
# 启用JWT配置，替换为自己的安全随机密钥
app.config['JWT_SECRET_KEY'] = 'your-super-safe-jwt-key-2024' 
db = SQLAlchemy(app)
ma = Marshmallow(app)
migrate = Migrate(app, db)
# 初始化JWT管理器
jwt = JWTManager(app)

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    user_name = db.Column(db.String(100), nullable=False)
    password = db.Column(db.String(255), nullable=False)  # 哈希密码长度较长，调整字段长度
    email_address = db.Column(db.String(80), unique=True, nullable=False)
    dob = db.Column(db.DateTime, nullable=True)
    address = db.Column(db.String(200))
    uid = db.Column(db.Text(length=16), nullable = False)

    def __init__(self, user_name, password, email_address, dob, address, uid):
        self.user_name = user_name
        self.password = password
        self.email_address = email_address
        self.dob = dob
        self.address = address
        self.uid = uid

# 用户注册Schema
class UserSchema(Schema):
    user_name = fields.String(required=True, validate=validate.Length(min=5))
    password = fields.String(required=True, validate=validate.Length(min=6))  # 增加密码长度验证
    email_address = fields.String(required=True, error_messages={"required": "邮箱地址为必填项"})
    dob = fields.DateTime(required= True, error_messages={"required": "出生日期为必填项"})
    address = fields.String()
    uid = fields.UUID(error_messages={'null': 'ID字段不能为空', 'invalid_uuid': 'ID必须是有效的UUID'})

    @validates('dob')
    def is_not_in_future(self,value):
        now = datetime.datetime.now()
        if value > now:
            raise ValidationError("出生日期不能是未来时间！")

    class Meta:
        fields = ('id','user_name', 'password', 'email_address', 'dob', 'address')   

# 登录专用Schema
class LoginSchema(Schema):
    user_name = fields.String(required=True, validate=validate.Length(min=5))
    password = fields.String(required=True, validate=validate.Length(min=6))

# 初始化Schema
user_schema = UserSchema()
users_schema = UserSchema(many=True)
login_schema = LoginSchema()

@app.route('/users', methods=['GET'])
def get_all_users():
    conn = engine.connect()
    str_sql = text("SELECT * FROM flask.`user`;")
    results = conn.execute(str_sql).fetchall()
    return users_schema.dump(results)

@app.route('/register', methods=['POST'])
def user_register():
    errors = user_schema.validate(request.get_json())
    if errors:
        return jsonify(errors), 400

    data_dict = user_schema.load(request.get_json())
    conn = engine.connect()

    user_name = data_dict['user_name']
    password = generate_password_hash(data_dict['password'], method='sha256')
    email_address = data_dict['email_address']
    dob = data_dict['dob']
    address = data_dict.get('address', '')  # 可选字段默认空值
    uid = uuid.uuid4().hex

    sql = text("""INSERT INTO user (user_name, password, email_address, dob, address, uid ) 
                  VALUES(:user_name , :password, :email_address, :dob, :address, :uid)""")
    data = {"user_name": user_name, "password": password, "email_address": email_address, 
            "dob": dob, "address": address, "uid": uid}
    conn.execute(sql, data)
    return jsonify({'Message':'新用户创建成功'}), 201

@app.route('/login', methods=['POST'])
def login():
    # 先验证登录数据格式
    errors = login_schema.validate(request.get_json())
    if errors:
        return jsonify(errors), 400

    request_data = request.get_json()
    user_name = request_data['user_name']
    input_password = request_data['password']
    conn = engine.connect()

    # 根据用户名查询用户，使用正确的参数绑定
    sql = text("SELECT user_name, password FROM user WHERE user_name = :user_name")
    user_data = conn.execute(sql, {"user_name": user_name}).fetchone()

    if user_data:
        # 校验哈希密码
        stored_hash = user_data[1]
        if check_password_hash(stored_hash, input_password):
            access_token = create_access_token(identity=user_name)
            return jsonify(message='登录成功', access_token=access_token), 200
    # 用户不存在或密码错误
    return jsonify(message='用户名或密码错误'), 401

if __name__ == '__main__':
    app.run(debug=True)

关键修改点说明 #

1. 启用JWT管理器：取消JWT_SECRET_KEY和jwt = JWTManager(app)的注释，设置安全密钥。
2. 修复登录SQL查询：把恒真条件改成根据用户名精准查询，避免匹配所有用户。
3. 增加登录数据验证：新增LoginSchema确保登录时用户名和密码必填且符合长度要求。
4. 哈希密码校验：查询到用户后，用check_password_hash对比输入密码和存储的哈希值，而非直接对比明文。
5. 调整密码字段长度：将password字段长度从100改为255，避免哈希值被截断。
6. 完善响应状态码：注册成功返回201，验证失败返回400，登录失败返回401，符合REST规范。

内容的提问来源于stack exchange，提问作者R P