怎样在LLM应用中实现输入/输出的加密和完整性校验？

简介：为什么LLM应用需要加密和校验？

随着大语言模型（LLM）被广泛应用于处理敏感的用户查询和生成专属内容，数据安全成为了部署中的首要挑战。用户的Prompt可能包含个人身份信息（PII）、商业机密或专有数据。在这些数据通过网络传输（客户端到API网关，或API网关到模型服务）时，我们必须保证两个核心要素：

1. 机密性 (Confidentiality): 确保只有合法的发送方和接收方才能读取数据（加密）。
2. 完整性/认证性 (Integrity & Authenticity): 确保数据在传输过程中未被篡改，并确认数据来源的合法性（HMAC或认证加密）。

在AI基础设施中，实现端到端的加密（E2EE）是保护LLM流量的最佳实践。本文将使用工业标准的 AES-256-GCM 认证加密模式，它能同时提供机密性和完整性校验。

准备工作：安装依赖

我们将使用Python中最权威的加密库 cryptography。

pip install cryptography

核心实现：AES-GCM 加密和解密模块

AES-GCM (Galois/Counter Mode) 通过引入一个认证标签（Authentication Tag）来解决传统加密模式缺乏完整性校验的问题。如果解密时标签不匹配，则表明数据在传输中被篡改，解密操作将失败。

1. 密钥生成与设置

出于演示目的，我们使用一个预共享密钥（Pre-Shared Key, PSK）。在生产环境中，密钥应通过安全的密钥管理系统（KMS）或强大的Diffie-Hellman握手进行交换。

import os
import base64
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes

# 生产环境密钥应至少为256位（32字节）
# 假设这是客户端和服务端共享的秘密密钥
SHARED_ENCRYPTION_KEY = os.urandom(32) 

# 辅助函数：将bytes转换为base64字符串，以便于在JSON或HTTP Headers中传输
def b64encode(data): 
    return base64.urlsafe_b64encode(data).decode('utf-8')

def b64decode(data): 
    return base64.urlsafe_b64decode(data.encode('utf-8'))

2. 加密函数 (用于客户端发送Prompt或服务端发送Response)

加密函数需要生成一个随机的 Nonce (IV) 并返回密文、Nonce 和认证标签 (Tag)。

def encrypt_payload(plaintext: str, key: bytes):
    """使用AES-256-GCM加密数据，并返回密文、Nonce和Tag。"""
    # 1. 生成随机的Nonce (GCM推荐12字节)
    nonce = os.urandom(12)

    # 2. 初始化加密器
    cipher = Cipher(algorithms.AES(key), modes.GCM(nonce), backend=default_backend())
    encryptor = cipher.encryptor()

    # 3. 执行加密
    ciphertext = encryptor.update(plaintext.encode('utf-8')) + encryptor.finalize()

    # 4. 获取认证标签 (完整性校验的关键)
    tag = encryptor.tag

    # 5. 编码为可传输的格式
    return {
        "ciphertext": b64encode(ciphertext),
        "nonce": b64encode(nonce),
        "tag": b64encode(tag)
    }

3. 解密与完整性校验函数 (用于服务端接收Prompt或客户端接收Response)

解密函数在执行解密时，会自动校验提供的认证标签。如果标签不匹配，意味着数据被篡改，decryptor.finalize() 将抛出 InvalidTag 异常。

from cryptography.exceptions import InvalidTag

def decrypt_payload(encrypted_data: dict, key: bytes):
    """使用AES-256-GCM解密数据并进行完整性校验。"""
    try:
        # 1. 解码传输数据
        ciphertext = b64decode(encrypted_data['ciphertext'])
        nonce = b64decode(encrypted_data['nonce'])
        tag = b64decode(encrypted_data['tag'])

        # 2. 初始化解密器，传入Nonce和Tag
        cipher = Cipher(algorithms.AES(key), modes.GCM(nonce, tag), backend=default_backend())
        decryptor = cipher.decryptor()

        # 3. 执行解密。如果Tag不匹配，finalize() 将抛出 InvalidTag 异常
        plaintext = decryptor.update(ciphertext) + decryptor.finalize()

        return plaintext.decode('utf-8')

    except InvalidTag:
        raise ValueError("数据完整性校验失败：认证标签无效或数据已被篡改！")
    except Exception as e:
        raise ValueError(f"解密失败: {e}")

实际应用示例：LLM交互流

假设我们有一个LLM客户端和一个提供推理服务的服务器。

步骤 1：客户端加密用户Prompt

用户输入一个包含敏感信息的Prompt。

# 客户端操作
client_prompt = "请为我总结我昨天发送的，关于项目'Alpha-3'的内部邮件内容。"
print(f"原始Prompt: {client_prompt}")

encrypted_request = encrypt_payload(client_prompt, SHARED_ENCRYPTION_KEY)

print("\n--- 客户端加密后的请求数据包 ---")
import json
print(json.dumps(encrypted_request, indent=2))

# 客户端将此JSON数据包发送给LLM服务器

步骤 2：服务器解密并执行推理

LLM服务器接收到请求后，使用相同的密钥解密Prompt。

# LLM服务器操作

try:
    decrypted_prompt = decrypt_payload(encrypted_request, SHARED_ENCRYPTION_KEY)
    print(f"\n--- 服务端成功解密Prompt ---")
    print(f"解密后的Prompt: {decrypted_prompt}")

    # 模拟LLM推理过程
    llm_response = "项目Alpha-3的邮件核心内容是关于预算调整和Q3上线时间表。"

except ValueError as e:
    print(f"错误：{e}")
    llm_response = "Error: Integrity check failed."

步骤 3：服务器加密Response并发回客户端

推理完成后，服务器必须加密Response以保证传输机密性。

# LLM服务器操作 (加密Response)
encrypted_response = encrypt_payload(llm_response, SHARED_ENCRYPTION_KEY)

# 服务器将 encrypted_response 发回客户端
print("\n--- 服务端加密后的响应数据包 ---")
print(json.dumps(encrypted_response, indent=2))

4. 客户端解密Response

客户端接收到Response后，进行解密和完整性校验。

# 客户端操作
final_response = decrypt_payload(encrypted_response, SHARED_ENCRYPTION_KEY)
print(f"\n--- 客户端最终解密的Response ---")
print(f"模型输出: {final_response}")

完整性校验失败模拟

为了展示完整性校验（HMAC/Tag）的工作原理，我们模拟在传输过程中篡改密文。

# 模拟中间人攻击：在传输过程中，篡改密文
tampered_request = encrypted_request.copy()
# 随机改变密文中的一个字节
tampered_ciphertext_b = b64decode(tampered_request['ciphertext'])
tampered_ciphertext_b = tampered_ciphertext_b[:-1] + b'X'

tampered_request['ciphertext'] = b64encode(tampered_ciphertext_b)

print("\n--- 模拟篡改后的数据包发送给服务器 ---")

# LLM服务器尝试解密篡改后的数据
try:
    decrypt_payload(tampered_request, SHARED_ENCRYPTION_KEY)
except ValueError as e:
    print(f"[成功捕获篡改] 服务器报错: {e}")

输出结果将是: [成功捕获篡改] 服务器报错: 数据完整性校验失败：认证标签无效或数据已被篡改！ 这证明了 AES-GCM 成功保护了数据的完整性，并拒绝了被篡改的请求。