Amazon SNS トピックにプライベート HTTP または HTTPS エンドポイントをサブスクライブする方法を教えてください。

所要時間2分

プライベート HTTP または HTTPS エンドポイントを Amazon Simple Notiﬁcation Service (Amazon SNS) トピックにサブスクライブしたいです。

解決策

プライベート HTTP または HTTPS エンドポイントを Amazon SNS トピックにサブスクライブするには、次の手順を実行します。

プライベートエンドポイントと同じ Amazon VPC に Amazon VPC セキュリティグループ LambdaSG を作成します。

Amazon VPC コンソールを開きます。
ナビゲーションペインの [セキュリティ] で [セキュリティグループ] を選択します。次に、[セキュリティグループの作成] を選択します。
[セキュリティグループ名] に LambdaSG と入力します。
VPC には、プライベートエンドポイントが配置された Amazon 仮想プライベートクラウド (VPC) を選択します。
[セキュリティグループの作成] を選択します。

プライベートエンドポイントと同じ Amazon VPC およびサブネット内に Lambda 関数を作成し、LambdaSG セキュリティグループを使用して設定する

Lambda コンソールを開きます。
[関数の作成] を選択します。
[一から作成] を選択します。
[関数名] に、関数の目的を説明する名前を入力します。(Private-endpoint-Amazon-SNS-topic-subscription など。)
[ランタイム] で [Python 3.12] を選択します。
[追加の設定] を選択します。次に、[VPC を有効にする] を選択します。
VPC では、プライベートエンドポイントが配置されたサブネットを選択します。
[関数の作成] を選択します。

プライベートエンドポイントのセキュリティグループルールを編集し、Lambda 関数のセキュリティグループからのインバウンド接続を許可する

Amazon VPC コンソールを開きます。
ナビゲーションペインの [セキュリティ] で [セキュリティグループ] を選択します。
[セキュリティグループ] で、プライベートエンドポイントのセキュリティグループを選択します。
[インバウンドルールを編集] を選択します。
[種類] で [HTTP] または [HTTPS] を選択します。[プロトコル] フィールドと [ポート範囲] フィールドは自動的に入力されます。
[ソース] で [カスタム] を選択します。次に、LambdaSG セキュリティグループを選択します。
**[ルールを保存]**を選択します。

受信した Amazon SNS 通知をプライベートエンドポイントに渡すように Lambda 関数を設定する

Lambda コンソールを開きます。
ナビゲーションペインで、[関数] を選択します。
[関数名] で、以前に作成した関数を選択します。

[コード] の [ソース] で、デフォルトのコードを次のコードに置き換えます。

from __future__ import print_function
import json
import urllib3

http = urllib3.PoolManager()

def lambda_handler(event, context):
    url = "PRIVATE_HTTP/S_ENDPOINT_URL"

    sns_message_payload = event["Records"][0]["Sns"]

    sns_message_headers = {
        "x-amz-sns-message-id": sns_message_payload['MessageId'],
        "x-amz-sns-message-type": sns_message_payload["Type"],
        "x-amz-sns-subscription-arn" : event["Records"][0]["EventSubscriptionArn"],
        "x-amz-sns-topic-arn" : sns_message_payload["TopicArn"]
    }

    try:
        r = http.request('POST', url, headers=sns_message_headers, body=json.dumps(sns_message_payload))
        print(r.data)
    except Exception as e:
        print(e)

重要: url 値は、実際のプライベートエンドポイントの URL に置き換えます。