Flask中JWT令牌数据库存储失败及私钥签名异常求助

问题概述 #

在Flask+PostgreSQL架构的认证功能中，遇到两个核心问题：

• 用户注册后，JWT令牌能正常打印但无法存入数据库
• 使用私钥签名JWT令牌时存在异常

相关代码片段 #

# Function to generate token
def generate_token(username, email, expiration_time=20):
    # Load the private key
    with open('private_key.pem', 'r') as f:
        private_key = f.read()
        
    payload_data ={
        'username': username,
        'email': email,
        'exp': int(time.time()) + expiration_time
    }
    token = jwt.encode(
        payload=payload_data,
        key=private_key,
        algorithm='RS256',
    )
    
    return token

# Route for user registration
@user_bp.route('/register', methods=['POST'])
def register_user():
    # Extract json data from HTTP request using the get_json() method of the request object
    data = request.get_json()
    # Check if all mandatory fields are filled, if not return error
    required_fields = ['username', 'email', 'password', 'confirm_password']
    for field in required_fields:
        if field not in data or not data[field]:
            return jsonify({'error': f'Field {field} missing or empty!'}), 400

    username = data['username']
    email = data['email']
    password = data['password']
    confirm_password = data['confirm_password']

    if password != confirm_password:
        return jsonify({'error': 'Passwords do not match!'}), 400

    try:
        conn = connect_db()
        cursor = conn.cursor()

        # Check if username and email of the user already exist in the database, if so return error
        cursor.execute("SELECT * FROM users WHERE username=%s OR email=%s", (username, email))
        if cursor.fetchone() is not None:
            return jsonify({'error': 'Username or email already exists!'}), 400

        # Encryption before insertion into the database
        hashed_password = bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt())

        # Creation of JWT token
        expiration_time = 20  # Token expiration time in 20 seconds 
        token = generate_token(username, email, expiration_time)
        print("token before inserting into db:", token)
        # Insert the new user into the database
        cursor.execute("INSERT INTO users (username, email, password,token) VALUES (%s, %s, %s, %s) RETURNING *",
                       (username, email, hashed_password.decode('utf-8'), token))
        new_user = cursor.fetchone()
        print("token after insertion:", new_user[3] if new_user else "No user returned")
        conn.commit()

        if new_user:
            user_data = {
                'id': new_user[0],
                'username': new_user[1],
                'email': new_user[2],
                'token': new_user[3]
            }
            
        else:
            return jsonify({'error': 'Error retrieving user data after registration!'}), 500

    except Exception as e:
        return jsonify({'error': str(e)}), 500
    finally:
        if 'conn' in locals():
            conn.close()
    return jsonify({'success': True, 'message': 'User registered successfully!'}), 201

排查与解决方案 #

1. JWT令牌未存入数据库的解决 #

可能原因及修复：

• 字段类型不匹配：检查PostgreSQL的users表，确保token字段类型为TEXT（JWT令牌长度较长，VARCHAR可能因长度限制导致插入失败）。若字段类型为VARCHAR，需设置足够长度（如VARCHAR(2048)）或改为TEXT。
• 数据类型错误：旧版本PyJWT中jwt.encode()返回bytes类型，而PostgreSQL的TEXT字段无法存储bytes。需将token转为字符串，在generate_token函数末尾添加：

  if isinstance(token, bytes):
      token = token.decode('utf-8')
  

• 表字段名大小写问题：PostgreSQL默认对未加引号的字段名转为小写，若表字段名为Token（首字母大写），插入语句中的token会匹配不到字段，导致值未存入。需确保插入语句的字段名与表结构一致。
• 事务提交验证：在conn.commit()前打印new_user的token值，确认数据库是否返回了正确的token。若new_user[3]为空，说明插入时字段映射错误。

2. 私钥签名JWT异常的解决 #

可能原因及修复：

• 依赖缺失：PyJWT的RS256算法依赖cryptography库，需安装完整依赖：

  pip install pyjwt[crypto]
  

• 私钥格式错误：确保private_key.pem是有效的RSA私钥，格式需符合以下两种之一：
  • PKCS#1格式：

    -----BEGIN RSA PRIVATE KEY-----
    ... 私钥内容 ...
    -----END RSA PRIVATE KEY-----
    

  • PKCS#8格式：

    -----BEGIN PRIVATE KEY-----
    ... 私钥内容 ...
    -----END PRIVATE KEY-----
    

  可通过openssl工具生成合法私钥：

  openssl genrsa -out private_key.pem 2048
  

• 私钥加载编码问题：读取私钥时指定编码，避免乱码：

  with open('private_key.pem', 'r', encoding='utf-8') as f:
      private_key = f.read()
  

• 异常捕获与调试：在generate_token函数中添加异常捕获，打印具体错误信息，便于定位问题：

  def generate_token(username, email, expiration_time=20):
      try:
          with open('private_key.pem', 'r', encoding='utf-8') as f:
              private_key = f.read()
          payload_data ={
              'username': username,
              'email': email,
              'exp': int(time.time()) + expiration_time
          }
          token = jwt.encode(
              payload=payload_data,
              key=private_key,
              algorithm='RS256',
          )
          if isinstance(token, bytes):
              token = token.decode('utf-8')
          return token
      except Exception as e:
          print(f"Token generation failed: {str(e)}")
          raise
  

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