GridDB传输数据SSL/TLS加密配置及数据加解密报错排查

问题根源 #

虽然你已经处理了字符串与字节的转换，但GridDB的STRING类型在存储/检索过程中可能会对字符串进行隐式处理（如编码转换、字符转义或截断），导致还原后的字节流与原始加密token不一致，最终触发TypeError: token must be bytes错误（实际是无效token的间接表现）。

修复方案 #

改用GridDB的BYTE_ARRAY类型直接存储加密后的字节数据，完全跳过字符串转换步骤，避免中间处理带来的问题。

修改后的完整代码 #

from cryptography.fernet import Fernet
import griddb_python as griddb

# 生成密钥并加密数据
key = Fernet.generate_key()
cipher_suite = Fernet(key)
data = "Sensitive Data"
encrypted_data = cipher_suite.encrypt(data.encode())

# 建立GridDB连接
factory = griddb.StoreFactory.get_instance()
try:
    gridstore = factory.get_store(
        host="localhost", port=10001, cluster_name="defaultCluster",
        username="admin", password="admin"
    )

    # 创建或获取容器：将data列类型改为BYTE_ARRAY
    container_name = "data_encryption_test"
    conInfo = griddb.ContainerInfo(
        name=container_name,
        column_info_list=[["id", griddb.Type.INTEGER],
                          ["data", griddb.Type.BYTE_ARRAY]],  # 此处修改列类型
        type=griddb.ContainerType.COLLECTION, row_key=True)
    
    container = gridstore.put_container(conInfo)
    container.set_auto_commit(False)
    
    # 存储加密数据：直接传入字节数据，无需转字符串
    row = [1, encrypted_data]
    container.put_row(row)
    container.commit()

    # 检索并解密数据：直接获取字节数据，无需转码
    query = container.query("select *")
    rs = query.fetch()
    while rs.has_next():
        row = rs.next()
        encrypted_data_retrieved = row[1]
        decrypted_data = cipher_suite.decrypt(encrypted_data_retrieved).decode()
        print("Decrypted Data:", decrypted_data)

except griddb.GSException as e:
    print("GridDB Error:", e)

额外验证（若坚持使用STRING类型） #

如果必须用STRING类型存储，需验证存储前后的加密字符串完全一致：

• 存储前打印encrypted_data.decode()的内容
• 检索后打印row[1]的内容
• 对比两者是否完全相同（包括末尾的填充符=）

但优先推荐使用BYTE_ARRAY类型，更贴合加密数据的二进制属性，避免编码层的潜在问题。

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