[Python] AWS Lambda: Unable to decrypt RDS Activity Stream using KMS...

I have a CloudFormation template which I am using to configure lambda to decrypt AWS RDS database activity stream logs

I used the lambda function from: https://github.com/aws-samples/optimising-aws-kms-decryption-cost-for-das/blob/main/lambda.py

However, in lambda logs, I see InvalidCipherTextException, and I do not understand why I am getting it, because everything seems to be alright.

Lambda logs from CloudWatch:

2025-05-01T18:50:01.777Z
Processing record:
{
"kinesis": {
"kinesisSchemaVersion": "1.0",
"partitionKey": "741a3939-d892-4df5-934a-9479252cb48d",
"sequenceNumber": "49662849271037707186170471169412978102974199784156430338",
"data": "eyJ0eXBlIjogIkRhdGFiYXNlQWN0aXZpdHlNb25pdG9yaW5nUmVjb3JkcyIsInZlcnNpb24iOiAiMS4yIiw...",
"approximateArrivalTimestamp": 1746125400.734
},
"eventSource": "aws:kinesis",
"eventVersion": "1.0",
"eventID": "shardId-000000000000:49662849271037707186170471169412978102974199784156430338",
"eventName": "aws:kinesis:record",
"invokeIdentityArn": "arn:aws:iam::xxx:role/ops-rds-stream-decrypted-logs-DevDecryptLambdaRole-g0cLkmhkldp9",
"awsRegion": "eu-central-1",
"eventSourceARN": "arn:aws:kinesis:eu-central-1:xxx:stream/aws-rds-das-cluster-xxx"
}


Processing record: {"kinesis": {"kinesisSchemaVersion": "1.0", "partitionKey": "741a3939-d892-4df5-934a-9479252cb48d", "sequenceNumber": "49662849271037707186170471169412978102974199784156430338", "data": "eyJ0eXBlIjogIkRhdGFiYXNlQWN0aXZpdHlNb25pdG9yaW5nUmVjb3JkcyIsInZlcnNpb24iOiAiMS4yIiwiZGF0YWJhc2VBY3Rpdml0eUV2ZW50cyI6ICJBWUFEZUNTbG1tUTFOTjE1UHJmZjZXcVBpMjRBWHdBQkFCVmhkM010WTNKNWNIUnZMWEIxWW14cFl5MXJaWGtBUkVGNFEwd3JNSGhJWm5WSVNXaFZSbGhDTTNZeGNWQkVLMHBDY1dkelNIUmpWMmwwZEdwUFREZ3hNSGhEVG10clJGQXZUa00wVWpoMVRWZGFZVEZyUTFKNlVUMDlBQUVBQWtKREFCdEVZWFJoUzJWNUFBQUFnQUFBQUF3Wnkwdk8vWmVhWDRWZ3g3NEFNSlBTVVlrb1dSM0FKV1QyU2VvcUZJbWl4cHB3Y0g5VzhqdkN2bDdBOVkxNHpHdjR5Q081dnBLTUhGYllBVms4eGdJQUFBQUFEQUFFQUFBQUFBQUFBQUFBQUFBQUFBQXNUZ2lCSEhBSXNUQXpScmZFcFZDci8vLy8vd0FBQUFFQUFBQUFBQUFBQUFBQUFBRUFBQUlJeXoxWnBFLyt4OXF6WmZPUW9EYXB4U3dSRkwwbGJaQ0ppeHhkVnB0dmRodUhRNkM3Z3NJUGlBdTU4SURueGt2WHlNTDRYZTBNZnhNMUh5bWdsYmVCS2FWVDlKQkZtWVpxMTRwa3dGUTlob1dxTUVqc3cxMllwN2IzcDZyRmJIaEkyVDIzdU91TnF5cVNLTGloVVdWbE5PN0FqcWZYNjNWeDhaVXdITXNEWWRkQmFvRkJKVm9WRXpIS3RnbE5DMU9KZXlQa2VqMzIzeFFJUG5JY0RBbFRzdkxpYVlzSWtPdm4rYVRXZVdqaWNoZEdhUngrRkg0ZVJxQmp4ODRBK1VIQ21tbG45WG02QWk3eTB4UjM1N0l5YW5LWW40RUdJT2ZiN0NtSmx0YXE5OHYweUpRVUFrL0lFNzFuSkFpRC9WWkRtdUFjRUJLZkl1V0czaXNieFhIQXJ2bWFRVWdYcVdRaXRlelgvS1hlcE9KeG8zYXd0ZXJUYVVnNmtVeUtzZkpyVHRYUEhKdmd4STY0QWp6eUNXb1l4UXRnR0NZZERrK09NVXEvVmVyOUZ1Ky9IcXRuV2krRzJEMUpwU1gwb1VaTlk3bm80N09FUUE5OFE0bnRaeWNseFptRW1Qc2F2V0hIYWtXNHAyajZhbVRoNjd3NVBnR2E0T0dUZzNEdXhOYjJOeWdGNzJTKzFib1VJSWJHVXQ1ZFJYQkRTUmNodHNYQjNMZ2lRazUrOEZOc3kvYWlVa2NXTjIrRys4Ym5RS250bmlUTThISVIvbnphYTI0dmJIR3VrT0V2WE5QMFc3WmhNTDRmNEU1NHJRY1lnQjN5aWJKNGRMcnB3Nms3dGRnaTRyM3NoYWNPVS9jbmlMaVBYSEZjSkhud1g0NXNRQWkxNVFOcVZFTUN6bE82M2NKL3o0T2VyTmN1N1Frd1lNd01JbENiYkNPUEhVNEFaekJsQWpFQXhpTndDNTc5L2QrdXVHTWxhRTE1dlZiSWdDZ1hPN3Zad3czeXFQS25zMUFUNTgzejBMclVaTngvQTlwU0c3RWZBakFYbk16dkkyMVRhYU5Zem1iekRLZ2czMkJ3ajhCdWhESUhSUDYrNmU2aWViNXR0VHMzQ3B4blkvaHUwY0l0Z1VZPSIsImtleSIgOiAiQVFJREFIZ0RIVGN6V0VXejB1MFB0N0VPMkpISHhTTUVIZFd5bk1rUS9xMDZuVGpmMndGbEcvVDYxNW1TRzI0VzRobm1Fa2JnQUFBQWZqQjhCZ2txaGtpRzl3MEJCd2FnYnpCdEFnRUFNR2dHQ1NxR1NJYjNEUUVIQVRBZUJnbGdoa2dCWlFNRUFTNHdFUVFNTWU0Y0djVFhxZmVXRkJTa0FnRVFnRHY4K3ZKcUM1VTFhZEk3YlZLbkE1eGVNS3ByTHIyQ3hiRVRJUnMxaGdrTURudVl5a2gzb2JtUzdISmNFVHJQeThnMDMwNzdieWVrM1dwMXVnPT0ifQ==", "approximateArrivalTimestamp": 1746125400.734}, "eventSource": "aws:kinesis", "eventVersion": "1.0", "eventID": "shardId-000000000000:49662849271037707186170471169412978102974199784156430338", "eventName": "aws:kinesis:record", "invokeIdentityArn": "arn:aws:iam::618550961437:role/ops-rds-stream-decrypted-logs-DevDecryptLambdaRole-g0cLkmhkldp9", "awsRegion": "eu-central-1", "eventSourceARN": "arn:aws:kinesis:eu-central-1:xxx:stream/aws-rds-das-cluster-xxx"}
2025-05-01T18:50:01.777Z
Base64-decoded data: b'
{
"type": "DatabaseActivityMonitoringRecords",
"version": "1.2",
"databaseActivityEvents": "AYADeCSlmmQ1NN15Prff6WqPi24AXwABABVhd3MtY3J5cHRvLXB1YmxpYy1rZXkAREF4Q0wrMHhIZnVISWh...",
"key": "AQIDAHgDHTczWEWz0u0Pt7EO2JHHxSMEHdWynMkQ/q06nTjf2wFlG/T615mSG24W4hnmEkbgAAAAfjB8Bgk..."
}
'



Decoded record data:
{
"type": "DatabaseActivityMonitoringRecords",
"version": "1.2",
"databaseActivityEvents": "AYADeCSlmmQ1NN15Prff6WqPi24AXwABABVhd3MtY3J5cHRvLXB1YmxpYy1rZXkAREF4Q0wrMHhIZnVISWh...",
"key": "AQIDAHgDHTczWEWz0u0Pt7EO2JHHxSMEHdWynMkQ/q06nTjf2wFlG/T615mSG24W4hnmEkbgAAAAfjB8Bgk..."
}



Decoded payload: b'\x01\x80\x03x$\xa5\x9ad54\xddy>\xb7\xdf\xe9j\x8f\x8bn\x00_\x00\x01\x00\x15aws-crypto-public-...'


Decoded data key: b"\x01\x02\x03\x00x\x03\x1d73XE\xb3\xd2\xed\x0f\xb7\xb1\x0e\xd8\x91\xc7\xc5#\x04\x1d\xd5\xb2\x9c\xc9\x10\xfe\xad:\x9d8\xdf\xdb\x01e\x1b\xf4\xfa\xd7\x99\x92\x1bn\x16\xe2..."
2025-05-01T18:50:01.789Z

Decrypting data key: b"\x01\x02\x03\x00x\x03\x1d73XE\xb3\xd2\xed\x0f\xb7\xb1\x0e\xd8\x91\xc7\xc5#\x04\x1d\xd5\xb2\x9c\xc9\x10\xfe\xad:\x9d8\xdf\xdb\x01e\x1b\xf4\xfa\xd7\x99\x92\x1bn\x16\xe2..."


Error processing record: An error occurred (InvalidCiphertextException) when calling the Decrypt operation:


Cloudformation template that I have written:

AWSTemplateFormatVersion: 2010-09-09
Description: Dev-only setup to decrypt RDS DAS logs using KMS and store them in S3

Parameters:
SourceKinesisStreamARN:
Type: String
Default: arn:aws:kinesis:eu-central-1:xxx:stream/aws-rds-das-cluster-xxxx
KMSKeyARN:
Type: String
Default: arn:aws:kms:eu-central-1:xxx:key/xxx
Region:
Type: String
Default: eu-central-1
ResourceId:
Type: String
Default: srv-rds-cluster-databasecluster-xxx # Aurora RDS Cluster ID

Resources:
# S3 Bucket to store decrypted logs
DevDecryptedLogsBucket:
Type: AWS::S3::Bucket
Properties:
BucketName: rds-das-dev-decrypted-logs # New S3 Bucket in dev account
LifecycleConfiguration:
Rules:
- Status: Enabled
Transitions:
- StorageClass: GLACIER
TransitionInDays: 30

# IAM Role for Lambda to read from Kinesis and write to S3
DevDecryptLambdaRole:
Type: AWS::IAM::Role
Properties:
PermissionsBoundary: !Sub arn:aws:iam::${AWS::AccountId}olicy/ScopePermissions
AssumeRolePolicyDocument:
Version: '2012-10-17'
Statement:
- Effect: Allow
Principal:
Service: lambda.amazonaws.com
Action: sts:AssumeRole
Policies:
- PolicyName: DevDecryptLambdaPolicy
PolicyDocument:
Version: '2012-10-17'
Statement:
- Effect: Allow
Action:
- kinesis:GetRecords
- kinesis:GetShardIterator
- kinesisescribeStream
- kinesis:ListStreams
Resource: !Ref SourceKinesisStreamARN
- Effect: Allow
Action:
- s3utObject
Resource: arn:aws:s3:::rds-das-dev-decrypted-logs/*
- Effect: Allow
Action:
- kmsecrypt
Resource: !Ref KMSKeyARN
- Effect: Allow
Action:
- logs:CreateLogGroup
- logs:CreateLogStream
- logsutLogEvents
Resource: '*'

# Lambda function to decrypt the logs and store them in S3
DevDASDecryptLambda:
Type: AWS::Lambda::Function
Properties:
Handler: index.lambda_handler
Runtime: python3.10
Architectures:
- arm64
Timeout: 60
MemorySize: 512
Role: !GetAtt DevDecryptLambdaRole.Arn
Environment:
Variables:
S3_BUCKET: !Ref DevDecryptedLogsBucket # S3 bucket for decrypted logs
KMS_KEY_ID: !Ref KMSKeyARN # KMS key ID for decryption
REGION_NAME: !Ref Region # Region for decryption
RESOURCE_ID: !Ref ResourceId # RDS Aurora cluster ID
STREAM_NAME: !Ref SourceKinesisStreamARN # Kinesis stream name
Code:
ZipFile: |
import json
import zlib
import aws_encryption_sdk
import boto3
import base64
import os
from aws_encryption_sdk import CommitmentPolicy
from aws_encryption_sdk.internal.crypto import WrappingKey
from aws_encryption_sdk.key_providers.raw import RawMasterKeyProvider
from aws_encryption_sdk.identifiers import WrappingAlgorithm, EncryptionKeyType

# Constants for AWS resources and region. Please use environment variables instead
REGION_NAME = os.environ['REGION_NAME']
RESOURCE_ID = os.environ['RESOURCE_ID']
STREAM_NAME = os.environ['STREAM_NAME']

# Initialize the AWS Encryption SDK client with a specific commitment policy
enc_client = aws_encryption_sdk.EncryptionSDKClient(commitment_policy=CommitmentPolicy.FORBID_ENCRYPT_ALLOW_DECRYPT)

# Custom key provider class for raw encryption/decryption operations
class MyRawMasterKeyProvider(RawMasterKeyProvider):
provider_id = "BC" # Custom provider ID

def __new__(cls, *args, **kwargs):
# Overriding the object creation process for proper initialization
obj = super(RawMasterKeyProvider, cls).__new__(cls)
return obj

def __init__(self, plain_key):
# Initializing the parent class and setting up a wrapping key
super().__init__()
self.wrapping_key = WrappingKey(
wrapping_algorithm=WrappingAlgorithm.AES_256_GCM_IV12_TAG16_NO_PADDING,
wrapping_key=plain_key,
wrapping_key_type=EncryptionKeyType.SYMMETRIC)

def _get_raw_key(self, key_id):
# Method to retrieve the raw key; here, it returns the initialized wrapping key
return self.wrapping_key

# Class for caching decrypted data keys using AWS KMS
class KMSDataKeyCache():
def __init__(self, session):
# Initialize the KMS client and a simple dictionary for caching keys
self.kms_client = session.client('kms', region_name=REGION_NAME)
self.key_cache = {}

def getDecrypted(self, data_key_decoded):
# Attempt to retrieve the decrypted key from cache or decrypt it using KMS
if data_key_decoded in self.key_cache:
print("Cache hit for data key.")
return self.key_cache[data_key_decoded]
else:
print(f"Decrypting data key: {data_key_decoded}")
# Decrypt the key using KMS and store it in the cache
data_key_decrypt_result = self.kms_client.decrypt(
CiphertextBlob=data_key_decoded,
EncryptionContext={'aws:rds:dbc-id': RESOURCE_ID})
self.key_cache[data_key_decoded] = data_key_decrypt_result['Plaintext']
print(f"Data key decrypted successfully.")
return data_key_decrypt_result['Plaintext']

# Function to decrypt payload with a provided data key
def decrypt_payload(payload, data_key):
print(f"Decrypting payload with data key: {data_key}") # Debugging: Print the full data key
# Setup the key provider and decrypt the payload
my_key_provider = MyRawMasterKeyProvider(data_key)
my_key_provider.add_master_key("DataKey")
decrypted_plaintext, header = enc_client.decrypt(
source=payload,
materials_manager=aws_encryption_sdk.materials_managers.default.DefaultCryptoMaterialsManager(
master_key_provider=my_key_provider))
print(f"Decrypted payload: {decrypted_plaintext}") # Debugging: Print the full decrypted payload
return decrypted_plaintext

# Function to decrypt and then decompress the payload
def decrypt_decompress(payload, key):
print(f"Decompressing payload...") # Debugging: Indicate decompression
decrypted = decrypt_payload(payload, key)
decompressed = zlib.decompress(decrypted, zlib.MAX_WBITS + 16)
print(f"Decompressed payload: {decompressed}") # Debugging: Print the full decompressed payload
return decompressed

# The main Lambda handler function
def lambda_handler(event, context):
# Initialize a session and the KMS data key cache
print("Initializing KMS Data Key Cache...")
session = boto3.session.Session()
kms_data_key_cache = KMSDataKeyCache(session)

# Process each record in the event
for record in event['Records']:
try:
print(f"Processing record: {json.dumps(record)}") # Debugging: Print the full record
# Decode and parse the incoming data
data = base64.b64decode(record['kinesis']['data'])
print(f"Base64-decoded data: {data}") # Debugging: Print the full base64-decoded data
record_data = json.loads(data)
print(f"Decoded record data: {json.dumps(record_data)}") # Debugging: Print the full decoded record data
payload_decoded = base64.b64decode(record_data['databaseActivityEvents'])
print(f"Decoded payload: {payload_decoded}") # Debugging: Print the full decoded payload
data_key_decoded = base64.b64decode(record_data['key'])
print(f"Decoded data key: {data_key_decoded}") # Debugging: Print the full decoded data key

# Get the decrypted data key from the cache or KMS
decrypted_data_key = kms_data_key_cache.getDecrypted(data_key_decoded)
print(f"Decrypted data key: {decrypted_data_key}") # Debugging: Print the full decrypted data key

# Decrypt and decompress the payload
decrypted_decompressed_payload = decrypt_decompress(payload_decoded, decrypted_data_key)
plaintext = decrypted_decompressed_payload.decode('utf-8')
print(f"Decrypted and decompressed payload: {plaintext}") # Debugging: Print the full plaintext

# Load the JSON events and log them
events = json.loads(plaintext)
print("Processed events:", events) # Debugging: Print the full events processed

except Exception as e:
# Log any errors encountered during processing
print(f"Error processing record: {str(e)}")

# Return a success status code and message
return {
'statusCode': 200,
'body': json.dumps('Processing Complete')
}

# Event source mapping to trigger the Lambda function from Kinesis stream
DevDecryptLambdaEventSource:
Type: AWS::Lambda::EventSourceMapping
Properties:
BatchSize: 100
EventSourceArn: !Ref SourceKinesisStreamARN
FunctionName: !Ref DevDASDecryptLambda
StartingPosition: LATEST

Outputs:
DevDecryptedLogsBucketName:
Description: The S3 bucket where decrypted logs will be stored
Value: rds-das-dev-decrypted-logs

Continue reading...