OCIのIAMポリシーで陥りがちなミスと解決方法　　　④OCIでアラートメールを送信したい！編

Last updated at 2025-11-25Posted at 2025-11-25

目的

本記事では、「OCI Notifications」というOCI推奨の通知サービスを使って自身が設定したメールアドレス宛に「削除・ポリシー変更・ユーザー操作」などといったイベントをアラートメールとして送信する方法をご紹介いたします。
この方法では SMTP・環境変数などは不要で、OCIのネイティブ機能を使って安全かつ確実にメール通知を行うことが可能です。

前提知識

AWS IAMの基本的な概念を理解している
OCIコンソールにアクセス可能
基本的なOCIサービス（Compute、Networking等）の操作経験
Python（バージョンは2025/10/16時点で3.14.0が最新版）がインストール済みであること
第一弾の記事を読んでいること→https://qiita.com/tkhsyuk/items/92a5ea85a8e71494eaaf
→Python SDKが設定済みであることが前提となっています。

アラートメールを自動送信する手順

①OCIコンソールにログインする。
https://www.oracle.com/jp/cloud/free/

②コンソールにログイン後、左上の赤枠をクリックする。

③タブの中の「Developer Services」をクリックする。

④その中の「Notifications」をクリックする。

⑤Topics画面に遷移するので、その中の「Create topic」をクリックする。

⑥下記の項目にそれぞれ情報を入れて「create」をクリックする。

・Name：test（名前は何でもOK）
・Description：テスト（空白でもOK）

⑦「test」という名前でトピックが１つ追加されていることを確認する。

⑧トピックが追加されていることを確認したら、左タブにある「subscription」をクリックする。

⑨サブスクリプション画面の「Create subscription」をクリックする。

⑩以下の項目に情報を記載して「create」をクリックする。

・subscription topic：test（先ほど作成したトピックを選択）
・protocol：Email（メールアドレス宛に送信したいならこちらを選択）
・Email：（アラートメールを送りたいメールアドレスを記入）

⑪詳細画面に戻ると、サブスクリプションが１つ追加されていることが確認できます。
※この時点ではStateが「Pending(保留中)」になっていますが、この状態が正常です。

⑫　⑩で自身が設定したメールアドレス宛に下記のような招待メールが届いていると思うので、下部の「サブスクリプションの確認」をクリックする。

⑬クリックすると、サブスクリプションの確認完了ページが表示されます。
下記のようなページが表示されたらOCIの「Notifications」サービスが使用可能になり、OCI SDKを使ったメール通知が可能になります。

⑭　⑬でサブスクリプションの確認をすると、⑪で確認したステータスが「Active」に切り替わっていることが確認できます。

⑮最後に、以下のスクリプトを実行すればアラートメールが自身が設定したメールアドレス宛に送信されます。
スクリプト内にある「トピックOCID」は以下の画面でコピー&ペーストが出来ます↓

import oci
from datetime import datetime, timedelta, timezone

class IAMAuditSystem:
    """OCI IAM監視システム（本番用・異常なし通知あり・シンプル版）"""

    def __init__(self, config, topic_id):
        self.config = config
        self.audit_client = oci.audit.AuditClient(config)
        self.notification_client = oci.ons.NotificationDataPlaneClient(config)
        self.topic_id = topic_id

    def check_suspicious_activity(self, hours_back=24):
        """怪しいアクティビティをチェック"""
        end_time = datetime.now(timezone.utc)
        start_time = end_time - timedelta(hours=hours_back)

        try:
            events = self.audit_client.list_events(
                compartment_id=self.config["tenancy"],
                start_time=start_time,
                end_time=end_time
            )
        except Exception as e:
            print(f"イベント取得失敗: {e}")
            return []

        suspicious_events = []
        security_keywords = ['policy', 'group', 'user', 'delete', 'terminate']

        for event in events.data:
            event_name = getattr(event, 'event_name', 'Unknown')
            event_time = getattr(event, 'event_time', None)

            try:
                principal = (
                    event.data.identity.principal_name
                    if event.data and event.data.identity
                    else "Unknown"
                )
            except AttributeError:
                principal = "Unknown"

            if not event_time:
                continue

            # セキュリティ関連の操作判定
            if any(k in event_name.lower() for k in security_keywords):
                reason = ""

                # 業務時間外チェック
                if event_time.weekday() < 5 and not (9 <= event_time.hour < 18):
                    reason = "業務時間外操作"
                elif event_time.weekday() >= 5:
                    reason = "休日操作"

                # 削除/終了操作
                if any(del_kw in event_name.lower() for del_kw in ['delete', 'terminate']):
                    reason = "削除・終了操作"

                if reason:
                    suspicious_events.append({
                        "event_name": event_name,
                        "user": principal,
                        "time": event_time.strftime("%Y-%m-%d %H:%M:%S UTC"),
                        "reason": reason
                    })

        return suspicious_events

    def send_alert(self, suspicious_events):
        """アラートをOCI Notificationsに送信（異常なしも通知）"""
        if not suspicious_events:
            subject = "[OCI Alert] 異常なし（監視正常）"
            body = (
                "監視対象の期間内で怪しいアクティビティは検出されませんでした。\n"
                f"確認時刻: {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S UTC')}\n\n"
                "OCI監視システムは正常に稼働しています。"
            )
        else:
            subject = f"[OCI Alert] 怪しい操作を検出 ({len(suspicious_events)}件)"
            body = "以下のアクティビティが検出されました。\n\n"

            for e in suspicious_events:
                body += f"Event: {e['event_name']}\n"
                body += f"User: {e['user']}\n"
                body += f"Time: {e['time']}\n"
                body += f"Reason: {e['reason']}\n"
                body += "-------------------------\n"

        try:
            message_details = oci.ons.models.MessageDetails(
                body=body,
                title=subject
            )
            self.notification_client.publish_message(
                topic_id=self.topic_id,
                message_details=message_details
            )
            print(f"OCI Notifications経由で通知送信完了: {subject}")

        except Exception as e:
            print(f"通知送信失敗: {e}")


def main():
    try:
        config = oci.config.from_file()

        # Notifications トピックOCIDを指定
        topic_id = "ocid1.onstopic.oc1.ap-tokyo-1.xxxxxxx"

        audit_system = IAMAuditSystem(config, topic_id)

        suspicious_events = audit_system.check_suspicious_activity(24)
        audit_system.send_alert(suspicious_events)

    except Exception as e:
        print(f"エラー: {e}")


if __name__ == "__main__":
    main()

⑯上記のスクリプトを実行すると、下記のような実行結果が表示されます。

【コマンドプロンプト】

【メール】

まとめ

上記のスクリプトを実行することにより、自身が設定したメールアドレス宛にユーザーの不適切なアクティビティをアラートメールとして送信することが可能になります。

なお、上記で紹介したスクリプトは手動で実行した場合の例であり、自動でスクリプトを実行したい場合は、OCIの「Function」サービスを使うのがオススメです。

参考リンク

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up

OCIのIAMポリシーで陥りがちなミスと解決方法 ④OCIでアラートメールを送信したい！編

目的

前提知識

アラートメールを自動送信する手順

まとめ

参考リンク

OCIのIAMポリシーで陥りがちなミスと解決方法　　　④OCIでアラートメールを送信したい！編