VPC Reachability Analyzer + Lambda 関数でセキュリティグループ変更を監視~通知する仕組みを実装してみた

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "sts:AssumeRole"
            ],
            "Principal": {
                "Service": [
                    "lambda.amazonaws.com"
                ]
            }
        }
    ]
}

import boto3
import os
import time

def lambda_handler(event, context):
    ec2 = boto3.client('ec2')
    sns = boto3.client('sns')

    # 環境変数から設定値を取得
    analyzer_path_id = os.environ['NETWORK_INSIGHTS_PATH_ID']
    sns_topic_arn = os.environ['SNS_TOPIC_ARN']

    # Reachability Analysis 実行
    response = ec2.start_network_insights_analysis(
        NetworkInsightsPathId=analyzer_path_id
    )
    analysis_id = response['NetworkInsightsAnalysis']['NetworkInsightsAnalysisId']

    print(f"Started Reachability Analysis: {analysis_id}")

    # 結果が出るまでポーリング
    while True:
        result = ec2.describe_network_insights_analyses(
            NetworkInsightsAnalysisIds=[analysis_id]
        )
        analysis = result['NetworkInsightsAnalyses'][0]
        status = analysis['Status']

        print(f"Status: {status}")

        if status in ['succeeded', 'failed']:
            break
        time.sleep(5)

    # 判定結果を評価
    if status == 'succeeded':
        path_found = analysis.get('NetworkPathFound', False)

        if not path_found:
            # 到達不可ケース → SNS 通知
            message = (
                "=== Reachability check failed (unreachable) ===\n"
                f"Path ID: {analyzer_path_id}\n"
                f"Analysis ID: {analysis_id}\n"
                f"Status: {status}\n"
                f"NetworkPathFound: {path_found}"
            )
            sns.publish(
                TopicArn=sns_topic_arn,
                Subject="[ALERT] Security Group change caused connectivity issue",
                Message=message
            )
            print("Unreachable notification sent.")
        else:
            print("Connectivity succeeded. No notification sent.")

    else:
        # 解析自体が失敗したケース
        message = (
            "=== Reachability analysis execution error ===\n"
            f"Path ID: {analyzer_path_id}\n"
            f"Analysis ID: {analysis_id}\n"
            f"Status: {status}"
        )
        sns.publish(
            TopicArn=sns_topic_arn,
            Subject="[ERROR] Reachability Analyzer failed to run",
            Message=message
        )
        print("Execution error notification sent.")

{
  "source": ["aws.ec2"],
  "detail-type": ["AWS API Call via CloudTrail"],
  "detail": {
    "eventSource": ["ec2.amazonaws.com"],
    "eventName": ["AuthorizeSecurityGroupIngress", "RevokeSecurityGroupIngress", "AuthorizeSecurityGroupEgress", "RevokeSecurityGroupEgress"]
  }
}

=== Reachability check failed (unreachable) ===
Path ID: nip-xxxxxxxxxxxx
Analysis ID: nia-xxxxxxxxxxxxx
Status: succeeded
NetworkPathFound: False