数据加密方法推荐

建议采用“AEAD + 信封加密”的方案，结合传输层加密，专门针对数据管道的中间数据（临时文件、分片、缓存、消息负载、临时表等）的低延迟、短生命周期特点。

一、总体方案
- 静态加密（at rest）：信封加密（Envelope Encryption）+ AEAD
  - 数据密钥（DEK）：随机生成的对称密钥，256 位。
  - 加密算法：AES-256-GCM（首选，硬件加速广泛）或 XChaCha20-Poly1305（更适合纯软件环境和长流式数据）。两者均为 AEAD，提供机密性与完整性。
  - 密钥加密密钥（KEK）：由云 KMS（如 AWS KMS、GCP Cloud KMS、Azure Key Vault）管理，用于保护 DEK。
  - 每对象/分片使用唯一 DEK；将密文与加密后的 DEK 一同保存（或使用对象存储的托管加密，如 S3 SSE-KMS）。
- 传输加密（in transit）：TLS 1.2+，推荐 TLS 1.3
  - 启用前向保密（ECDHE），禁用过时套件。
  - 内部服务及代理之间启用双向 TLS（mTLS）。

二、关键实现要点
- 唯一随机 nonce
  - AES-GCM 需使用 96 位（12 字节）唯一 nonce；同一 DEK 下重用 nonce 会造成严重安全风险。
  - 流/分片场景如难以保证 nonce 唯一，考虑使用 AES-GCM-SIV（若生态支持）或 XChaCha20-Poly1305（nonce 长度大，误用韧性更好）。
- AAD（Associated Data）
  - 将不需要保密但需要完整性保护的元数据（如 pipeline 任务 ID、分区键、schema 版本、时间窗口）作为 AAD，以防止元数据被篡改。
- 分片与流式数据
  - 对大文件或分片写入，避免对多个分片复用同一 nonce。可为每分片独立 DEK，或以主 DEK 通过 HKDF 派生子密钥（salt 使用分片 ID），每分片唯一 nonce。
  - 流式场景优先使用 libsodium 的 secretstream（XChaCha20-Poly1305），原生支持连续帧、重放防护与帧边界完整性。
- 密钥管理与轮换
  - 通过 KMS 生成 DEK（或在应用内生成后用 KMS 加密），在完成加密后立即从内存清除明文 DEK。
  - 保留加密后的 DEK（CiphertextBlob），并配合定期轮换 KEK。中间数据生命周期短，可选择到期自动销毁，降低轮换复杂度。
- 性能与兼容
  - 有 AES-NI 的环境优先 AES-GCM；在纯软件或高并发 I/O 下，ChaCha20-Poly1305常更稳定。
  - 压缩优先于加密（若数据可压缩），以降低带宽与存储开销。

三、示例：Python（AWS KMS + AES-GCM 的信封加密）
- 适用：中间数据对象（如一个分片文件、消息负载），每对象独立 DEK，AAD 携带数据血缘信息。

加密
- 依赖：boto3、cryptography
- 步骤：
  1) KMS 生成数据密钥（DEK）
  2) 生成唯一 nonce（12 字节）
  3) 使用 AESGCM 加密，附带 AAD
  4) 清除明文 DEK；保存密文 + 加密 DEK + nonce + 元数据

示例代码（简化）：
- 注意：此示例将 nonce 与密文分开保存；实际可拼接存储。确保每次加密使用不同 nonce。

from boto3 import client
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
import os, json

kms = client("kms", region_name="us-east-1")
kms_key_id = "arn:aws:kms:...:key/..."  # KEK

def encrypt_blob(plaintext: bytes, aad_meta: dict):
    # 1) 生成 DEK（明文+密文）
    resp = kms.generate_data_key(KeyId=kms_key_id, KeySpec="AES_256")
    dek_plain = resp["Plaintext"]          # bytes
    dek_encrypted = resp["CiphertextBlob"] # bytes

    try:
        # 2) 随机 12 字节 nonce（GCM 推荐）
        nonce = os.urandom(12)
        aad = json.dumps(aad_meta, separators=(",", ":")).encode("utf-8")

        # 3) AEAD 加密
        aesgcm = AESGCM(dek_plain)
        ciphertext = aesgcm.encrypt(nonce, plaintext, aad)

        return {
            "ciphertext": ciphertext,
            "nonce": nonce,
            "aad": aad_meta,
            "dek_encrypted": dek_encrypted,  # 与数据一起存储
        }
    finally:
        # 4) 清除明文 DEK
        dek_plain = None

def decrypt_blob(package: dict) -> bytes:
    # 解密 DEK
    dek_plain = kms.decrypt(CiphertextBlob=package["dek_encrypted"])["Plaintext"]
    try:
        aesgcm = AESGCM(dek_plain)
        aad = json.dumps(package["aad"], separators=(",", ":")).encode("utf-8")
        return aesgcm.decrypt(package["nonce"], package["ciphertext"], aad)
    finally:
        dek_plain = None

- 存储建议：将 dek_encrypted、nonce、ciphertext 与必要元数据（AAD）以同一记录或对象元数据保存，便于解密。

四、在常见数据管道组件中的应用
- 对象存储（如 S3、GCS）：可直接采用托管加密（SSE-KMS），或自行实现上述信封加密；为中间数据设置短 TTL 与访问策略。
- 消息队列（如 Kafka、Pub/Sub）：启用 TLS/mTLS；如需载荷保密，生产者端对消息体做 AEAD 加密（每消息或每批次独立 nonce 与 DEK）。
- 临时表与缓存（如临时数据库或分布式缓存）：启用静态加密（数据库 TDE/卷加密）；对高敏感列做应用层 AEAD 加密；控制生命周期与访问权限。
- 分布式计算（Spark/Flink shuffle/spill）：开启本地磁盘或卷加密；若落盘敏感，考虑应用层 AEAD 加密分片并确保 nonce 管理。

五、校验清单
- 每次加密使用唯一 nonce；避免在同一 DEK 下复用。
- 使用 AEAD（AES-GCM 或 XChaCha20-Poly1305），并正确传入 AAD。
- DEK 与密文分离存储：密文保存加密后的 DEK，明文 DEK不落盘。
- 启用 TLS 1.3/mTLS，禁用弱算法，确保前向保密。
- 配置最小权限、审计日志与自动过期策略，匹配中间数据的短生命周期。

该方案在主流云与本地环境均易实施，兼顾安全性、完整性与性能，适合数据管道中间数据的加密需求。

Scope
- Credential types differ in handling:
  - User passwords: never reversible; use memory‑hard hashing.
  - API keys, OAuth tokens, service credentials: require reversible storage for operational use; use authenticated encryption with strong key management.
  - In transit: enforce TLS with modern cipher suites.

Recommendation summary
- Passwords: Argon2id hashing with per‑credential random salt and an application‑level pepper.
- Reversible secrets (API keys, tokens): AES‑256‑GCM (or ChaCha20‑Poly1305) via envelope encryption with a cloud KMS/HSM.
- Transport: TLS 1.2+ (prefer TLS 1.3) with ECDHE and AEAD ciphers.
- Key management: store and rotate keys in KMS/HSM; version keys; audit access.

Details

1) Password storage (non‑reversible)
- Algorithm: Argon2id (preferred), alternatively scrypt or bcrypt.
- Parameters: tune to ~100–250 ms per hash on your production hardware.
  - Example Argon2id: time_cost=2–4, memory_cost=64–256 MiB, parallelism=1–2.
  - Example bcrypt: cost 12–14.
  - Example scrypt: N=2^14–2^16, r=8, p=1–2.
- Salt: unique, random ≥16 bytes per password (library‑generated).
- Pepper: application‑level secret (32+ bytes), stored in KMS/secret manager, applied server‑side (e.g., HMAC or concatenation before hashing).
- Verification: constant‑time comparison; rate‑limit and add account lockouts.

Python example (Argon2id):
- Dependencies: argon2-cffi, secrets

  from argon2 import PasswordHasher
  from argon2.low_level import Type
  import hmac, hashlib, secrets

  # Tuned parameters (example)
  ph = PasswordHasher(
      time_cost=3,
      memory_cost=65536,  # 64 MiB
      parallelism=2,
      hash_len=32,
      type=Type.ID
  )

  PEPPER = secrets.token_bytes(32)  # Load from a secret manager/KMS in production

  def hash_password(password: str) -> str:
      # Apply pepper via HMAC to avoid length leaks
      mac = hmac.new(PEPPER, password.encode('utf-8'), hashlib.sha256).digest()
      return ph.hash(mac)

  def verify_password(password: str, stored_hash: str) -> bool:
      mac = hmac.new(PEPPER, password.encode('utf-8'), hashlib.sha256).digest()
      try:
          ph.verify(stored_hash, mac)
          return True
      except Exception:
          return False

2) Reversible secrets (API keys, OAuth refresh tokens, DB passwords)
- Algorithm: AEAD authenticated encryption: AES‑256‑GCM (preferred) or ChaCha20‑Poly1305.
- Mode: Envelope encryption using a cloud KMS/HSM:
  - Generate a random 256‑bit data encryption key (DEK) per record or per batch.
  - Encrypt the secret with DEK using AES‑GCM with a unique 12‑byte nonce.
  - Encrypt (wrap) the DEK with the KMS master key (KEK).
  - Store: ciphertext, nonce, auth tag, and wrapped DEK alongside the record.
- Associated data (AAD): bind record metadata (e.g., user_id, key_id) to prevent misuse across records.
- Rotation: version KMS keys; rotate DEKs periodically or lazily on read; maintain key_id in metadata.

Python example (AES‑GCM with AWS KMS envelope encryption):
- Dependencies: boto3, cryptography

  import os, json, base64
  from cryptography.hazmat.primitives.ciphers.aead import AESGCM
  import boto3

  kms = boto3.client('kms')

  def encrypt_secret(plaintext: bytes, aad: bytes = b'') -> dict:
      # 1) Generate a random DEK from KMS
      resp = kms.generate_data_key(KeyId='arn:aws:kms:region:acct:key/KEY_ID', KeySpec='AES_256')
      dek_plain = resp['Plaintext']            # bytes
      dek_encrypted = resp['CiphertextBlob']   # bytes (wrapped by KMS)

      try:
          # 2) Encrypt with AES‑GCM
          aesgcm = AESGCM(dek_plain)
          nonce = os.urandom(12)
          ct = aesgcm.encrypt(nonce, plaintext, aad)

          return {
              'ciphertext_b64': base64.b64encode(ct).decode(),
              'nonce_b64': base64.b64encode(nonce).decode(),
              'dek_wrapped_b64': base64.b64encode(dek_encrypted).decode(),
              'aad_b64': base64.b64encode(aad).decode(),
              'kms_key_id': resp['KeyId'],
          }
      finally:
          # Zeroize DEK from memory
          dek_plain = b'\x00' * len(dek_plain)

  def decrypt_secret(record: dict) -> bytes:
      dek_encrypted = base64.b64decode(record['dek_wrapped_b64'])
      nonce = base64.b64decode(record['nonce_b64'])
      ct = base64.b64decode(record['ciphertext_b64'])
      aad = base64.b64decode(record['aad_b64'])

      # 1) Unwrap DEK
      resp = kms.decrypt(CiphertextBlob=dek_encrypted)
      dek_plain = resp['Plaintext']

      try:
          # 2) Decrypt with AES‑GCM
          aesgcm = AESGCM(dek_plain)
          return aesgcm.decrypt(nonce, ct, aad)
      finally:
          dek_plain = b'\x00' * len(dek_plain)

Notes:
- Prefer a single DEK per secret to simplify targeted rotation; per‑table DEK is acceptable with a rekey process.
- Store nonce and ciphertext together; AES‑GCM outputs tag appended to ciphertext by most libraries.
- If you must search by token value, store a separate indexed hash (e.g., HMAC‑SHA256 with a pepper) for lookup; never use deterministic encryption for credentials.

3) Transport security
- Require TLS 1.2+ (prefer TLS 1.3).
- Ciphers: ECDHE with AES‑GCM or ChaCha20‑Poly1305.
- Enable HSTS; disable legacy protocols (TLS 1.0/1.1), RC4, CBC‑only suites.

4) Key management and operations
- Use a managed secret store (AWS KMS + Secrets Manager, GCP Cloud KMS + Secret Manager, Azure Key Vault).
- Enforce IAM/RBAC and audit logs; no raw keys in code or env vars.
- Rotation:
  - Password pepper: rotate via dual‑pepper window (accept old and new during migration).
  - KMS keys: enable automatic rotation; track key versions; rewrap DEKs when rotating KEKs.
- Backup and recovery: export metadata only; keys remain in KMS/HSM.
- Monitoring: detect decryption failures and access anomalies.

Pitfalls to avoid
- Do not encrypt passwords; hash them with a memory‑hard function.
- Do not reuse nonces with AES‑GCM.
- Do not store salts or nonces as secrets; they can be stored with the record.
- Do not hardcode keys or peppers; load from KMS/secret manager at runtime.
- Do not skip AAD; binding metadata reduces misuse.

This approach provides strong confidentiality and integrity for credentials, with operationally sound key management and rotation aligned to data engineering practices.

推荐方案概要
- 核心：采用“信封加密（Envelope Encryption）+ AES-256-GCM（AEAD）+ KMS/HSM”。
- 传输：TLS 1.2/1.3（优先 1.3），仅允许 AEAD 套件。
- 特殊字段：对持卡人数据（PAN）优先采用代币化（Tokenization）；如必须保留可读格式，使用经审计的格式保持加密（FPE，FF1/FF3-1），但优先选用合规的商用/云服务方案。
- 粒度：存储层默认全量加密（SSE/TDE），对敏感列做应用层列级 AEAD 加密（可选确定性加密用于等值查询）。
- 密钥：KMS 管理根密钥（KEK），每数据集/分区/批次生成数据密钥（DEK）；支持轮换与审计。

为何选型
- AES-256-GCM 属于 AEAD，兼顾机密性与完整性（认证标签），性能优（并行化好）。
- 信封加密让数据在不重加密明文的情况下通过“重包裹”实现密钥轮换，运维代价低。
- KMS/HSM 提供硬件或等价级别的密钥保护、审计与权限控制，满足金融数据合规与审计需求。

参考架构与实现要点
1) 密钥管理
- 根密钥（KEK）：存放于云 KMS（如 AWS KMS、GCP KMS、Azure Key Vault）或自建 HSM/Vault。严格的 IAM/ACL 与审计。
- 数据密钥（DEK）：按数据域/表/分区/时间窗生成；只驻留内存，用后销毁；密钥标识与版本在元数据中追踪。
- 轮换策略：
  - KEK：定期（如半年/年）轮换，采用“重包裹（rewrap）”DEK，无需重加密数据。
  - DEK：按时间或数据量阈值轮换（如每日/每 10–100 GB），降低同密钥重用风险。
- AAD（Associated Data）：把租户ID、表名、列名、key_version、分区ID等放入 AAD，加强上下文绑定、防“调包”。

2) 静态数据加密
- 对象存储/数据湖：
  - 最简：启用 SSE-KMS（如 S3 SSE-KMS / GCS CMEK / Azure SSE with CMK）。
  - 更强：客户端信封加密后再落盘（端到端），避免服务端明文暴露。
- 数据库/数据仓库：
  - TDE/平台加密：作为基础防线（磁盘丢失、快照泄露场景），不可替代应用层列级加密。
  - 列级加密：对高敏感字段（PAN、账号、身份证号）使用应用侧 AES-256-GCM；仅在应用侧解密，DB 仅存密文。
  - 确定性加密（可选）：如需等值查询/关联，可对部分字段使用确定性 AEAD（建议 AES-GCM-SIV 或 AES-SIV）。谨慎使用，防频率分析，配合盐/分桶/访问控制。

3) 传输加密
- TLS 1.3 优先；TLS 1.2 时仅允许 ECDHE+AES-GCM 或 CHACHA20-POLY1305。
- 禁用过时算法与压缩；启用证书钉扎和最小权限的双向 TLS（服务间）。

4) PAN 等持卡人数据
- 首选代币化（Vault/云原生 Tokenization 服务），业务仅使用 token；仅保留后四位明文展示。
- 如确需可读格式：采用经评估的 FPE（FF1/FF3-1），并受控使用场景与范围；避免自研密码实现。

5) 审计与运维
- 审计事件：密钥生成/使用/解包失败、解密失败、策略变更。
- 统一元数据：存储 key_id、key_version、aad、nonce、算法、时间戳。
- 备份与导出：备份密文与加密的 DEK；确保目标环境具备相同或已迁移的 KEK 权限。

实现示例：Python 客户端信封加密（AWS KMS + AES-GCM）
说明：示例展示核心流程；生产环境需补充重试、超时、审计、错误处理和密钥缓存。

- 加密
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives import constant_time
import os, json, boto3

kms = boto3.client("kms", region_name="ap-southeast-1")
KEY_ID = "arn:aws:kms:...:key/xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"

def envelope_encrypt(plaintext_bytes: bytes, aad: dict):
    # 1) 生成 DEK（明文与密文）
    resp = kms.generate_data_key(KeyId=KEY_ID, KeySpec="AES_256")
    dek_plain = resp["Plaintext"]
    dek_encrypted = resp["CiphertextBlob"]

    # 2) 使用 DEK 进行 AES-256-GCM 加密
    aesgcm = AESGCM(dek_plain)
    nonce = os.urandom(12)  # 96-bit GCM nonce
    aad_bytes = json.dumps(aad, separators=(",", ":"), sort_keys=True).encode("utf-8")
    ciphertext = aesgcm.encrypt(nonce, plaintext_bytes, aad_bytes)  # 包含 tag

    # 3) 擦除内存中的明文 DEK（Python 无法完全擦除，这里尽可能覆盖）
    del dek_plain

    # 4) 返回存储对象
    return {
        "ciphertext": ciphertext,  # 含 tag
        "nonce": nonce,
        "dek_encrypted": dek_encrypted,
        "aad": aad,
        "alg": "AES-256-GCM",
        "key_id": KEY_ID
    }

- 解密
def envelope_decrypt(obj: dict) -> bytes:
    dek_plain = kms.decrypt(CiphertextBlob=obj["dek_encrypted"], KeyId=obj["key_id"])["Plaintext"]
    aesgcm = AESGCM(dek_plain)
    aad_bytes = json.dumps(obj["aad"], separators=(",", ":"), sort_keys=True).encode("utf-8")
    plaintext = aesgcm.decrypt(obj["nonce"], obj["ciphertext"], aad_bytes)
    del dek_plain
    return plaintext

- 存储结构（示例）
在表/对象的元数据中存储：
- ciphertext: bytes
- nonce: bytes
- dek_encrypted: bytes (由 KMS 保护)
- aad: JSON（tenant_id, table, column, key_version, partition_id）
- key_id, alg, created_at

列级加密集成（PostgreSQL 示例思路）
- 不依赖数据库内置函数进行对称加密（pgcrypto 不提供 AEAD），由应用层完成 AES-GCM。
- 表结构增加密文列与加密元数据列：
  - acct_no_ct (bytea), acct_no_nonce (bytea), acct_no_dek (bytea), acct_no_aad (jsonb), acct_no_alg (text), key_version (int)
- 写入前应用层 envelope_encrypt，读取时 envelope_decrypt。DB 侧仅能基于密文进行有限操作。

等值查询需求的处理
- 对需要 JOIN/WHERE 等值的字段，引入确定性加密索引列：
  - 使用 AES-GCM-SIV/AES-SIV（误用安全，支持确定性），或 KMS/HSM 提供的确定性 AEAD。
  - 仅用于索引列（不可逆），密文列仍存随机化 AEAD 以返回原文。
  - 严格限制访问和结果集最小化，防频率分析；必要时分桶或引入盐策略。

常见陷阱与校验清单
- 禁止重复使用 GCM nonce；用 CSPRNG 生成 96 位随机 nonce。
- 不要自行实现密码学算法或 FPE；优先采用成熟库或合规服务。
- 明文 DEK 只驻留内存，严禁落盘或日志。
- 区分 TDE/SSE 与 应用层列级加密的边界：TDE/SSE 不能保护 DBA/高权限用户在 DB 内部读取明文。
- 完整性与反篡改依赖 AEAD（GCM 的 tag）；文件级可分块加密并对块顺序/元数据做 AAD 绑定。
- 轮换流程预演与回滚策略，确保老数据可按 key_version 解密。

结论
- 对金融数据，采用“信封加密（KMS/HSM）+ AES-256-GCM（AEAD）+ 列级应用侧加密 + 代币化（针对 PAN）+ 全链路 TLS”的组合是兼顾安全性、可运维性与性能的主流做法。该方案适用于数据湖、数据库与数据仓库的统一安全基线，并满足密钥轮换、审计与合规要求。