如何将AWS Encryption SDK对接LocalStack KMS进行本地测试?
解决AWS Encryption SDK指向LocalStack KMS实例的问题
要让AWS Encryption SDK连接到LocalStack的KMS服务(localhost:4566),你需要为StrictAwsKmsMasterKeyProvider配置自定义的KMS客户端参数,或者直接传入已配置好的boto3 KMS客户端。以下是两种可行的修改方案:
方案1:传入已配置的boto3 KMS客户端
你已经创建了指定endpoint_url的boto3 KMS客户端,只需将它传递给StrictAwsKmsMasterKeyProvider的kms_clients参数即可:
import aws_encryption_sdk import boto3 # 初始化指向LocalStack KMS的boto3客户端 kms = boto3.client("kms", endpoint_url="http://localhost:4566") # 创建KMS密钥 create_key_response = kms.create_key() key_id = create_key_response["KeyMetadata"]["KeyId"] # 初始化Encryption SDK客户端 client = aws_encryption_sdk.EncryptionSDKClient() # 传入已配置的KMS客户端到密钥提供者 kms_key_provider = aws_encryption_sdk.StrictAwsKmsMasterKeyProvider( key_ids=[key_id], kms_clients=[kms] # 关键:添加这一行指定使用LocalStack的KMS客户端 ) # 加密文本 my_plaintext = "password123" my_ciphertext, encryptor_header = client.encrypt( source=my_plaintext, key_provider=kms_key_provider ) # 解密文本 decrypted_plaintext, decryptor_header = client.decrypt( source=my_ciphertext, key_provider=kms_key_provider ) # 验证解密结果 assert decrypted_plaintext.decode("utf-8") == my_plaintext print("加解密验证成功")
方案2:通过client_config直接指定端点
如果不想提前创建boto3客户端,可以在初始化StrictAwsKmsMasterKeyProvider时,通过client_config参数直接设置endpoint_url:
import aws_encryption_sdk import boto3 # 初始化指向LocalStack KMS的boto3客户端,用于创建密钥 kms = boto3.client("kms", endpoint_url="http://localhost:4566") create_key_response = kms.create_key() key_id = create_key_response["KeyMetadata"]["KeyId"] # 初始化Encryption SDK客户端 client = aws_encryption_sdk.EncryptionSDKClient() # 配置KMS密钥提供者,直接指定LocalStack端点 kms_key_provider = aws_encryption_sdk.StrictAwsKmsMasterKeyProvider( key_ids=[key_id], client_config={ "endpoint_url": "http://localhost:4566" # 关键:在这里指定LocalStack端点 } ) # 后续加解密逻辑与方案1一致 my_plaintext = "password123" my_ciphertext, encryptor_header = client.encrypt( source=my_plaintext, key_provider=kms_key_provider ) decrypted_plaintext, decryptor_header = client.decrypt( source=my_ciphertext, key_provider=kms_key_provider ) assert decrypted_plaintext.decode("utf-8") == my_plaintext print("加解密验证成功")
注意事项
- 确保LocalStack的KMS服务已在
4566端口正常运行,可通过kms.list_keys()验证连接状态。 - 两种方案核心都是让Encryption SDK使用指向LocalStack的KMS客户端,方案1更适合需要复用boto3客户端的场景。
内容的提问来源于stack exchange,提问作者hwong557
相关产品推荐
相关产品推荐

