如何优化Lambda检测未启用KMS加密的DynamoDB表的代码逻辑？

Great question! When dealing with DynamoDB CreateTable events, you're right that the sSEDescription field won't be present at all if the table isn't encrypted with KMS—not set to null or an empty value. So the optimal approach is to check for the existence of this field, along with validating its contents when it is present. Let's walk through the best practices and optimized code for this.

Key Observations About the Event Structure #

When a DynamoDB table is created without KMS encryption, the requestParameters section of the CloudWatch Event will not include the sSEDescription key. If encryption is enabled, you'll see the full object with sSEType: "KMS", kMSMasterKeyArn, and status: "ENABLED".

Optimized Python Lambda Implementation #

Here's a robust version of your function that handles edge cases, follows AWS best practices, and avoids common pitfalls like KeyError:

import boto3
import logging
import os

# Configure logging for easier debugging in CloudWatch
logger = logging.getLogger()
logger.setLevel(logging.INFO)

def lambda_handler(event, context):
    # Pull SNS Topic ARN from environment variables (avoids hardcoding)
    sns_topic_arn = os.environ.get("SNS_TOPIC_ARN")
    if not sns_topic_arn:
        logger.error("SNS_TOPIC_ARN environment variable is not configured")
        return

    # Safely extract nested event data to avoid KeyError if fields are missing
    event_detail = event.get("detail", {})
    request_params = event_detail.get("requestParameters", {})
    table_name = request_params.get("tableName", "Unknown Table")

    # Check if the table uses KMS encryption
    sse_description = request_params.get("sSEDescription")
    is_kms_encrypted = False

    if sse_description:
        # Validate both the encryption type and status for full confidence
        sse_type = sse_description.get("sSEType")
        sse_status = sse_description.get("status")
        if sse_type == "KMS" and sse_status == "ENABLED":
            is_kms_encrypted = True

    # Trigger alert if encryption is missing
    if not is_kms_encrypted:
        logger.info(f"Detected unencrypted DynamoDB table: {table_name}")
        sns_client = boto3.client("sns")
        try:
            sns_client.publish(
                TopicArn=sns_topic_arn,
                Subject="Unencrypted DynamoDB Table Created",
                Message=(
                    f"Alert: DynamoDB table '{table_name}' was created without KMS encryption.\n"
                    f"Full event details:\n{event}"
                )
            )
            logger.info(f"Successfully sent alert to SNS topic: {sns_topic_arn}")
        except Exception as e:
            logger.error(f"Failed to send SNS alert: {str(e)}")
    else:
        logger.info(f"DynamoDB table '{table_name}' uses valid KMS encryption—no action needed.")

Key Optimizations Explained #

• Safe Nested Field Access: Using dict.get() instead of direct key access (e.g., event["detail"]) prevents KeyError if the event structure changes or optional fields are missing. This makes your function more resilient to unexpected event formats.
• Environment Variables: Storing the SNS Topic ARN in an environment variable instead of hardcoding it makes the function easier to maintain and deploy across different environments (dev/prod).
• Logging: Adding detailed logs helps you debug issues in CloudWatch Logs, track which tables were flagged, and verify that alerts were sent successfully.
• Robust Encryption Check: Validating both sSEType and status ensures you only flag tables that truly lack KMS encryption (e.g., you won't accidentally flag a table where encryption was configured but failed to enable).
• Error Handling: Wrapping the SNS publish call in a try/except block ensures your Lambda doesn't fail entirely if there's an issue with the SNS service, and logs the error for investigation.

Additional Best Practices #

• Test with Sample Events: Use CloudWatch Event sample payloads (both encrypted and unencrypted table creations) to test your function before deploying it. You can generate these by creating test tables or referencing AWS documentation examples.
• Add Resource Tag Exceptions (Optional): If you have exceptions (e.g., test tables that don't need encryption), extend the function to check for specific tags on the table and skip alerts for those resources.
• Consider AWS Config for Ongoing Monitoring: For a more comprehensive solution, pair this real-time alert with an AWS Config rule to continuously monitor existing tables for encryption status changes.

内容的提问来源于stack exchange，提问作者Judi