DynamoDBの暗号化データをKMSで復号化してメールで送る(DynamoDB Triggers)

下記で、KMSで暗号化したデータをDynamoDBに登録
(SQS → Lambda → DynamoDB)したので、

Lambda(Python)でデータをKMSで暗号化してSQSに登録する

今回は、下記のDynamoDB TriggersでDynamoDBにデータが登録されたタイミングで
再度LambdaでKMSを用いてデータを復号化し、その内容をメール(SES)してみました。

DynamoDB Triggers (DynamoDB Streams + Lambda)のPythonでの練習

図で書くと、こんな感じです。

DynamoDBのテーブルは次の通りです。

ですので、DynamoDB Streamsで渡されるLambdaのイベントの中身は、
こんな感じになります。

{'Records': [{
    'eventID': '...',
    'eventVersion': '1.0',
    'dynamodb': {
        'Keys': { 'uuid': {'S': 'urn:uuid:...'}},
        'NewImage': {
            'key3': {'S': '…'},
            'key3': {'S': '...'},
            'key1': {'S': '...'},
            'uuid': {'S': '...'}
        },
        'StreamViewType': 'NEW_AND_OLD_IMAGES',
        'SequenceNumber': ’...',
        'SizeBytes': 830
    },
    'awsRegion': 'ap-northeast-1',
    'eventName': 'INSERT',
    'eventSourceARN': '...',
    'eventSource': 'aws:dynamodb'
}]}

Lambdaのコードは、こんな感じです。

import logging
import boto3
import base64
import json

logger = logging.getLogger()
logger.setLevel(logging.INFO)

source = 'suzuki@suz-lab.com'
toAddress = 'suzuki@suz-lab.com'
subject = 'SUZ-LAB TEST'

def lambda_handler(event, context):

    try:
        logger.info(event)
        responses = []

        text = []
        kms = boto3.client('kms')
        for record in event['Records']:
            if record['eventName'] == 'INSERT':
                text.append(kms.decrypt(
                    CiphertextBlob = base64.b64decode(
                        record['dynamodb']['NewImage']['key1']['S']
                    )
                )['Plaintext'])
                text.append(kms.decrypt(
                    CiphertextBlob = base64.b64decode(
                        record['dynamodb']['NewImage']['key2']['S']
                    )
                )['Plaintext'])
                text.append(kms.decrypt(
                    CiphertextBlob = base64.b64decode(
                        record['dynamodb']['NewImage']['key3']['S']
                    )
                )['Plaintext'])

        responses.append(boto3.client(
                'ses',
                region_name = 'us-east-1'
            ).send_email(
                Source = source,
                Destination = {'ToAddresses': [toAddress]},
                Message = {
                    'Subject': {'Data': subject},
                    'Body': {'Text': {'Data': json.dumps(text)}}
                }
            )
        )

        logger.info(responses)
        return responses

    except Exception as e:
        logger.error(e)
        raise e

実際に上記のLambdaがDynamoDB Streamsと連動するように、
イベントの設定をします。

ちなみに、設定項目のBatch sizeとStarting positionは下記の通りです。

• Batch size: Lambdaがストリームから受け取るレコード数の最大値(1 -10000)
• Starting position: Lambdaが読み始めるストリームの場所
• TRIM_HORIZON: 最も古くに保存されたレコードからスタート
• LATEST: 到着した新しいレコードからスタート

当然、Lambdaに関連付けられているIAMロール(lambda_basic_execution)に
SESを操作するポリシーをアタッチする必要もあります。

この状態で、DynamoDBに(暗号化した)データを登録すると、
次のように、復号化されたデータがメールで送信されてきました。

ここまでが、会員登録して、必要な情報を暗号化してDBに登録して、
確認メールを送信する流れって感じでしょうか？