1
# 错误：使用不安全的随机数生成器
2
import random
3
key = random.randbytes(32)  # 伪随机，可预测！
4
# 错误：使用时间戳作为密钥种子
5
import time
6
key = hashlib.sha256(str(time.time()).encode()).digest()  # 可预测！
7

8
# 正确：使用密码学安全随机数生成器
9
import os
10
key = os.urandom(32)  # CSPRNG，不可预测
11

12
# 正确：使用 KMS 生成密钥（密钥不离开 KMS）
13
import boto3
14
kms = boto3.client('kms')
15
response = kms.generate_data_key(KeyId='alias/my-key', KeySpec='AES_256')
16
# 明文密钥只在内存中短暂存在，用后立即销毁

1
# Python: 安全的密钥内存管理
2
import ctypes
3

4
def secure_delete(buffer):
5
    """安全擦除内存中的密钥"""
6
    if isinstance(buffer, bytes):
7
        # 获取 buffer 的内存地址并覆写
8
        ctypes.memset(id(buffer), 0, len(buffer))
9
    del buffer
10

11
# 使用示例
12
data_key = kms.generate_data_key(KeyId='alias/my-key', KeySpec='AES_256')['Plaintext']
13
try:
14
    aesgcm = AESGCM(data_key)
15
    ciphertext = aesgcm.encrypt(nonce, plaintext, None)
16
finally:
17
    secure_delete(data_key)  # 确保密钥被擦除

1
# AWS KMS：信封加密
2
import boto3
3

4
kms = boto3.client('kms')
5

6
# 生成数据密钥（信封加密）
7
response = kms.generate_data_key(
8
    KeyId='alias/my-key',
9
    KeySpec='AES_256'
10
)
11
plaintext_key = response['Plaintext']       # 明文数据密钥（用后即焚）
12
encrypted_key = response['CiphertextBlob']  # 加密的数据密钥（存储）
13

14
# 用数据密钥加密数据
15
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
16
aesgcm = AESGCM(plaintext_key)
17
ciphertext = aesgcm.encrypt(nonce, plaintext, None)
18

19
# 存储加密数据 + 加密的数据密钥
20
save_to_db(ciphertext, encrypted_key)
21

22
# 解密
23
encrypted_key = load_from_db()
24
response = kms.decrypt(CiphertextBlob=encrypted_key)
25
plaintext_key = response['Plaintext']
26
aesgcm = AESGCM(plaintext_key)
27
plaintext = aesgcm.decrypt(nonce, ciphertext, None)

1
# 信封加密完整流程
2
def encrypt_data(plaintext, kms_key_id):
3
    # 1. 从 KMS 获取数据密钥
4
    response = kms.generate_data_key(KeyId=kms_key_id, KeySpec='AES_256')
5
    data_key = response['Plaintext']           # 明文密钥
6
    encrypted_data_key = response['CiphertextBlob']  # 加密密钥
7

8
    # 2. 用数据密钥加密数据
9
    nonce = os.urandom(12)
10
    aesgcm = AESGCM(data_key)
11
    ciphertext = aesgcm.encrypt(nonce, plaintext, None)
12

13
    # 3. 立即销毁明文密钥
14
    del data_key
15

16
    # 4. 返回加密数据 + 加密密钥 + nonce
17
    return {
18
        'ciphertext': ciphertext,
19
        'encrypted_data_key': encrypted_data_key,
20
        'nonce': nonce
21
    }
22

23
def decrypt_data(encrypted):
24
    # 1. 用 KMS 解密数据密钥
25
    response = kms.decrypt(CiphertextBlob=encrypted['encrypted_data_key'])
26
    data_key = response['Plaintext']
27

28
    # 2. 用数据密钥解密数据
29
    aesgcm = AESGCM(data_key)
30
    plaintext = aesgcm.decrypt(encrypted['nonce'], encrypted['ciphertext'], None)
31

32
    # 3. 销毁明文密钥
33
    del data_key
34

35
    return plaintext

1
# AWS CloudHSM：Level 3 HSM 操作示例
2
from cloudhsmcli import HsmClient
3

4
# 连接到 CloudHSM 集群
5
client = HsmClient(
6
    cluster_id='cluster-xxxxx',
7
    username='crypto_user',
8
    password=os.environ['HSM_PASSWORD']
9
)
10

11
# 在 HSM 内生成 AES 密钥（密钥永不离开 HSM）
12
key_handle = client.generate_aes_key(
13
    key_length=256,
14
    label='billing-dek',
15
    attributes=[
16
        'ENCRYPT', 'DECRYPT',    # 允许加密/解密
17
        'WRAP', 'UNWRAP',        # 允许密钥包装
18
        'EXTRACTABLE:FALSE',     # 密钥不可导出
19
        'SENSITIVE:TRUE',        # 敏感密钥
20
    ]
21
)
22

23
# 在 HSM 内执行加密（数据发送到 HSM，密钥不离开）
24
ciphertext = client.encrypt(
25
    key_handle=key_handle,
26
    plaintext=b'sensitive data',
27
    mechanism='AES_GCM'
28
)
29

30
# 在 HSM 内执行解密
31
plaintext = client.decrypt(
32
    key_handle=key_handle,
33
    ciphertext=ciphertext,
34
    mechanism='AES_GCM'
35
)

1
# AWS KMS：自动密钥轮换
2
kms.enable_key_rotation(KeyId='alias/my-key')
3

4
# 手动轮换：创建新密钥，更新别名
5
kms.create_key(Description='New version of my-key')
6
kms.update_alias(AliasName='alias/my-key', TargetKeyId=new_key_id)
7
# 旧密钥仍可解密旧数据，新数据用新密钥加密

1
# 别名轮换流程
2
def rotate_key_via_alias(key_alias):
3
    # 1. 创建新密钥
4
    new_key = kms.create_key(
5
        Description=f'Rotated key for {key_alias}',
6
        Tags=[{'TagKey': 'RotatedFrom', 'TagValue': key_alias}]
7
    )
8
    new_key_id = new_key['KeyMetadata']['KeyId']
9

10
    # 2. 更新别名指向新密钥
11
    kms.update_alias(
12
        AliasName=f'alias/{key_alias}',
13
        TargetKeyId=new_key_id
14
    )
15

16
    # 3. 新数据自动使用新密钥（通过别名）
17
    # 4. 旧密钥保留，用于解密旧数据
18
    print(f'Key rotated: {key_alias} -> {new_key_id}')
19
    print(f'Old key retained for decryption')

1
# 信封加密密钥轮换
2
def rotate_envelope_key(old_kek_id, new_kek_id, encrypted_dek):
3
    # 1. 用旧 KEK 解密 DEK
4
    response = kms.decrypt(CiphertextBlob=encrypted_dek)
5
    plaintext_dek = response['Plaintext']
6

7
    # 2. 用新 KEK 重新加密 DEK
8
    new_encrypted_dek = kms.encrypt(
9
        KeyId=new_kek_id,
10
        Plaintext=plaintext_dek
11
    )['CiphertextBlob']
12

13
    # 3. 销毁明文 DEK
14
    del plaintext_dek
15

16
    # 4. 更新存储：替换加密的 DEK
17
    # 业务数据不需要重新加密！
18
    return new_encrypted_dek

1
# 双写轮换
2
class DualWriteKeyManager:
3
    def __init__(self, old_key_id, new_key_id):
4
        self.old_key_id = old_key_id
5
        self.new_key_id = new_key_id
6
        self.rotation_start = datetime.utcnow()
7

8
    def encrypt(self, plaintext):
9
        """轮换期间：用新密钥加密，但保留旧密钥解密能力"""
10
        # 新数据只用新密钥加密
11
        return envelope_encrypt(plaintext, self.new_key_id)
12

13
    def decrypt(self, encrypted_data, encrypted_dek):
14
        """轮换期间：尝试新密钥解密，失败则用旧密钥"""
15
        try:
16
            return kms.decrypt(CiphertextBlob=encrypted_dek)
17
        except kms.exceptions.NotFoundException:
18
            return kms.decrypt(KeyId=self.old_key_id, CiphertextBlob=encrypted_dek)
19

20
    def is_rotation_complete(self):
21
        """检查是否所有数据都已用新密钥重新加密"""
22
        return all_data_reencrypted(self.new_key_id)

1
# Vault Transit 引擎：加密即服务
2

3
# 1. 启用 Transit 引擎
4
vault secrets enable transit
5

6
# 2. 创建加密密钥
7
vault write -f transit/keys/my-app-key \
8
  type=aes256-gcm96 \
9
  exportable=false \
10
  deletion_allowed=false
11

12
# 3. 加密数据
13
vault write transit/encrypt/my-app-key \
14
  plaintext=$(echo -n "sensitive data" | base64)
15
# => Key            Value
16
#    ciphertext     vault:v1:ABCDEFG...
17

18
# 4. 解密数据
19
vault write transit/decrypt/my-app-key \
20
  ciphertext="vault:v1:ABCDEFG..."
21
# => Key         Value
22
#    plaintext   c2Vuc2l0aXZlIGRhdGE=
23

24
# 5. 轮换密钥
25
vault write -f transit/keys/my-app-key/rotate
26
# 新数据用 v2 密钥加密，旧数据仍可用 v1 解密
27

28
# 6. 查看密钥版本
29
vault read transit/keys/my-app-key
30
# => keys          map[1:map[...] 2:map[...]]
31
#    latest_version 2
32
#    min_version   1

1
# Python: 使用 hvac 库操作 Vault Transit
2
import hvac
3

4
client = hvac.Client(url='https://vault.example.com', token=os.environ['VAULT_TOKEN'])
5

6
# 加密
7
encrypt_response = client.secrets.transit.encrypt_data(
8
    name='my-app-key',
9
    plaintext=base64.b64encode(b'sensitive data').decode()
10
)
11
ciphertext = encrypt_response['data']['ciphertext']  # vault:v1:...
12

13
# 解密
14
decrypt_response = client.secrets.transit.decrypt_data(
15
    name='my-app-key',
16
    ciphertext=ciphertext
17
)
18
plaintext = base64.b64decode(decrypt_response['data']['plaintext'])
19

20
# 批量加密（高性能）
21
batch_data = [
22
    {'plaintext': base64.b64encode(f'record-{i}'.encode()).decode()}
23
    for i in range(1000)
24
]
25
batch_response = client.secrets.transit.encrypt_data(
26
    name='my-app-key',
27
    batch_input=batch_data
28
)

1
# Vault 动态数据库凭据
2
# 配置 PostgreSQL 连接
3
resource "vault_database_secret_backend_connection" "postgres" {
4
  backend       = "database"
5
  name          = "billing-postgres"
6
  allowed_roles = ["billing-readonly", "billing-readwrite"]
7

8
  postgresql {
9
    connection_url = "postgresql://{{username}}:{{password}}@postgres:5432/billing"
10
    username       = "vault_admin"
11
    password       = var.db_password
12
  }
13
}
14

15
# 定义只读角色
16
resource "vault_database_secret_backend_role" "readonly" {
17
  backend = vault_database_secret_backend_connection.postgres.backend
18
  name    = "billing-readonly"
19
  db_name = vault_database_secret_backend_connection.postgres.name
20

21
  default_ttl = "1h"     # 凭据有效期 1 小时
22
  max_ttl     = "4h"     # 最大续期 4 小时
23

24
  creation_statements = [
25
    "CREATE ROLE \"{{name}}\" WITH LOGIN PASSWORD '{{password}}' VALID UNTIL '{{expiration}}';",
26
    "GRANT SELECT ON ALL TABLES IN SCHEMA public TO \"{{name}}\";"
27
  ]
28
  revocation_statements = [
29
    "DROP ROLE IF EXISTS \"{{name}}\";"
30
  ]
31
}

1
# 密钥泄露应急响应流程
2
class KeyLeakResponse:
3
    """密钥泄露应急响应"""
4

5
    def step1_contain(self, leaked_key_id):
6
        """1. 遏制：立即禁用泄露密钥"""
7
        kms.disable_key(KeyId=leaked_key_id)
8
        # 撤销所有使用该密钥的 IAM 策略
9
        revoke_key_policies(leaked_key_id)
10

11
    def step2_assess(self, leaked_key_id):
12
        """2. 评估：确定泄露范围"""
13
        # 查询审计日志：该密钥被谁使用过
14
        usage_log = query_cloudtrail(key_id=leaked_key_id)
15
        # 确定哪些数据用该密钥加密
16
        affected_data = find_encrypted_data(key_id=leaked_key_id)
17
        return usage_log, affected_data
18

19
    def step3_rotate(self, leaked_key_id):
20
        """3. 轮换：创建新密钥"""
21
        new_key_id = kms.create_key(
22
            Description=f'Replacement for leaked key {leaked_key_id}'
23
        )['KeyMetadata']['KeyId']
24
        kms.update_alias(
25
            AliasName=get_alias_for_key(leaked_key_id),
26
            TargetKeyId=new_key_id
27
        )
28
        return new_key_id
29

30
    def step4_reencrypt(self, new_key_id, affected_data):
31
        """4. 重加密：用新密钥重新加密所有受影响数据"""
32
        for data_record in affected_data:
33
            # 解密旧数据
34
            plaintext = decrypt_with_key(data_record, leaked_key_id)
35
            # 用新密钥重新加密
36
            reencrypt_data(data_record, plaintext, new_key_id)
37

38
    def step5_audit(self, leaked_key_id):
39
        """5. 审计：记录事件并改进流程"""
40
        incident_report = {
41
            'leaked_key_id': leaked_key_id,
42
            'detection_time': datetime.utcnow(),
43
            'containment_time': ...,
44
            'affected_data_count': ...,
45
            'root_cause': investigate_leak_cause(),
46
            'prevention_measures': suggest_improvements()
47
        }
48
        save_incident_report(incident_report)