Skip to content

Instantly share code, notes, and snippets.

@filipeandre
Last active August 13, 2025 16:45
Show Gist options
  • Save filipeandre/8c4d54d2fb12ebd2c1d4cdae2be97a9b to your computer and use it in GitHub Desktop.
Save filipeandre/8c4d54d2fb12ebd2c1d4cdae2be97a9b to your computer and use it in GitHub Desktop.
AWS CloudFormation demonstration with **three stacks** showing end-to-end external KMS key creation, import, and usage,
AWSTemplateFormatVersion: '2010-09-09'
Description: >
Creates an EXTERNAL‑origin KMS key and retrieves import parameters (public key + import token)
via a Lambda-backed custom resource. Stores them in SSM Parameter Store (SecureString) and
outputs base64 values for use by Stack B.
Parameters:
AliasName:
Type: String
Default: 'ext/demo'
KeyDescription:
Type: String
Default: 'External-origin KMS key (imported material)'
ValidityWindowSeconds:
Type: Number
Default: 86400
Description: Validity for import parameters (token & public key). Max ~24h; refresh by updating stack.
WrappingAlgorithm:
Type: String
AllowedValues: [RSAES_OAEP_SHA_256]
Default: RSAES_OAEP_SHA_256
WrappingKeySpec:
Type: String
AllowedValues: [RSA_2048]
Default: RSA_2048
SsmKeyPrefix:
Type: String
Default: '/kms/external-demo'
Resources:
ExternalKey:
Type: AWS::KMS::Key
Properties:
Description: !Ref KeyDescription
Origin: EXTERNAL
Enabled: true
KeyUsage: ENCRYPT_DECRYPT
KeySpec: SYMMETRIC_DEFAULT
KeyPolicy:
Version: '2012-10-17'
Statement:
- Sid: RootAdmin
Effect: Allow
Principal:
AWS: !Sub 'arn:${AWS::Partition}:iam::${AWS::AccountId}:root'
Action: 'kms:*'
Resource: '*'
Alias:
Type: AWS::KMS::Alias
Properties:
AliasName: !Sub 'alias/${AliasName}'
TargetKeyId: !Ref ExternalKey
ParamRole:
Type: AWS::IAM::Role
Properties:
AssumeRolePolicyDocument:
Version: '2012-10-17'
Statement:
- Effect: Allow
Principal:
Service: lambda.amazonaws.com
Action: sts:AssumeRole
Policies:
- PolicyName: kms-import-params
PolicyDocument:
Version: '2012-10-17'
Statement:
- Effect: Allow
Action:
- kms:GetParametersForImport
Resource: !GetAtt ExternalKey.Arn
- Effect: Allow
Action:
- ssm:PutParameter
Resource: '*'
- Effect: Allow
Action:
- logs:CreateLogGroup
- logs:CreateLogStream
- logs:PutLogEvents
Resource: '*'
ParamFn:
Type: AWS::Lambda::Function
Properties:
Runtime: python3.12
Handler: index.handler
Role: !GetAtt ParamRole.Arn
Timeout: 60
Code:
ZipFile: |
import boto3, base64, cfnresponse, os
kms = boto3.client('kms')
ssm = boto3.client('ssm')
def handler(event, context):
try:
props = event['ResourceProperties']
key_id = props['KeyId']
validity = int(props['ValidityWindowSeconds'])
alg = props['WrappingAlgorithm']
spec = props['WrappingKeySpec']
prefix = props['SsmKeyPrefix']
resp = kms.get_parameters_for_import(
KeyId=key_id,
WrappingAlgorithm=alg,
WrappingKeySpec=spec
)
pub_b64 = base64.b64encode(resp['PublicKey']).decode('ascii')
tok_b64 = base64.b64encode(resp['ImportToken']).decode('ascii')
ssm.put_parameter(Name=f"{prefix}/public_key_base64", Value=pub_b64, Type='SecureString', Overwrite=True)
ssm.put_parameter(Name=f"{prefix}/import_token_base64", Value=tok_b64, Type='SecureString', Overwrite=True)
data = {
'PublicKeyBase64': pub_b64,
'ImportTokenBase64': tok_b64,
'ParametersValidTo': resp['ParametersValidTo'].isoformat()
}
cfnresponse.send(event, context, cfnresponse.SUCCESS, data)
except Exception as e:
cfnresponse.send(event, context, cfnresponse.FAILED, {'Error': str(e)})
ParamCustom:
Type: Custom::KmsImportParams
Properties:
ServiceToken: !GetAtt ParamFn.Arn
KeyId: !Ref ExternalKey
ValidityWindowSeconds: !Ref ValidityWindowSeconds
WrappingAlgorithm: !Ref WrappingAlgorithm
WrappingKeySpec: !Ref WrappingKeySpec
SsmKeyPrefix: !Ref SsmKeyPrefix
Outputs:
KeyId:
Value: !Ref ExternalKey
KeyArn:
Value: !GetAtt ExternalKey.Arn
PublicKeyBase64:
Value: !GetAtt ParamCustom.PublicKeyBase64
ImportTokenBase64:
Value: !GetAtt ParamCustom.ImportTokenBase64
ParametersValidTo:
Value: !GetAtt ParamCustom.ParametersValidTo
SsmPublicKey:
Value: !Sub '${SsmKeyPrefix}/public_key_base64'
SsmImportToken:
Value: !Sub '${SsmKeyPrefix}/import_token_base64'
AWSTemplateFormatVersion: '2010-09-09'
Description: >
Imports externally generated key material into an EXTERNAL‑origin KMS key.
You must supply the base64 ImportToken and base64 WrappedKeyMaterial produced using
the PublicKey returned by Stack A. (See notes for an OpenSSL one‑liner.)
Parameters:
KeyId:
Type: String
Description: The target EXTERNAL KMS KeyId from Stack A.
ImportTokenBase64:
Type: String
Description: Base64 import token from Stack A outputs/SSM.
WrappedKeyMaterialBase64:
Type: String
Description: Base64 of the RSA‑OAEP‑SHA256 encrypted 32‑byte key material.
ExpiryDays:
Type: Number
Default: 0
Description: >-
Number of days to keep the key material active (0 = never expire). If >0, the
material will expire automatically.
Resources:
ImportRole:
Type: AWS::IAM::Role
Properties:
AssumeRolePolicyDocument:
Version: '2012-10-17'
Statement:
- Effect: Allow
Principal:
Service: lambda.amazonaws.com
Action: sts:AssumeRole
Policies:
- PolicyName: kms-import
PolicyDocument:
Version: '2012-10-17'
Statement:
- Effect: Allow
Action: kms:ImportKeyMaterial
Resource: !Sub 'arn:${AWS::Partition}:kms:${AWS::Region}:${AWS::AccountId}:key/${KeyId}'
- Effect: Allow
Action:
- logs:CreateLogGroup
- logs:CreateLogStream
- logs:PutLogEvents
Resource: '*'
ImportFn:
Type: AWS::Lambda::Function
Properties:
Runtime: python3.12
Handler: index.handler
Role: !GetAtt ImportRole.Arn
Timeout: 60
Code:
ZipFile: |
import boto3, base64, cfnresponse, datetime
kms = boto3.client('kms')
def handler(event, context):
try:
p = event['ResourceProperties']
key_id = p['KeyId']
token = base64.b64decode(p['ImportTokenBase64'])
wrapped = base64.b64decode(p['WrappedKeyMaterialBase64'])
expiry_days = int(p['ExpiryDays'])
args = {
'KeyId': key_id,
'ImportToken': token,
'EncryptedKeyMaterial': wrapped
}
if expiry_days > 0:
args['ValidTo'] = datetime.datetime.utcnow() + datetime.timedelta(days=expiry_days)
else:
args['ExpirationModel'] = 'KEY_MATERIAL_DOES_NOT_EXPIRE'
resp = kms.import_key_material(**args)
cfnresponse.send(event, context, cfnresponse.SUCCESS, {'Imported': 'true'})
except Exception as e:
cfnresponse.send(event, context, cfnresponse.FAILED, {'Error': str(e)})
ImportCustom:
Type: Custom::KmsImport
Properties:
ServiceToken: !GetAtt ImportFn.Arn
KeyId: !Ref KeyId
ImportTokenBase64: !Ref ImportTokenBase64
WrappedKeyMaterialBase64: !Ref WrappedKeyMaterialBase64
ExpiryDays: !Ref ExpiryDays
Outputs:
Imported:
Value: !GetAtt ImportCustom.Imported
Description: 'true' if succeeded.
Metadata:
Notes: |
Generate a random 32‑byte key and wrap with the public key from Stack A using OpenSSL:
# Save base64 public key to file (from Stack A output PublicKeyBase64)
echo "$PUBLIC_KEY_B64" | base64 -d > public_key.der
# Create 32‑byte random key material
head -c 32 /dev/urandom > key_material.bin
# Wrap with RSA‑OAEP‑SHA256 (public key is DER RSA)
openssl pkeyutl -encrypt -pubin -inkey public_key.der \
-pkeyopt rsa_padding_mode:oaep -pkeyopt rsa_oaep_md:sha256 \
-in key_material.bin -out wrapped.bin
# Prepare values for template parameters
IMPORT_TOKEN_B64=<StackA.ImportTokenBase64>
WRAPPED_MATERIAL_B64=$(base64 -w0 wrapped.bin)
AWSTemplateFormatVersion: '2010-09-09'
Description: >
Demonstrates application‑side (client‑side style) encryption with KMS by calling
kms:Encrypt to produce ciphertext and storing it in S3, and kms:Decrypt to read back.
Two inline Lambda functions: PutObjectEncrypted and GetObjectDecrypted.
Parameters:
KeyArn:
Type: String
Description: ARN of the KMS key (imported/ready) from Stack A/B.
BucketName:
Type: String
Default: ''
Description: Optional explicit bucket name. Leave blank to auto‑generate.
Conditions:
HasBucketName: !Not [!Equals [!Ref BucketName, '']]
Resources:
DemoBucket:
Type: AWS::S3::Bucket
Properties:
BucketName: !If [HasBucketName, !Ref BucketName, !Ref 'AWS::NoValue']
VersioningConfiguration:
Status: Enabled
LambdaRole:
Type: AWS::IAM::Role
Properties:
AssumeRolePolicyDocument:
Version: '2012-10-17'
Statement:
- Effect: Allow
Principal:
Service: lambda.amazonaws.com
Action: sts:AssumeRole
Policies:
- PolicyName: demo-enc
PolicyDocument:
Version: '2012-10-17'
Statement:
- Effect: Allow
Action:
- kms:Encrypt
- kms:Decrypt
Resource: !Ref KeyArn
- Effect: Allow
Action:
- s3:PutObject
- s3:GetObject
Resource: !Sub '${DemoBucket.Arn}/*'
- Effect: Allow
Action:
- logs:CreateLogGroup
- logs:CreateLogStream
- logs:PutLogEvents
Resource: '*'
PutFn:
Type: AWS::Lambda::Function
Properties:
Runtime: python3.12
Handler: index.handler
Role: !GetAtt LambdaRole.Arn
Timeout: 30
Environment:
Variables:
BUCKET: !Ref DemoBucket
KEY_ARN: !Ref KeyArn
Code:
ZipFile: |
import os, boto3, base64, json
kms = boto3.client('kms')
s3 = boto3.client('s3')
def handler(event, context):
# expects JSON {"object_key": "foo.txt", "plaintext": "hello"}
bucket = os.environ['BUCKET']
key_arn = os.environ['KEY_ARN']
body = event if isinstance(event, dict) else {}
plaintext = body.get('plaintext', 'hello from kms encrypt')
object_key = body.get('object_key', 'demo.txt')
enc = kms.encrypt(KeyId=key_arn, Plaintext=plaintext.encode('utf-8'))
ct_b64 = base64.b64encode(enc['CiphertextBlob']).decode('ascii')
s3.put_object(Bucket=bucket, Key=object_key, Body=ct_b64.encode('ascii'), Metadata={'kms-key-arn': key_arn})
return {"bucket": bucket, "key": object_key, "kms_key_arn": key_arn}
GetFn:
Type: AWS::Lambda::Function
Properties:
Runtime: python3.12
Handler: index.handler
Role: !GetAtt LambdaRole.Arn
Timeout: 30
Environment:
Variables:
BUCKET: !Ref DemoBucket
Code:
ZipFile: |
import os, boto3, base64
kms = boto3.client('kms')
s3 = boto3.client('s3')
def handler(event, context):
# expects JSON {"object_key": "foo.txt"}
bucket = os.environ['BUCKET']
object_key = event.get('object_key', 'demo.txt')
obj = s3.get_object(Bucket=bucket, Key=object_key)
ct = base64.b64decode(obj['Body'].read())
pt = kms.decrypt(CiphertextBlob=ct)['Plaintext']
return pt.decode('utf-8')
PutPerm:
Type: AWS::Lambda::Permission
Properties:
Action: lambda:InvokeFunction
FunctionName: !Ref PutFn
Principal: 'apigateway.amazonaws.com'
GetPerm:
Type: AWS::Lambda::Permission
Properties:
Action: lambda:InvokeFunction
FunctionName: !Ref GetFn
Principal: 'apigateway.amazonaws.com'
Outputs:
BucketName:
Value: !Ref DemoBucket
PutFunctionName:
Value: !Ref PutFn
GetFunctionName:
Value: !Ref GetFn
HintInvokePut:
Value: 'aws lambda invoke --function-name <PutFunctionName> --payload {"object_key":"demo.txt","plaintext":"hi"} out.json'
HintInvokeGet:
Value: 'aws lambda invoke --function-name <GetFunctionName> --payload {"object_key":"demo.txt"} out.json'
AWSTemplateFormatVersion: '2010-09-09'
Description: >
Demo of client-side encryption using AWS Encryption SDK (ESDK) with a KMS keyring (generator + optional additional keys),
storing ciphertext in S3 and decrypting on read. Lambda functions run on arm64 / Python 3.12 and include AWS Lambda Powertools (arm64) layer.
Parameters:
KeyArn:
Type: String
Description: Generator KMS key ARN (primary key for the keyring).
AdditionalKeyArnsCsv:
Type: String
Default: ''
Description: Optional comma-separated list of additional KMS key ARNs for multi-keyring (e.g. arn1,arn2).
BucketName:
Type: String
Default: ''
Description: Optional explicit bucket name. Leave blank to auto-generate.
ESDKLayerArn:
Type: String
Default: ''
Description: Optional Lambda Layer ARN that contains the 'aws-encryption-sdk' package for Python 3.12 arm64.
Conditions:
HasBucketName: !Not [!Equals [!Ref BucketName, '']]
HasESDKLayer: !Not [!Equals [!Ref ESDKLayerArn, '']]
Resources:
DemoBucket:
Type: AWS::S3::Bucket
Properties:
BucketName: !If [HasBucketName, !Ref BucketName, !Ref 'AWS::NoValue']
VersioningConfiguration:
Status: Enabled
LambdaRole:
Type: AWS::IAM::Role
Properties:
AssumeRolePolicyDocument:
Version: '2012-10-17'
Statement:
- Effect: Allow
Principal: { Service: lambda.amazonaws.com }
Action: sts:AssumeRole
Policies:
- PolicyName: demo-esdk-enc
PolicyDocument:
Version: '2012-10-17'
Statement:
# ESDK KMS keyring typically uses GenerateDataKey + Decrypt (+ DescribeKey).
# For simplicity in demo we allow these on all keys. In production, scope to specific ARNs.
- Effect: Allow
Action:
- kms:GenerateDataKey
- kms:Decrypt
- kms:DescribeKey
Resource: "*"
- Effect: Allow
Action:
- s3:PutObject
- s3:GetObject
Resource: !Sub '${DemoBucket.Arn}/*'
- Effect: Allow
Action:
- logs:CreateLogGroup
- logs:CreateLogStream
- logs:PutLogEvents
Resource: '*'
PutFn:
Type: AWS::Lambda::Function
Properties:
Runtime: python3.12
Architectures: [ arm64 ]
Handler: index.handler
Role: !GetAtt LambdaRole.Arn
Timeout: 30
Layers:
# AWS Lambda Powertools for Python v3 (Python 3.12, arm64)
- !Sub arn:aws:lambda:${AWS::Region}:017000801446:layer:AWSLambdaPowertoolsPythonV3-python312-arm64:22
# Optional ESDK layer (must contain aws-encryption-sdk for Python 3.12/arm64)
- !If [HasESDKLayer, !Ref ESDKLayerArn, !Ref "AWS::NoValue"]
Environment:
Variables:
BUCKET: !Ref DemoBucket
KEY_ARN: !Ref KeyArn
ADDITIONAL_KEY_ARNS_CSV: !Ref AdditionalKeyArnsCsv
Code:
ZipFile: |
import os, base64, json, boto3
import aws_encryption_sdk
from aws_encryption_sdk.keyrings.aws_kms import AwsKmsKeyring
s3 = boto3.client('s3')
def build_keyring():
generator_key_id = os.environ['KEY_ARN']
addl_csv = os.environ.get('ADDITIONAL_KEY_ARNS_CSV', '').strip()
key_ids = [x.strip() for x in addl_csv.split(',') if x.strip()] if addl_csv else []
return AwsKmsKeyring(generator_key_id=generator_key_id, key_ids=key_ids)
def handler(event, context):
# expects JSON {"object_key": "foo.txt", "plaintext": "hello"}
bucket = os.environ['BUCKET']
body = event if isinstance(event, dict) else {}
plaintext = body.get('plaintext', 'hello from esdk keyring').encode('utf-8')
object_key = body.get('object_key', 'demo.txt')
keyring = build_keyring()
ciphertext, header = aws_encryption_sdk.encrypt(
source=plaintext,
keyring=keyring
)
ct_b64 = base64.b64encode(ciphertext).decode('ascii')
s3.put_object(
Bucket=bucket,
Key=object_key,
Body=ct_b64.encode('ascii'),
Metadata={
'esdk': 'true',
'generator-key-arn': os.environ['KEY_ARN']
}
)
# Return which keys were intended (for visibility)
addl = [k for k in os.environ.get('ADDITIONAL_KEY_ARNS_CSV','').split(',') if k.strip()]
return {"bucket": bucket, "key": object_key, "keyring_keys": [os.environ['KEY_ARN']] + addl}
GetFn:
Type: AWS::Lambda::Function
Properties:
Runtime: python3.12
Architectures: [ arm64 ]
Handler: index.handler
Role: !GetAtt LambdaRole.Arn
Timeout: 30
Layers:
- !Sub arn:aws:lambda:${AWS::Region}:017000801446:layer:AWSLambdaPowertoolsPythonV3-python312-arm64:22
- !If [HasESDKLayer, !Ref ESDKLayerArn, !Ref "AWS::NoValue"]
Environment:
Variables:
BUCKET: !Ref DemoBucket
KEY_ARN: !Ref KeyArn
ADDITIONAL_KEY_ARNS_CSV: !Ref AdditionalKeyArnsCsv
Code:
ZipFile: |
import os, base64, boto3
import aws_encryption_sdk
from aws_encryption_sdk.keyrings.aws_kms import AwsKmsKeyring
s3 = boto3.client('s3')
def build_keyring():
# For decrypt, using the same keyring definition is fine.
generator_key_id = os.environ['KEY_ARN']
addl_csv = os.environ.get('ADDITIONAL_KEY_ARNS_CSV', '').strip()
key_ids = [x.strip() for x in addl_csv.split(',') if x.strip()] if addl_csv else []
return AwsKmsKeyring(generator_key_id=generator_key_id, key_ids=key_ids)
def handler(event, context):
# expects JSON {"object_key": "foo.txt"}
bucket = os.environ['BUCKET']
object_key = event.get('object_key', 'demo.txt')
obj = s3.get_object(Bucket=bucket, Key=object_key)
ct_b64 = obj['Body'].read()
ciphertext = base64.b64decode(ct_b64)
keyring = build_keyring()
plaintext, header = aws_encryption_sdk.decrypt(
source=ciphertext,
keyring=keyring
)
return plaintext.decode('utf-8')
PutPerm:
Type: AWS::Lambda::Permission
Properties:
Action: lambda:InvokeFunction
FunctionName: !Ref PutFn
Principal: 'apigateway.amazonaws.com'
GetPerm:
Type: AWS::Lambda::Permission
Properties:
Action: lambda:InvokeFunction
FunctionName: !Ref GetFn
Principal: 'apigateway.amazonaws.com'
Outputs:
BucketName:
Value: !Ref DemoBucket
PutFunctionName:
Value: !Ref PutFn
GetFunctionName:
Value: !Ref GetFn
HintInvokePut:
Value: 'aws lambda invoke --function-name <PutFunctionName> --payload {"object_key":"demo.txt","plaintext":"hi"} out.json'
HintInvokeGet:
Value: 'aws lambda invoke --function-name <GetFunctionName> --payload {"object_key":"demo.txt"} out.json'
@filipeandre
Copy link
Author

Stack A – Create EXTERNAL KMS key + publish import params

Creates an Origin=EXTERNAL KMS key.
Custom resource fetches PublicKey (DER) + ImportToken and stores them in SSM (and outputs base64 strings).
This mirrors the “KMS side” handing the public wrapping key to the external party.

Stack B – Import key material (from wrapped material + token)
Takes KeyId, ImportTokenBase64, and WrappedKeyMaterialBase64 and performs kms:ImportKeyMaterial.
I included OpenSSL one-liners to generate a 32-byte random key and wrap it with RSA-OAEP-SHA256 using the public key from Stack A (no extra libs or Lambda layers needed).

Stack C – Demo S3 with client-side-style encryption using KMS
Two inline Python Lambdas:
PutObjectEncrypted: calls kms:Encrypt with your key, stores base64 ciphertext to S3.
GetObjectDecrypted: reads from S3, calls kms:Decrypt, returns plaintext.
This demonstrates app-side (client-side style) handling while leveraging KMS directly.

Run order (important)
1- Deploy Stack A (producer of import params).
2- Use the outputs (PublicKeyBase64, ImportTokenBase64) to wrap your key material with the OpenSSL commands shown inside Stack B’s metadata.
3- Deploy Stack B to import that wrapped material into the key.
4- Deploy Stack C with the resulting KeyArn and exercise the upload/download flows.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment