Coverage for lambda / drift-detection / handler.py: 100.00%
63 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""
2CloudFormation Drift Detection Lambda Handler.
4This Lambda is invoked on a schedule (default: daily) by an EventBridge rule.
5It initiates drift detection on a CloudFormation stack, polls until the
6detection completes, and publishes an SNS notification if any resources
7have drifted.
9Environment Variables:
10 STACK_NAME: Name of the CloudFormation stack to check for drift
11 SNS_TOPIC_ARN: ARN of the SNS topic to publish drift alerts to
12 REGION: AWS region (populated automatically by Lambda)
13 POLL_INTERVAL_SECONDS: Seconds between detection status polls (default: 10)
14 POLL_MAX_ATTEMPTS: Max poll attempts before giving up (default: 60 = 10 min)
16Invocation:
17 Triggered by an EventBridge rule. The event payload is ignored — the stack
18 name comes from the environment so the Lambda is bound to a specific stack
19 at deploy time (one Lambda per stack).
21Drift detection is asynchronous in CloudFormation:
22 1. DetectStackDrift returns a DriftDetectionId
23 2. Poll DescribeStackDriftDetectionStatus until DetectionStatus is COMPLETE or FAILED
24 3. If drift is detected (StackDriftStatus != IN_SYNC), fetch per-resource drifts
25 via DescribeStackResourceDrifts and publish a summary to SNS
26"""
28from __future__ import annotations
30import json
31import logging
32import os
33import time
34from typing import Any, cast
36import boto3
38# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit
39# Generated at (UTC): 2026-09-01T14:42:56Z
40# Generated from Git commit: 89b000378ed5a912a38c06f4feab2b029936ebcc
41# Flowchart(s) generated from this file:
42# * ``lambda_handler`` -> ``diagrams/code_diagrams/lambda/drift-detection/handler.lambda_handler.html``
43# (PNG: ``diagrams/code_diagrams/lambda/drift-detection/handler.lambda_handler.png``)
44# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``.
45# <pyflowchart-code-diagram> END
48logger = logging.getLogger()
49logger.setLevel(logging.INFO)
51# Poll configuration — kept short to stay well within Lambda's 15-minute max
52DEFAULT_POLL_INTERVAL_SECONDS = 10
53DEFAULT_POLL_MAX_ATTEMPTS = 60 # 10 minutes total at 10-second intervals
55# Terminal detection statuses returned by DescribeStackDriftDetectionStatus
56TERMINAL_DETECTION_STATUSES = {"DETECTION_COMPLETE", "DETECTION_FAILED"}
59def lambda_handler(event: dict[str, Any], context: Any) -> dict[str, Any]:
60 """Run CloudFormation drift detection and publish SNS alerts on drift.
62 Args:
63 event: EventBridge scheduled event (payload unused)
64 context: Lambda context
66 Returns:
67 Dict with drift detection results for logging/debugging
68 """
69 stack_name = os.environ.get("STACK_NAME")
70 sns_topic_arn = os.environ.get("SNS_TOPIC_ARN")
71 region = os.environ.get("REGION") or os.environ.get("AWS_REGION")
73 if not stack_name:
74 raise ValueError("STACK_NAME environment variable is required")
75 if not sns_topic_arn:
76 raise ValueError("SNS_TOPIC_ARN environment variable is required")
78 poll_interval = int(os.environ.get("POLL_INTERVAL_SECONDS", str(DEFAULT_POLL_INTERVAL_SECONDS)))
79 poll_max_attempts = int(os.environ.get("POLL_MAX_ATTEMPTS", str(DEFAULT_POLL_MAX_ATTEMPTS)))
81 logger.info(
82 "Starting drift detection for stack=%s region=%s",
83 stack_name,
84 region,
85 )
87 cfn = boto3.client("cloudformation", region_name=region)
88 sns = boto3.client("sns", region_name=region)
90 # 1. Initiate drift detection
91 detection_id = cfn.detect_stack_drift(StackName=stack_name)["StackDriftDetectionId"]
92 logger.info("Drift detection initiated: detection_id=%s", detection_id)
94 # 2. Poll until detection reaches a terminal state (or we time out)
95 status_response = _poll_detection_status(cfn, detection_id, poll_interval, poll_max_attempts)
96 detection_status = status_response.get("DetectionStatus")
97 stack_drift_status = status_response.get("StackDriftStatus")
99 logger.info(
100 "Drift detection finished: detection_status=%s stack_drift_status=%s",
101 detection_status,
102 stack_drift_status,
103 )
105 # 3. If detection failed, alert on the failure itself
106 if detection_status == "DETECTION_FAILED":
107 reason = status_response.get("DetectionStatusReason", "Unknown detection failure")
108 _publish_alert(
109 sns,
110 sns_topic_arn,
111 subject=f"[GCO] Drift detection FAILED for {stack_name}",
112 message={
113 "stack_name": stack_name,
114 "region": region,
115 "detection_status": detection_status,
116 "reason": reason,
117 },
118 )
119 return {
120 "stack_name": stack_name,
121 "detection_status": detection_status,
122 "stack_drift_status": None,
123 "drift_published": True,
124 }
126 # 4. If stack is in sync, nothing to alert on
127 if stack_drift_status == "IN_SYNC":
128 logger.info("Stack %s is IN_SYNC — no alert published", stack_name)
129 return {
130 "stack_name": stack_name,
131 "detection_status": detection_status,
132 "stack_drift_status": stack_drift_status,
133 "drift_published": False,
134 }
136 # 5. Drift detected — fetch drifted resources and publish SNS alert
137 drifted_resources = _list_drifted_resources(cfn, stack_name)
138 _publish_alert(
139 sns,
140 sns_topic_arn,
141 subject=f"[GCO] Drift detected in stack {stack_name}",
142 message={
143 "stack_name": stack_name,
144 "region": region,
145 "stack_drift_status": stack_drift_status,
146 "drifted_resource_count": len(drifted_resources),
147 "drifted_resources": drifted_resources,
148 },
149 )
151 return {
152 "stack_name": stack_name,
153 "detection_status": detection_status,
154 "stack_drift_status": stack_drift_status,
155 "drifted_resource_count": len(drifted_resources),
156 "drift_published": True,
157 }
160def _poll_detection_status(
161 cfn: Any,
162 detection_id: str,
163 poll_interval: int,
164 max_attempts: int,
165) -> dict[str, Any]:
166 """Poll DescribeStackDriftDetectionStatus until terminal or timeout."""
167 for attempt in range(max_attempts):
168 response = cfn.describe_stack_drift_detection_status(StackDriftDetectionId=detection_id)
169 status = response.get("DetectionStatus")
170 logger.debug("Poll attempt %d: detection_status=%s", attempt + 1, status)
171 if status in TERMINAL_DETECTION_STATUSES:
172 return cast("dict[str, Any]", response)
173 time.sleep(poll_interval)
175 # Timed out — return last response so caller can decide what to do
176 logger.warning(
177 "Drift detection did not complete within %d polls; returning last status",
178 max_attempts,
179 )
180 return cast("dict[str, Any]", response)
183def _list_drifted_resources(cfn: Any, stack_name: str) -> list[dict[str, str]]:
184 """List resources with a drift status other than IN_SYNC.
186 Returns a trimmed representation of each drifted resource suitable for
187 embedding in an SNS message.
188 """
189 drifted: list[dict[str, str]] = []
190 paginator = cfn.get_paginator("describe_stack_resource_drifts")
191 # Filter server-side to reduce response size for large stacks
192 drift_filter = ["MODIFIED", "DELETED", "NOT_CHECKED"]
193 for page in paginator.paginate(
194 StackName=stack_name, StackResourceDriftStatusFilters=drift_filter
195 ):
196 for resource in page.get("StackResourceDrifts", []):
197 drifted.append(
198 {
199 "logical_id": resource.get("LogicalResourceId", ""),
200 "physical_id": resource.get("PhysicalResourceId", ""),
201 "resource_type": resource.get("ResourceType", ""),
202 "drift_status": resource.get("StackResourceDriftStatus", ""),
203 }
204 )
205 return drifted
208def _publish_alert(sns: Any, topic_arn: str, subject: str, message: dict[str, Any]) -> None:
209 """Publish a JSON alert to SNS. Subject is truncated to the 100-char limit."""
210 # SNS subjects have a 100-char max; truncate defensively
211 truncated_subject = subject[:100]
212 sns.publish(
213 TopicArn=topic_arn,
214 Subject=truncated_subject,
215 Message=json.dumps(message, indent=2, default=str),
216 )
217 logger.info("Published drift alert to %s", topic_arn)