def __init__(self, scope: Construct, construct_id: str) -> None:
super().__init__(scope, construct_id)
stack = Stack.of(self)
# KMS key shared across CloudWatch log groups, DynamoDB, Lambda env vars,
# and AppConfig hosted configuration content in this app.
# CloudWatch Logs requires the Logs service principal to be granted access
# so it can encrypt data on behalf of the service.
# Note: SSM StringParameter cannot use CMK — CloudFormation does not support
# creating SecureString parameters. AppConfig support arrived later (via
# the kms_key_identifier property on CfnConfigurationProfile), wired below.
self.encryption_key = kms.Key(
self,
"EncryptionKey",
description=f"KMS key for {stack.stack_name} log groups and DynamoDB",
enable_key_rotation=True,
# 90 days is a common compliance-aligned cadence (PCI/HIPAA forks
# default to 90). Rotation is fully managed by AWS — key ID/ARN
# and policies stay constant, prior versions are retained for
# transparent decryption, no dependent redeploys required.
rotation_period=Duration.days(90),
removal_policy=RemovalPolicy.DESTROY,
)
# Confused-deputy guard: scope the Logs service principal grant to
# log-group ARNs in this account+region. See ``grant_logs_service_to_key``
# in ``nag_utils.py`` — three CMKs in this project share the statement.
grant_logs_service_to_key(
self.encryption_key,
region=stack.region,
account=stack.account,
partition=stack.partition,
)
# GuardDuty Lambda Protection inspects Lambda function config, including
# CMK-encrypted env vars. Without this grant the service role is denied
# kms:Decrypt and GuardDuty's coverage of this Lambda is incomplete.
# Scoped via aws:SourceAccount + aws:SourceArn to this account+region's
# detectors only. Applied to the backend CMK only because that's the
# key encrypting the Lambda — the frontend and WAF CMKs encrypt log
# groups and an S3 bucket that GuardDuty does not currently inspect
# through this key.
grant_guardduty_service_to_key(
self.encryption_key,
region=stack.region,
account=stack.account,
partition=stack.partition,
)
# DynamoDB table for Powertools idempotency.
# No table_name set — CDK generates one. Avoids blocking replacement-style
# schema changes and two deployments colliding in one account.
self.idempotency_table = dynamodb.Table(
self,
"IdempotencyTable",
partition_key=dynamodb.Attribute(name="id", type=dynamodb.AttributeType.STRING),
time_to_live_attribute="expiration",
billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
encryption=dynamodb.TableEncryption.CUSTOMER_MANAGED,
encryption_key=self.encryption_key,
contributor_insights_enabled=True,
removal_policy=RemovalPolicy.DESTROY,
point_in_time_recovery_specification=dynamodb.PointInTimeRecoverySpecification(
point_in_time_recovery_enabled=True,
),
)
# SSM parameter for Powertools Parameters.
# parameter_name omitted so CDK auto-generates. Lambda reads the value
# through the GREETING_PARAM_NAME env var, so the name doesn't need to
# be human-memorable.
self.greeting_param = ssm.StringParameter(
self,
"GreetingParameter",
string_value="hello world",
)
# AppConfig for Powertools Feature Flags
self.app_config_app = appconfig.CfnApplication(
self,
"FeatureFlagsApp",
name=f"{stack.stack_name}-features",
)
app_config_env = appconfig.CfnEnvironment(
self,
"FeatureFlagsEnv",
application_id=self.app_config_app.ref,
name=f"{stack.stack_name}-env",
)
# kms_key_identifier CMK-encrypts the hosted configuration content at
# rest in AppConfig. Required because the Lambda's CMK already covers
# logs/DDB/env-vars; pinning AppConfig to the same key keeps the
# auditable encryption surface inside one ARN.
app_config_profile = appconfig.CfnConfigurationProfile(
self,
"FeatureFlagsProfile",
application_id=self.app_config_app.ref,
name=f"{stack.stack_name}-features",
location_uri="hosted",
type="AWS.AppConfig.FeatureFlags",
kms_key_identifier=self.encryption_key.key_arn,
)
# Initial feature flags configuration. CFN registration runs as a
# side effect of construction, so no variable binding is needed.
appconfig.CfnHostedConfigurationVersion(
self,
"FeatureFlagsVersion",
application_id=self.app_config_app.ref,
configuration_profile_id=app_config_profile.ref,
content_type="application/json",
content=(
'{"version":"1","flags":{"enhanced_greeting":'
'{"name":"Enhanced Greeting","default":false}},'
'"values":{"enhanced_greeting":{"enabled":false}}}'
),
)
# Explicit Lambda log group with 1-week retention (implicit group has no retention).
# log_group_name omitted — CDK auto-generates a unique name and wires it into the
# Lambda function via the log_group property below.
lambda_log_group = logs.LogGroup(
self,
"HelloWorldFunctionLogGroup",
encryption_key=self.encryption_key,
retention=logs.RetentionDays.ONE_WEEK,
removal_policy=RemovalPolicy.DESTROY,
)
# Lambda function with automatic dependency bundling.
# environment_encryption pins the env-var encryption to our CMK so the
# security boundary stays inside one key — without it Lambda falls back
# to an AWS-managed key.
self.function = PythonFunction(
self,
"HelloWorldFunction",
runtime=_lambda.Runtime.PYTHON_3_13,
entry="lambda",
index="app.py",
handler="lambda_handler",
architecture=_lambda.Architecture.ARM_64,
memory_size=256,
timeout=Duration.seconds(10),
tracing=_lambda.Tracing.ACTIVE,
log_group=lambda_log_group,
logging_format=_lambda.LoggingFormat.JSON,
environment_encryption=self.encryption_key,
environment={
"POWERTOOLS_SERVICE_NAME": "hello-world",
"POWERTOOLS_METRICS_NAMESPACE": "HelloWorld",
"POWERTOOLS_LOG_LEVEL": "INFO",
"IDEMPOTENCY_TABLE_NAME": self.idempotency_table.table_name,
"GREETING_PARAM_NAME": self.greeting_param.parameter_name,
# Sourcing AppConfig identifiers from the CFN constructs (instead
# of re-formatting f"{stack.stack_name}-...") keeps the Lambda's
# reads in lockstep with the IAM grant below: any future rename
# of the AppConfig resources flows through .name automatically.
"APPCONFIG_APP_NAME": self.app_config_app.name,
"APPCONFIG_ENV_NAME": app_config_env.name,
"APPCONFIG_PROFILE_NAME": app_config_profile.name,
},
)
# Recursive-loop detection. Default is Terminate, but the L2 PythonFunction
# construct doesn't surface this property — set it explicitly on the
# underlying CfnFunction so the posture is visible in code rather than
# implicit in the runtime default.
cast(_lambda.CfnFunction, self.function.node.default_child).recursive_loop = "Terminate"
# Grant permissions
self.idempotency_table.grant_read_write_data(self.function)
self.greeting_param.grant_read(self.function)
# AppConfig least-privilege: both calls authorize against the
# application/environment/configuration ARN. The session token in the
# GetLatestConfiguration request body is opaque request data, not the
# IAM resource — IAM still evaluates the call against this profile ARN.
appconfig_profile_arn = (
f"arn:{stack.partition}:appconfig:{stack.region}:{stack.account}:"
f"application/{self.app_config_app.ref}/"
f"environment/{app_config_env.ref}/"
f"configuration/{app_config_profile.ref}"
)
self.function.add_to_role_policy(
statement=iam.PolicyStatement(
actions=["appconfig:StartConfigurationSession", "appconfig:GetLatestConfiguration"],
resources=[appconfig_profile_arn],
)
)
# Explicit API Gateway access log group with 1-week retention.
# log_group_name omitted — CDK auto-generates and passes it into the
# RestApi via LogGroupLogDestination below.
api_log_group = logs.LogGroup(
self,
"HelloWorldApiAccessLogs",
encryption_key=self.encryption_key,
retention=logs.RetentionDays.ONE_WEEK,
removal_policy=RemovalPolicy.DESTROY,
)
# API Gateway REST API
# cloud_watch_role=True (default) creates an implicit IAM role scoped to
# allow API Gateway to write execution logs to CloudWatch — this is a
# region-level account setting managed by CDK automatically.
self.api = apigw.RestApi(
self,
"HelloWorldApi",
cloud_watch_role=True,
cloud_watch_role_removal_policy=RemovalPolicy.DESTROY,
deploy_options=apigw.StageOptions(
stage_name="Prod",
tracing_enabled=True,
access_log_destination=apigw.LogGroupLogDestination(api_log_group),
access_log_format=apigw.AccessLogFormat.custom(
# Built from typed AccessLogField references — json_with_standard_fields
# only supports 10 fixed fields; custom() is the CDK API for extended formats.
"{"
+ ",".join(
[
f'"requestId":"{apigw.AccessLogField.context_request_id()}"',
f'"accountId":"{apigw.AccessLogField.context_owner_account_id()}"',
f'"apiId":"{apigw.AccessLogField.context_api_id()}"',
f'"stage":"{apigw.AccessLogField.context_stage()}"',
f'"resourcePath":"{apigw.AccessLogField.context_resource_path()}"',
f'"httpMethod":"{apigw.AccessLogField.context_http_method()}"',
f'"protocol":"{apigw.AccessLogField.context_protocol()}"',
f'"status":"{apigw.AccessLogField.context_status()}"',
f'"responseType":"{apigw.AccessLogField.context_error_response_type()}"',
f'"errorMessage":"{apigw.AccessLogField.context_error_message()}"',
f'"requestTime":"{apigw.AccessLogField.context_request_time()}"',
f'"ip":"{apigw.AccessLogField.context_identity_source_ip()}"',
f'"caller":"{apigw.AccessLogField.context_identity_caller()}"',
f'"user":"{apigw.AccessLogField.context_identity_user()}"',
f'"responseLength":"{apigw.AccessLogField.context_response_length()}"',
f'"xrayTraceId":"{apigw.AccessLogField.context_xray_trace_id()}"',
]
)
+ "}"
),
logging_level=apigw.MethodLoggingLevel.INFO,
data_trace_enabled=False,
),
)
hello_resource = self.api.root.add_resource("hello")
hello_resource.add_method("GET", apigw.LambdaIntegration(self.function))
hello_resource.add_cors_preflight(
allow_origins=apigw.Cors.ALL_ORIGINS,
allow_methods=["GET", "OPTIONS"],
# X-Amzn-Trace-Id is required for CloudWatch RUM to propagate the
# client-side X-Ray trace header into the API Gateway → Lambda
# segments so the browser and backend appear on the same trace.
# Idempotency-Key must be allowed by the preflight or browsers will
# block the actual request — the Lambda requires it (returns 400
# without it) so the preflight has to permit it explicitly.
allow_headers=[*apigw.Cors.DEFAULT_HEADERS, "X-Amzn-Trace-Id", "Idempotency-Key"],
)
# Explicit execution log group — API Gateway creates this outside CloudFormation
# when logging_level is enabled. Pre-creating it here transfers ownership to CFN
# so it is deleted on cdk destroy. Name format is fixed by the API Gateway service.
logs.LogGroup(
self,
"HelloWorldApiExecutionLogs",
log_group_name=f"API-Gateway-Execution-Logs_{self.api.rest_api_id}/Prod",
encryption_key=self.encryption_key,
retention=logs.RetentionDays.ONE_WEEK,
removal_policy=RemovalPolicy.DESTROY,
)
self._create_insights_queries(lambda_log_group, api_log_group)
# Application Insights
resource_group = rg.CfnGroup(
self,
"ApplicationResourceGroup",
name=f"ApplicationInsights-{stack.stack_name}",
resource_query=rg.CfnGroup.ResourceQueryProperty(
type="CLOUDFORMATION_STACK_1_0",
),
)
app_insights = appinsights.CfnApplication(
self,
"ApplicationInsightsMonitoring",
resource_group_name=resource_group.name,
auto_configuration_enabled=True,
)
app_insights.add_dependency(resource_group)
# CMK-encrypted log group for the AwsCustomResource provider Lambda.
# Passing log_group= here (instead of log_retention=) avoids the legacy
# LogRetention singleton path and lets us own every log group with our
# CMK — no dangling AWS-managed-key log group left after cdk destroy.
custom_resource_log_group = logs.LogGroup(
self,
"AwsCustomResourceLogGroup",
encryption_key=self.encryption_key,
retention=logs.RetentionDays.ONE_WEEK,
removal_policy=RemovalPolicy.DESTROY,
)
# Custom resource to delete the Application Insights auto-created CloudWatch
# dashboard on stack destroy. Application Insights creates a dashboard named
# after the resource group outside of CloudFormation, so CDK cannot own it
# directly. This Lambda-backed custom resource calls DeleteDashboards at
# destroy time so no dashboard is left behind after cdk destroy.
# Policy is scoped to the exact dashboard ARN — CloudWatch dashboards have
# a known global ARN format and the name is fixed by the resource group.
app_insights_dashboard_arn = (
f"arn:{stack.partition}:cloudwatch::{stack.account}:dashboard/{resource_group.name}"
)
app_insights_dashboard_cleanup = cr.AwsCustomResource(
self,
"AppInsightsDashboardCleanup",
on_delete=cr.AwsSdkCall(
service="CloudWatch",
action="deleteDashboards",
parameters={"DashboardNames": [resource_group.name]},
physical_resource_id=cr.PhysicalResourceId.of(resource_group.name),
),
policy=cr.AwsCustomResourcePolicy.from_sdk_calls(
resources=[app_insights_dashboard_arn],
),
install_latest_aws_sdk=False,
log_group=custom_resource_log_group,
)
# Must run after Application Insights has had a chance to create the dashboard
app_insights_dashboard_cleanup.node.add_dependency(app_insights)
# Monitoring dashboard via cdk-monitoring-constructs
# CloudWatch dashboards are global — scope the name to the stack so
# multiple regional deployments don't collide on the same dashboard name.
monitoring = MonitoringFacade(
self,
"Monitoring",
alarm_factory_defaults={
"actions_enabled": True,
"alarm_name_prefix": stack.stack_name,
},
dashboard_factory=DefaultDashboardFactory(
self,
"MonitoringDashboardFactory",
dashboard_name_prefix=stack.stack_name,
),
)
monitoring.monitor_lambda_function(lambda_function=self.function)
monitoring.monitor_api_gateway(api=self.api)
monitoring.monitor_dynamo_table(table=self.idempotency_table)
# Expose API URL for consumption by the enclosing stack and cross-stack refs
self.api_url = self.api.url
self._add_resource_suppressions(app_insights_dashboard_cleanup)