Python应用的 secrets 管理：集成HashiCorp Vault或AWS KMS的底层机制 - 智猿学院-前后端，数据库，人工智能，云计算等领域前沿技术讲座

Python 应用的 Secrets 管理：集成 HashiCorp Vault 或 AWS KMS 的底层机制

大家好！今天我们要深入探讨一个对于任何认真对待安全的 Python 应用来说都至关重要的话题：Secrets 管理。更具体地说，我们将研究如何将 HashiCorp Vault 或 AWS KMS 这两个强大的工具集成到我们的 Python 应用中，以安全地存储和检索敏感信息。

为什么需要 Secrets 管理？

在深入技术细节之前，我们先来明确为什么需要专门的 Secrets 管理解决方案。在开发过程中，我们经常需要在代码中使用各种敏感信息，例如数据库密码、API 密钥、SSH 密钥、证书等等。如果将这些信息直接硬编码到代码中，或者存储在配置文件中，会带来严重的风险：

代码泄露风险： 代码库可能会被意外泄露，例如上传到公共代码仓库，导致所有 Secrets 暴露。
权限提升风险： 如果攻击者获得了对应用的访问权限，就可以轻松地获取所有的 Secrets，从而提升权限并进行恶意操作。
审计困难： 很难追踪谁在何时访问了哪些 Secrets。
维护困难： 更改 Secrets 需要修改代码或配置文件，并重新部署应用，效率低下且容易出错。

因此，我们需要一种安全、集中化、可审计的方式来管理 Secrets，这就是 Secrets 管理解决方案的价值所在。

HashiCorp Vault 简介

HashiCorp Vault 是一个用于安全地存储和访问 Secrets 的工具。它可以安全地存储任何类型的 Secrets，例如密码、API 密钥、证书等。Vault 提供了细粒度的访问控制策略，可以限制哪些用户或应用可以访问哪些 Secrets。Vault 还可以进行审计，记录所有 Secrets 的访问行为。

Vault 的核心概念：

Secrets Engine： 用于存储和生成 Secrets 的组件。Vault 支持多种 Secrets Engine，例如 Key/Value Secrets Engine、Database Secrets Engine、PKI Secrets Engine 等。
Authentication Methods： 用于验证客户端身份的组件。Vault 支持多种 Authentication Methods，例如 Token、Username/Password、AppRole、Kubernetes、AWS IAM 等。
Policies： 用于定义访问控制策略的组件。Policies 决定了哪些用户或应用可以访问哪些 Secrets。
Audit Log： 用于记录所有 Secrets 访问行为的日志。

AWS KMS 简介

AWS KMS (Key Management Service) 是 AWS 提供的一项托管式密钥管理服务。它允许您安全地创建、存储和控制用于加密数据的密钥。KMS 主要用于加密静态数据和动态数据，但也可以用于存储和检索少量 Secrets。

KMS 的核心概念：

Customer Master Key (CMK)： 用于加密数据的顶级密钥。CMK 可以由 AWS 管理，也可以由用户自定义管理。
Envelope Encryption： 一种使用 CMK 加密数据密钥，然后使用数据密钥加密数据的技术。
Grant： 允许特定 AWS 账户或 IAM 用户使用 CMK 的权限。

集成 HashiCorp Vault 的底层机制

我们将探讨如何使用 hvac 这个 Python 库与 Vault 进行交互。hvac 是一个官方维护的 Vault 客户端库，提供了简洁易用的 API。

1. 安装 hvac:

pip install hvac

2. 配置 Vault 客户端:

import hvac
import os

# Vault 地址和 Token，可以通过环境变量获取，避免硬编码
vault_address = os.environ.get('VAULT_ADDR')
vault_token = os.environ.get('VAULT_TOKEN')

client = hvac.Client(url=vault_address, token=vault_token)

# 检查 Vault 是否可用
if not client.sys.is_initialized().get('initialized'):
    print("Vault is not initialized!")
    exit(1)

if not client.sys.is_sealed().get('sealed'):
    print("Vault is not sealed!")
else:
    print("Vault is sealed! Please unseal it.")
    exit(1)

3. 认证方式:

Vault 支持多种认证方式，常见的有：

Token: 最简单的认证方式，直接使用一个 Token 进行认证。
AppRole: 适用于应用程序的认证方式，通过 Role ID 和 Secret ID 进行认证。
Kubernetes: 适用于在 Kubernetes 集群中运行的应用程序的认证方式，通过 Service Account Token 进行认证。
AWS IAM: 适用于在 AWS 环境中运行的应用程序的认证方式，通过 IAM 角色进行认证。

例子：使用 AppRole 进行认证:

首先，需要在 Vault 中创建一个 AppRole：

vault auth enable approle
vault write auth/approle/role/my-app role_id=my-app
vault read auth/approle/role/my-app/secret-id

然后，在 Python 代码中使用 Role ID 和 Secret ID 进行认证：

import hvac
import os

vault_address = os.environ.get('VAULT_ADDR')

client = hvac.Client(url=vault_address)

role_id = os.environ.get('APPROLE_ROLE_ID')
secret_id = os.environ.get('APPROLE_SECRET_ID')

try:
    response = client.auth.approle.login(role_id=role_id, secret_id=secret_id)
    client.token = response['auth']['client_token']
    print("Successfully authenticated with AppRole!")
except Exception as e:
    print(f"Error authenticating with AppRole: {e}")
    exit(1)

4. 存储和检索 Secrets:

使用 Key/Value Secrets Engine 存储和检索 Secrets：

# 存储 Secret
secret_path = 'secret/data/my-app/db-password'
secret_data = {'data': {'password': 'my-secret-password'}}
client.secrets.kv.v2.create_or_update_secret(path=secret_path, secret=secret_data)

# 检索 Secret
read_response = client.secrets.kv.v2.read_secret_version(path=secret_path)
password = read_response['data']['data']['password']
print(f"Database password: {password}")

5. 错误处理:

在与 Vault 交互时，需要处理各种错误，例如网络错误、认证错误、权限错误等。hvac 库提供了丰富的异常类型，可以方便地进行错误处理。

try:
    read_response = client.secrets.kv.v2.read_secret_version(path=secret_path)
    password = read_response['data']['data']['password']
    print(f"Database password: {password}")
except hvac.exceptions.Forbidden as e:
    print(f"Permission denied: {e}")
except hvac.exceptions.InvalidPath as e:
    print(f"Secret path not found: {e}")
except Exception as e:
    print(f"An unexpected error occurred: {e}")

总结：使用 hvac 库与 Vault 交互，需要配置客户端，选择合适的认证方式，并进行错误处理。

集成 AWS KMS 的底层机制

我们将探讨如何使用 boto3 这个 AWS 官方 SDK 与 KMS 进行交互。

1. 安装 boto3:

pip install boto3

2. 配置 AWS 凭证:

boto3 库会尝试从以下位置获取 AWS 凭证：

环境变量：AWS_ACCESS_KEY_ID 和 AWS_SECRET_ACCESS_KEY
IAM 角色：如果应用运行在 EC2 实例或其他 AWS 服务上，可以配置 IAM 角色来提供凭证。
AWS 配置文件：~/.aws/credentials

3. 创建 KMS 客户端:

import boto3
import os

# KMS 区域，可以通过环境变量获取
kms_region = os.environ.get('AWS_REGION')

kms_client = boto3.client('kms', region_name=kms_region)

4. 加密 Secrets:

# KMS 密钥 ID，可以通过环境变量获取
key_id = os.environ.get('KMS_KEY_ID')

# 要加密的 Secret
secret = 'my-secret-password'

# 使用 KMS 加密 Secret
response = kms_client.encrypt(KeyId=key_id, Plaintext=secret.encode('utf-8'))
ciphertext = response['CiphertextBlob']

print(f"Ciphertext: {ciphertext.hex()}")

5. 解密 Secrets:

# 使用 KMS 解密 Secret
response = kms_client.decrypt(CiphertextBlob=ciphertext)
plaintext = response['Plaintext'].decode('utf-8')

print(f"Plaintext: {plaintext}")

6. 使用 Envelope Encryption:

Envelope Encryption 是一种最佳实践，可以提高安全性。

生成数据密钥：

response = kms_client.generate_data_key(KeyId=key_id, NumberOfBytes=32)
data_key = response['CiphertextBlob']
plaintext_data_key = response['Plaintext']

print(f"Data key (encrypted): {data_key.hex()}")

使用数据密钥加密 Secret：

可以使用 Python 的 cryptography 库或其他加密库来使用数据密钥加密 Secret。

from cryptography.fernet import Fernet

f = Fernet(plaintext_data_key)
encrypted_secret = f.encrypt(secret.encode('utf-8'))

print(f"Secret (encrypted): {encrypted_secret.decode('utf-8')}")

解密 Secret：

# 使用 KMS 解密数据密钥
response = kms_client.decrypt(CiphertextBlob=data_key)
decrypted_data_key = response['Plaintext']

# 使用解密后的数据密钥解密 Secret
f = Fernet(decrypted_data_key)
decrypted_secret = f.decrypt(encrypted_secret).decode('utf-8')

print(f"Secret (decrypted): {decrypted_secret}")

7. 错误处理:

与 KMS 交互时，也需要处理各种错误，例如凭证错误、权限错误、密钥不存在错误等。boto3 库会抛出相应的异常，可以方便地进行错误处理。

try:
    response = kms_client.encrypt(KeyId=key_id, Plaintext=secret.encode('utf-8'))
    ciphertext = response['CiphertextBlob']
    print(f"Ciphertext: {ciphertext.hex()}")
except kms_client.exceptions.NotFoundException as e:
    print(f"KMS key not found: {e}")
except Exception as e:
    print(f"An unexpected error occurred: {e}")

总结：使用 boto3 库与 KMS 交互，需要配置 AWS 凭证，创建 KMS 客户端，并进行错误处理。 Envelope Encryption 是一种提高安全性的最佳实践。

对比 Vault 和 KMS

特性	HashiCorp Vault	AWS KMS
用途	通用的 Secrets 管理平台，适用于各种类型的 Secrets，例如密码、API 密钥、证书等。	主要用于加密数据，但也支持存储和检索少量 Secrets。
存储容量	理论上没有限制，可以存储大量的 Secrets。	适用于存储少量 Secrets，例如数据库密码。对于大量 Secrets，建议使用其他存储服务，例如 AWS Secrets Manager 或 AWS Parameter Store。
访问控制	提供细粒度的访问控制策略，可以限制哪些用户或应用可以访问哪些 Secrets。	通过 IAM 角色和 Grant 进行访问控制。
审计	提供完整的审计日志，记录所有 Secrets 的访问行为。	提供审计日志，记录密钥的使用情况。
集成	可以与各种应用和服务集成，例如 Kubernetes、AWS、Azure、GCP 等。	与 AWS 服务集成良好，例如 EC2、S3、RDS 等。
复杂性	相对复杂，需要部署和维护 Vault 集群。	相对简单，只需要配置 AWS 凭证即可使用。
成本	需要考虑 Vault 集群的部署和维护成本。	按密钥的使用次数收费。

如何选择？

如果需要一个通用的 Secrets 管理平台，并且需要细粒度的访问控制和完整的审计日志，那么 HashiCorp Vault 是一个不错的选择。
如果主要用于加密数据，并且只需要存储和检索少量 Secrets，那么 AWS KMS 是一个不错的选择。
如果应用运行在 AWS 环境中，并且需要与 AWS 服务集成，那么 AWS KMS 是一个不错的选择。

最佳实践

不要将 Secrets 硬编码到代码中或存储在配置文件中。
使用环境变量或配置文件来配置 Vault 或 KMS 客户端。
使用细粒度的访问控制策略，限制哪些用户或应用可以访问哪些 Secrets。
定期轮换 Secrets。
启用审计日志，并定期审查审计日志。
使用 Envelope Encryption 来提高安全性。
使用自动化工具来管理 Secrets。

代码示例：整合 Vault 和 KMS

以下代码示例演示了如何根据环境变量来选择使用 Vault 或 KMS 来存储和检索 Secrets。

import os
import hvac
import boto3
from cryptography.fernet import Fernet

def get_secret(secret_path):
    """
    Retrieves a secret from either Vault or KMS based on environment variables.
    """
    secrets_backend = os.environ.get('SECRETS_BACKEND', 'vault').lower()

    if secrets_backend == 'vault':
        return get_secret_from_vault(secret_path)
    elif secrets_backend == 'kms':
        return get_secret_from_kms(secret_path)
    else:
        raise ValueError(f"Invalid SECRETS_BACKEND: {secrets_backend}. Must be 'vault' or 'kms'.")

def get_secret_from_vault(secret_path):
    """
    Retrieves a secret from HashiCorp Vault.
    """
    vault_address = os.environ.get('VAULT_ADDR')
    vault_token = os.environ.get('VAULT_TOKEN')

    client = hvac.Client(url=vault_address, token=vault_token)

    try:
        read_response = client.secrets.kv.v2.read_secret_version(path=secret_path)
        return read_response['data']['data']['password']  # Adjust key based on your Vault secret structure
    except hvac.exceptions.Forbidden as e:
        print(f"Permission denied: {e}")
        raise
    except hvac.exceptions.InvalidPath as e:
        print(f"Secret path not found: {e}")
        raise
    except Exception as e:
        print(f"An unexpected error occurred: {e}")
        raise

def get_secret_from_kms(secret_path):
    """
    Retrieves a secret from AWS KMS. This assumes the secret is an *encrypted* string
    stored as an environment variable whose name is `secret_path`.
    """

    kms_region = os.environ.get('AWS_REGION')
    kms_client = boto3.client('kms', region_name=kms_region)
    encrypted_secret = os.environ.get(secret_path)

    if not encrypted_secret:
        raise ValueError(f"Encrypted secret not found in environment variable: {secret_path}")

    try:
        response = kms_client.decrypt(CiphertextBlob=bytes.fromhex(encrypted_secret)) # Assumes hex encoded ciphertext
        return response['Plaintext'].decode('utf-8')
    except Exception as e:
        print(f"Error decrypting secret with KMS: {e}")
        raise

# Example usage:
if __name__ == "__main__":
    try:
        db_password = get_secret('my-app/db-password')  # Assumes Vault path, or KMS env var
        print(f"Database password: {db_password}")
    except Exception as e:
        print(f"Failed to retrieve secret: {e}")

在使用前请确保已经配置了相关的环境变量 (VAULT_ADDR, VAULT_TOKEN, AWS_REGION, SECRETS_BACKEND 等等).

总结：选择合适的工具并遵循最佳实践，可以有效地保护应用程序中的敏感信息

我们讨论了 Python 应用中 Secrets 管理的重要性，并深入研究了如何集成 HashiCorp Vault 和 AWS KMS 这两个强大的工具。掌握这些技术，并遵循最佳实践，可以显著提高应用程序的安全性，并降低敏感信息泄露的风险。

更多IT精英技术系列讲座，到智猿学院