はじめに
普段の開発や運用でメールボックスを常に監視するのは大変ですよね。そこで今回は、エンジニアの必須ツールである**「Slack」にアラートを直接飛ばす仕組み**にアップグレードします！
ただ通知先を変えるだけではありません。今回はAWSアーキテクチャの醍醐味である**「SQS（待ち行列）」と「AWS Lambda（サーバーレスプログラム）」**を間に挟みます。 もし世界中から同時にサイバー攻撃を受けても、システムがパンクすることなく確実に通知を届けてくれる「疎結合（そけつごう）アーキテクチャ」を、実質0円で体験してみましょう！
今回の完成イメージ
- CloudWatchが攻撃を検知し、SNS（通知ハブ）へ発報。
- SNSが**「SQS（キュー）」**にメッセージをポンッと入れる。
- SQSにデータが入ったのをキッカケに、**「Lambda（Pythonプログラム）」**が自動起動。
- LambdaがSlackに「🚨 攻撃検知！」とメッセージを送信！
Slackで「通知受け取り用のURL」を発行する
まずは、Slack側に「ここ宛にメッセージを送ってね」という専用の窓口（Webhook URL）を作ります。
- パソコンのブラウザで Slack API の管理画面 (https://api.slack.com/apps) にアクセスし、右上の 「Create New App」 をクリックします。
- 「From scratch」 を選びます。
- App Name に
WP-Security-Botなど好きな名前を入力し、通知を送りたいワークスペースを選択して 「Create App」 を押します。
- 左メニューから 「Incoming Webhooks」 をクリックし、右上のスイッチを 「On」 にします。
- 画面下のほうにある 「Add New Webhook to Workspace」 をクリックします。
通知を飛ばしたいチャンネル（例：#general や #alert など）を選んで「許可する」を押します。
Webhook URL （https://hooks.slack.com/services/...）が発行されるので、「Copy」ボタンを押してメモ帳などに控えておいてください。
SNSトピック（通知ハブ）を作成する
まずは、CloudWatchからの警報を受け取る「ハブ」を作成します。
- AWSマネジメントコンソールで 「SNS」 と検索し、「Amazon SNS」を開きます。」
- 左側のメニューから 「トピック」 を選び、右上の 「トピックの作成」 をクリックします。
- タイプ: 必ず 「スタンダード」 を選択します。（※FIFOを選ぶとCloudWatchから直接送れなくなります）
画面一番下の 「トピックの作成」 をクリックします。
メトリクスフィルター（攻撃のカウント器）を作成する
次に、Lightsailから送られてきているアクセスログの中から、「攻撃とみなす特定の文字（例：wp-login.php）」を見つけ出してカウントする設定を行います。
- AWSコンソールで 「CloudWatch」 を開きます。
左メニューの「ログ」から 「ロググループ」 を選び、Lightsailのアクセスログ（例：lightsail-wp-access-log）をクリックします。
- 「メトリクスフィルター」 タブを開き、「メトリクスフィルターを作成」 をクリックします。
- フィルターパターン: ログの中から探したいキーワードを入力します。
- 例：
"wp-admin"（※ダブルクォーテーションで囲むと確実です）
- 例：
できる限り、ログの中からパターンをテストすることをおすすめします
- 「次へ」をクリックし、フィルターの詳細を以下のように設定します。
- フィルター名:
Detect-WP-Login-Attack - メトリクス名前空間:
WordPressSecurity（※新しいカテゴリ名を作ります） - メトリクス名:
LoginAccessCount - メトリクス値:
1（※キーワードを1回見つけるごとに「1」カウントするという意味です）
- フィルター名:
- 「メトリクスフィルターを作成」 をクリックします。
CloudWatchアラーム（警報装置）を作成する
最後に、ステップ2で作ったカウント器が「規定の回数」を超えたら、ステップ1で作ったSNSに発報する設定を行います。ここが一番繋ぎ忘れがちなポイントです！
- いま作成したメトリクスフィルターの画面右上に表示される、「アラームを作成」 というボタンをクリックします。
- 【メトリクスと条件の指定】
- 統計:
合計 (Sum)に変更します。 - 期間:
1分または5分に設定します。（テスト時は1分がおすすめです） - 条件:
静的 (Static)＞以上 (Greater/Equal)を選びます。 - しきい値:
5と入力します。（※「1分間に5回アクセスがあったら」という意味になります）
- 統計:
- 【アクションの設定（超重要！）】
- アラーム状態トリガー:
アラーム状態のままにします。 - 通知の送信先:
既存のSNSトピックを選択を選びます。 - その下の「通知の送信先」の検索窓をクリックし、ステップ1で作った
wp-attack-alert-topicを選択します！
- アラーム状態トリガー:
- 「次へ」をクリックします。
- 【名前と説明の追加】
- アラーム名:
WP-Login-Attack-Alarmなどと入力し、「次へ」をクリックします。
- アラーム名:
- 最後に設定内容を確認し、画面一番下の 「アラームの作成」 をクリックします。
💡 動作確認（テスト）のやり方
ここまで設定できたら、CloudWatchとSNSが本当に繋がっているかテストしてみましょう。
- ご自身のWordPressのログインURL（例：
https://polepolelife.com/wp-login.php）をブラウザで開きます。 - F5キー（更新）を5〜6回ほど、パパパッと連打します。
- CloudWatchの「すべてのアラーム」画面を開いて数分待ちます。
- アラームの状態が緑色の「OK」から、赤色の「アラーム状態」 に変われば大成功です！
アラーム状態になった瞬間、設定したSNSトピックに向けて裏側でメッセージが飛んでいます。このSNSにSQSを連携（サブスクライブ）させておけば、見事なピタゴラスイッチの完成です。
メッセージの待合室「SQS」を作成する
大量の通知が来てもパンクしないように、メッセージを一時的に貯めておく「キュー（SQS）」を作ります。
- AWSマネジメントコンソールで 「SQS」 と検索し、「Simple Queue Service」を開きます。
- 右上の 「キューを作成」 をクリックします。
タイプは 「標準」 のままでOKです。
名前 に wp-alert-queue と入力します。
その他の設定はすべて初期状態のままで、一番下の 「キューの作成」 をクリックします。
💡 超重要：SNS（第3回で作ったアラーム）とSQSを繋ぐ
いま作ったSQSの画面に「SNS サブスクリプション」というタブがあります。
「Amazon SNS トピックをサブスクライブ」 をクリックします。
SNSトピック（例：wp-alert-topic）を選んで「保存」を押します。 (※これだけで、CloudWatchのアラームが鳴ると、自動的にこのSQSにメッセージが届くようになります！) SNSトピックなかったら作っておいてください
設定終了したら保存します
自動実行プログラム「AWS Lambda」を作成する
SQSにメッセージが届いたら、それを拾ってSlackに転送するPythonプログラムを作ります。
- AWSの検索バーで 「Lambda」 と検索して開きます。
- 右上の 「関数の作成」 をクリックします。
- 以下のように設定します。
- 関数名:
NotifyToSlackPolePole - ランタイム:
Python 3.12（または最新のPython） - アーキテクチャ:
x86_64
- 関数名:
- 右下の 「関数の作成」 をクリックします。
環境変数（Slack URL）の設定
コードの中に直接URLを書くのはセキュリティ上良くないため、環境変数に登録します。
- Lambdaの画面中段にある 「設定」 タブ ＞ 左側の 「環境変数」 ＞ 「編集」 をクリック。
- 「環境変数の追加」 を押し、以下を入力して保存します。
- キー：
SLACK_WEBHOOK_URL - 値：SlackのWebhook URL
- キー：
Pythonコードの貼り付け
「コード」 タブに戻り、元々書かれているコードをすべて消して、以下のコードを貼り付けます。
import json
import urllib.request
import os
import logging
logger = logging.getLogger()
logger.setLevel(logging.INFO)
SLACK_WEBHOOK_URL = os.environ['SLACK_WEBHOOK_URL']
def lambda_handler(event, context):
try:
# SQSから送られてきたメッセージを処理
for record in event['Records']:
# SNSからSQSに渡された中身（body）を取り出す
sns_message = json.loads(record['body'])
alarm_info = json.loads(sns_message['Message'])
# アラームの名前や理由を取得
alarm_name = alarm_info.get('AlarmName', '不明なアラーム')
alarm_desc = alarm_info.get('NewStateReason', '詳細なし')
# Slackに送るメッセージのデザイン
slack_payload = {
"text": f"🚨 *WordPress セキュリティ警告*\n*アラーム名:* {alarm_name}\n*詳細:* {alarm_desc}\n至急、AWS CloudWatchを確認してください！",
"icon_emoji": ":rotating_light:"
}
# Slackへ送信
req = urllib.request.Request(
SLACK_WEBHOOK_URL,
data=json.dumps(slack_payload).encode('utf-8'),
headers={'Content-Type': 'application/json'}
)
urllib.request.urlopen(req)
logger.info("Slack通知に成功しました。")
return {'statusCode': 200, 'body': 'Success'}
except Exception as e:
logger.error(f"エラーが発生しました: {e}")
return {'statusCode': 500, 'body': 'Error'}
貼り付けたら、ファイル上部の 「Deploy」 ボタンを押してコードを保存します。
Lambdaに「SQSを読み取る権限」を渡す
LambdaがSQSの中身を見るためには「許可証（IAMロール）」が必要です。
- Lambdaの 「設定」 タブ ＞ 左側の 「アクセス権限」 を開きます。
- 実行ロールの青いリンク（
NotifyToSlack-role-xxxxのような名前）をクリックすると、IAMの画面が開きます。
- 「許可を追加」 ＞
「ポリシーをアタッチ」 をクリックします。
- 検索窓で
AWSLambdaSQSQueueExecutionRoleと検索し、チェックを入れて 「許可を追加」 を押します。 (※これでLambdaがSQSを読み取れるようになりました！)
SQSとLambdaを合体させる！
最後の仕上げです。「SQSにデータが入ったらLambdaを動かす」というトリガーを設定します。
- Lambdaの画面に戻り、上部の 「＋ トリガーを追加」 をクリックします。
SQSキューの入力欄をクリックし、ステップ2で作った wp-alert-queue を選びます。
右下の 「追加」 をクリックします
完了！実際に攻撃テストをしてみよう
お疲れ様でした！これでAWS内の美しい連携システムが完成しました。 実際に動くかテストしてみましょう。
【テストの手順】
- 自分のブログのログイン画面（
https://あなたのドメイン/wp-login.php）をブラウザで開きます。 - F5キー（更新）を5〜6回ほど、連続して連打します。
- そのまま5分ほど待ちます。
設定したSlackのチャンネルに、「🚨 WordPress セキュリティ警告」 というメッセージがピコン！と通知されれば大成功です！！
まとめ
今回は、直接Lambdaを動かすのではなく、あえて間に「SQS」を挟みました。 これにより、仮に何万回という凄まじい攻撃を受けてアラームが鳴りやまない状態になっても、SQSがバッファ（緩衝材）となり、システムを落とすことなく順番に処理してくれます。
これぞクラウドの醍醐味！実質0円でこのエンタープライズ品質のシステムが組めるのは本当に楽しいですね。 次回はいよいよ、この通知システムをさらに進化させ、**「攻撃者のIPアドレスをCloudflareのWAFで全自動ブロックする仕組み」**に挑戦します！
