はじめに

AWS環境のSNS運用中に、トピックのサブスクライブ先として社外ドメインのアドレス(hogehoge@gmail.comなどの私用アドレス)が追加されている事例を確認しました。

想定外の運用を未然に防ぐため、運用者のIAMポリシーに特定ドメインのサブスクリプション追加を拒否する方法が一番かと思います。
(参考：Amazon SNSによるメール通知で指定ドメイン以外のサブスクリプション作成を拒否するポリシーを作ってみた )

未然に防ぐほか、すでに作成されているトピックを定期監視し、想定外のドメインが登録されていた場合にSNS通知される仕組みを作りました。

全体構成

・EventBridgeスケジュールでLambda関数を定期実行
・アカウント内のSNSトピックを調査し、指定のドメイン以外がサブスクライブ先に登録されていた場合は、通知用のSNSトピックSNSdomainCheck-alertsを呼び出す
・通知用のSNSトピックで登録したメールアドレスに通知

指定のドメインはLambdaの環境変数で管理

指定ドメインと通知用のSNSトピックARNはLambdaの環境変数で管理します。

ドメインの追加/変更があってもコード自体を修正しなくて済む

▼関数

▼環境変数

ALLOWED_DOMAINSの値をカンマ区切りで追加していけばOK。
通知用トピックのARNはそう変わらないのでハードコーディングしてもよかったのですが、なんとなく嫌で環境変数に。

▼実際に通知されたメール

指定ドメインではない、@hoge.co.jpがサブスクライブ先として登録されているため、ルール違反のトピックとして通知された（という想定）。

CloudFormationでのデプロイ

ツールに必要なもの

・EventBridge用のポリシーとロール
・Lambda用のポリシーとロール
・EventBridgeスケジュール
・Lambda関数
・通知用のSNSトピック

図のように、別のアカウント環境からStackSetsとしてデプロイさせる場合はアカウント間でAssumeRoleできるようRoleの作成が必要です。同じアカウント環境でテンプレートからデプロイする場合は必要ありません。

EventBridgeのポリシー

Lambdaの実行権限とパスロールのみ

EventBridge policy

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Action": [
                "lambda:InvokeFunction"
            ],
            "Resource": "arn:aws:lambda:<LambdaARN>",
            "Effect": "Allow"
        },
        {
            "Action": [
                "iam:PassRole"
            ],
            "Resource": "arn:aws:iam::<EventBridgeRoleARN>",
            "Effect": "Allow"
        }
    ]
}

Lambdaのポリシー

SNSのトピックの取得とログ保存の権限のみ
※ログ部分は必要に応じて削除可能

Lambda policy

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Action": [
                "sns:ListTopics",
                "sns:ListSubscriptionsByTopic",
                "sns:Publish"
            ],
            "Resource": "*",
            "Effect": "Allow"
        },
        {
            "Action": [
                "logs:CreateLogGroup",
                "logs:CreateLogStream",
                "logs:PutLogEvents"
            ],
            "Resource": "*",
            "Effect": "Allow"
        }
    ]
}

Lambda関数

Lambda code

ALLOWED_DOMAINS = os.environ['ALLOWED_DOMAINS'].split(',')
NOTIFY_TOPIC_ARN = os.environ['NOTIFY_TOPIC_ARN']

def lambda_handler(event, context):
    sns = boto3.client('sns')
    topics = sns.list_topics()['Topics']
    alert_messages = []

    for topic in topics:
        topic_arn = topic['TopicArn']
        subs = sns.list_subscriptions_by_topic(TopicArn=topic_arn)['Subscriptions']
        
        for sub in subs:
            if sub['Protocol'] in ['email', 'email-json']:
                endpoint = sub['Endpoint'].lower()
                # 許可ドメインに一致しない場合に通知対象とする
                if not any(endpoint.endswith(domain.lower()) for domain in ALLOWED_DOMAINS):
                    alert_messages.append(
                        f"Unexpected email in topic {topic_arn}: {endpoint}"
                    )

    if alert_messages:
        message = "\n".join(alert_messages)
        sns.publish(
            TopicArn=NOTIFY_TOPIC_ARN,
            Subject="SNS Subscription Domain Alert",
            Message=message
        )

Lambdaコードはテンプレートにインラインして、Variablesオプションで環境変数も一緒にデプロイしてしまうのが楽でした。

yaml一部抜粋

SNSdomainCheck:
  Type: AWS::Lambda::Function
  Properties:
    FunctionName: SNSdomain-checker
    Handler: index.lambda_handler
    Role: !GetAtt LambdaExecutionRole.Arn
    Runtime: python3.12
    Code:
      ZipFile: |
          <Lambdaコード>
    Environment:
      Variables:
        ALLOWED_DOMAINS: "@example.co.jp,@example.jp"
        NOTIFY_TOPIC_ARN: <通知用トピックARN>

EvenBridgeのスケジュールと通知用のSNSトピック作成部分のテンプレートは特別なものもないので割愛します

おわりに

設計中に、「試験問題で問われるベストプラクティスみたいだな～」と思っていました。
試験だったら冒頭に紹介した「未然に防ぐ」方法が正解で、この方法は「指定のドメイン以外でもエンドポイントに登録できてしまうため不正解」と解説で書かれていそうです。
試験って実践で役に立つんだなと初めて実感しました。(不正解選択肢ですが)