Python cryptography加密SQLite表后解密遇InvalidToken错误求助

以下是导致cryptography.fernet.InvalidToken错误的核心原因及修复方案：

1. 密码提示误导导致输入错误 #

解密类的密码提示写错，容易让用户输入错误密码：

# 修改Data_Decryptor_SQLite类__init__方法中的提示文本
self.password = getpass.getpass(prompt = "Enter password to unlock: ").encode()

2. 盐文件路径错误或盐不一致 #

加密和解密必须使用完全相同的盐，确保get_salt方法中的文件路径准确，且加密解密共用同一个盐文件。更可靠的方式是将盐存储到SQLite内部的元数据表，避免依赖外部文件：

# 在Data_Encryptor_SQLite的create_table方法中添加盐存储逻辑
def create_table(self):
    # 原创建业务表代码...
    # 创建元数据表存储盐
    self.conn.execute("CREATE TABLE IF NOT EXISTS encryption_meta (salt BLOB)")
    self.conn.execute("DELETE FROM encryption_meta")
    self.conn.execute("INSERT INTO encryption_meta VALUES (?)", (self.salt,))
    self.conn.commit()

# 在Data_Decryptor_SQLite的get_salt方法中从SQLite读取盐
def get_salt(self):
    cursor = self.conn.cursor()
    cursor.execute("SELECT salt FROM encryption_meta LIMIT 1")
    result = cursor.fetchone()
    if not result:
        raise ValueError("无效加密数据库：未找到盐元数据")
    self.salt = result[0]

3. 数据读取/存储的格式问题 #

SQLite存储时可能意外引入空白字符，解密前先清理：

def decrypt_data(self, data):
    cleaned_data = data.strip()
    return self.cipher.decrypt(cleaned_data.encode()).decode()

4. 密钥生成一致性验证 #

在加密和解密时添加密钥验证，确保密钥生成逻辑正确：

# 在hash_password方法末尾添加验证
def hash_password(self):
    # 原密钥生成代码...
    test_str = "test_validation"
    encrypted = self.cipher.encrypt(test_str.encode())
    decrypted = self.cipher.decrypt(encrypted).decode()
    if decrypted != test_str:
        raise RuntimeError("密钥生成失败，加密解密不匹配")

完整修复后的代码示例 #

import sqlite3
import pandas as pd
from cryptography.fernet import Fernet
import base64
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
import getpass
import sys
        
class Data_Encryptor_SQLite:
    def __init__(self, db_name, df, table_name):
        self.db_name = db_name
        self.df = df
        self.table_name = table_name
        self.password = getpass.getpass(prompt = "Enter password to lock: ").encode()
        self.conn = sqlite3.connect(db_name)
        self.salt = None

        if not self.password:
            raise ValueError("密码不能为空")
        
        self.generate_or_load_salt()
        self.hash_password()
        self.create_table()
        self.encrypt_and_store_dataframe()
    
    def generate_or_load_salt(self):
        try:
            with open('encryption_salt.key', 'rb') as file:
                self.salt = file.read()
        except FileNotFoundError:
            self.salt = Fernet.generate_key()[:16]
            with open('encryption_salt.key', 'wb') as file:
                file.write(self.salt)
        
    def encrypt_data(self, data):
        return self.cipher.encrypt(str(data).encode()).decode()

    def hash_password(self):
        kdf = PBKDF2HMAC(
            algorithm=hashes.SHA256(), 
            length=32, 
            salt=self.salt, 
            iterations=100000
        )
        key = base64.urlsafe_b64encode(kdf.derive(self.password))
        self.cipher = Fernet(key)
        
        test_str = "test_key_check"
        assert self.cipher.decrypt(self.cipher.encrypt(test_str.encode())).decode() == test_str

    def create_table(self):
        columns_def = ', '.join([f"{col} TEXT" for col in self.df.columns])
        create_table_sql = f"CREATE TABLE IF NOT EXISTS {self.table_name} ({columns_def})"
        self.conn.execute(create_table_sql)
        self.conn.execute("CREATE TABLE IF NOT EXISTS encryption_meta (salt BLOB)")
        self.conn.execute("DELETE FROM encryption_meta")
        self.conn.execute("INSERT INTO encryption_meta VALUES (?)", (self.salt,))
        self.conn.commit()

    def encrypt_and_store_dataframe(self):
        for _, row in self.df.iterrows():
            encrypted_row = [self.encrypt_data(val) for val in row]
            insert_sql = f"INSERT INTO {self.table_name} VALUES ({','.join(['?' for _ in range(len(row))])})"
            self.conn.execute(insert_sql, encrypted_row)
        self.conn.commit()
        self.conn.close()
        print('加密表创建完成')


class Data_Decryptor_SQLite:
    def __init__(self, db_name, table_name):
        self.db_name = db_name
        self.table_name = table_name
        self.password = getpass.getpass(prompt = "Enter password to unlock: ").encode()
        self.conn = sqlite3.connect(db_name)
        self.salt = None
        self.df = None

        if not self.password:
            raise ValueError("密码不能为空")
        
        self.get_salt()
        self.hash_password()
        self.fetch_and_decrypt_dataframe()
        
    def get_salt(self):
        cursor = self.conn.cursor()
        cursor.execute("SELECT salt FROM encryption_meta LIMIT 1")
        result = cursor.fetchone()
        if not result:
            raise ValueError("无效加密数据库：未找到盐元数据")
        self.salt = result[0]
        
    def decrypt_data(self, data):
        cleaned_data = data.strip()
        return self.cipher.decrypt(cleaned_data.encode()).decode()

    def hash_password(self):
        kdf = PBKDF2HMAC(
            algorithm=hashes.SHA256(), 
            length=32, 
            salt=self.salt, 
            iterations=100000
        )
        key = base64.urlsafe_b64encode(kdf.derive(self.password))
        self.cipher = Fernet(key)
        
        test_str = "test_key_check"
        assert self.cipher.decrypt(self.cipher.encrypt(test_str.encode())).decode() == test_str
          
    def fetch_and_decrypt_dataframe(self):
        cursor = self.conn.cursor()
        cursor.execute(f"SELECT * FROM {self.table_name}")
        encrypted_data = cursor.fetchall()
        columns = [d[0] for d in cursor.description]
        decrypted_data = []
        for row in encrypted_data:
            decrypted_row = [self.decrypt_data(val) for val in row]
            decrypted_data.append(decrypted_row)
        self.df = pd.DataFrame(decrypted_data, columns=columns)
        self.conn.close()

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