Slack 監査ログを収集する

このドキュメントでは、Amazon S3 を使用して Slack 監査ログを Google Security Operations に取り込む方法について説明します。パーサーは、まずブール値を正規化し、事前定義されたフィールドをクリアします。次に、「message」フィールドを JSON として解析し、JSON 以外のメッセージは破棄して処理します。特定のフィールド（date_create と user_id）の有無に応じて、パーサーは異なるロジックを適用して、メタデータ、プリンシパル、ネットワーク、ターゲット、情報などの未加工ログフィールドを UDM にマッピングし、セキュリティ結果を構築します。

始める前に

次の前提条件を満たしていることを確認してください。

• Google SecOps インスタンス
• Slack Enterprise Grid テナントと管理コンソールへの特権アクセス
• AWS（S3、IAM、Lambda、EventBridge）への特権アクセス

Slack の前提条件（アプリ ID、OAuth トークン、組織 ID）を収集する

1. Slack 管理コンソールにログインします。
2. https://api.slack.com/apps にアクセスし、[Create New App > From scratch] をクリックします。
3. 一意のアプリ名を入力し、Slack ワークスペースを選択します。
4. [アプリを作成] をクリックします。
5. 左側のサイドバーで [OAuth & Permissions] に移動します。
6. [Scopes] セクションに移動し、次のユーザー トークン スコープ（auditlogs:read）を追加します。
7. [Install to Workspace] > [Allow] をクリックします。
8. インストールが完了したら、[組織レベルのアプリ] に移動します。
9. [組織にインストール] をクリックします。
10. 組織のオーナー/管理者アカウントでアプリを承認します。
11. xoxp- で始まる [User OAuth Token] をコピーして安全に保存します（これが SLACK_AUDIT_TOKEN です）。
12. Slack 管理コンソールの [Settings & Permissions] > [Organization settings] で確認できる組織 ID をメモします。

Google SecOps 用に AWS S3 バケットと IAM を構成する

1. バケットの作成のユーザーガイドに沿って、Amazon S3 バケットを作成します。
2. 後で参照できるように、バケットの名前とリージョンを保存します（例: slack-audit-logs）。
3. IAM ユーザーの作成のユーザーガイドに沿って、ユーザーを作成します。
4. 作成したユーザーを選択します。
5. [セキュリティ認証情報] タブを選択します。
6. [アクセスキー] セクションで [アクセスキーを作成] をクリックします。
7. [ユースケース] で [サードパーティ サービス] を選択します。
8. [次へ] をクリックします。
9. 省略可: 説明タグを追加します。
10. [アクセスキーを作成] をクリックします。
11. [CSV ファイルをダウンロード] をクリックして、[アクセスキー] と [シークレット アクセスキー] を保存し、後で使用できるようにします。
12. [完了] をクリックします。
13. [権限] タブを選択します。
14. [権限ポリシー] セクションで、[権限を追加] をクリックします。
15. [権限を追加] を選択します。
16. [ポリシーを直接アタッチする] を選択します。
17. AmazonS3FullAccess ポリシーを検索して選択します。
18. [次へ] をクリックします。
19. [権限を追加] をクリックします。

S3 アップロードの IAM ポリシーとロールを構成する

1. AWS コンソールで、[IAM> ポリシー> ポリシーの作成> JSON タブ] に移動します。

2. 次のポリシーを入力します。

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Sid": "AllowPutObjects",
         "Effect": "Allow",
         "Action": "s3:PutObject",
         "Resource": "arn:aws:s3:::slack-audit-logs/*"
       },
       {
         "Sid": "AllowGetStateObject",
         "Effect": "Allow",
         "Action": "s3:GetObject",
         "Resource": "arn:aws:s3:::slack-audit-logs/slack/audit/state.json"
       }
     ]
   }
   

   • 別のバケット名を入力した場合は、slack-audit-logs を置き換えます。

3. [次へ] > [ポリシーを作成] をクリックします。

4. [IAM] > [ロール] > [ロールの作成] > [AWS サービス] > [Lambda] に移動します。

5. 新しく作成したポリシーを関連付けます。

6. ロールに「SlackAuditToS3Role」という名前を付けて、[ロールを作成] をクリックします。

Lambda 関数を作成する

1. AWS コンソールで、[Lambda] > [Functions] > [Create function] に移動します。
2. [Author from scratch] をクリックします。
3. 次の構成情報を提供してください。

1. 関数を作成したら、[コード] タブを開き、スタブを削除して、次のコード（slack_audit_to_s3.py）を入力します。

   #!/usr/bin/env python3
   # Lambda: Pull Slack Audit Logs (Enterprise Grid) to S3 (no transform)
   
   import os, json, time, urllib.parse
   from urllib.request import Request, urlopen
   from urllib.error import HTTPError, URLError
   import boto3
   
   BASE_URL = "https://api.slack.com/audit/v1/logs"
   
   TOKEN        = os.environ["SLACK_AUDIT_TOKEN"]  # org-level user token with auditlogs:read
   BUCKET       = os.environ["S3_BUCKET"]
   PREFIX       = os.environ.get("S3_PREFIX", "slack/audit/")
   STATE_KEY    = os.environ.get("STATE_KEY", "slack/audit/state.json")
   LIMIT        = int(os.environ.get("LIMIT", "200"))             # Slack recommends <= 200
   MAX_PAGES    = int(os.environ.get("MAX_PAGES", "20"))
   LOOKBACK_SEC = int(os.environ.get("LOOKBACK_SECONDS", "3600")) # First-run window
   HTTP_TIMEOUT = int(os.environ.get("HTTP_TIMEOUT", "60"))
   HTTP_RETRIES = int(os.environ.get("HTTP_RETRIES", "3"))
   RETRY_AFTER_DEFAULT = int(os.environ.get("RETRY_AFTER_DEFAULT", "2"))
   # Optional server-side filters (comma-separated "action" values), empty means no filter
   ACTIONS      = os.environ.get("ACTIONS", "").strip()
   
   s3 = boto3.client("s3")
   
   def _get_state() -> dict:
       try:
           obj = s3.get_object(Bucket=BUCKET, Key=STATE_KEY)
           st = json.loads(obj["Body"].read() or b"{}")
           return {"cursor": st.get("cursor")}
       except Exception:
           return {"cursor": None}
   
   def _put_state(state: dict) -> None:
       body = json.dumps(state, separators=(",", ":")).encode("utf-8")
       s3.put_object(Bucket=BUCKET, Key=STATE_KEY, Body=body, ContentType="application/json")
   
   def _http_get(params: dict) -> dict:
       qs  = urllib.parse.urlencode(params, doseq=True)
       url = f"{BASE_URL}?{qs}" if qs else BASE_URL
       req = Request(url, method="GET")
       req.add_header("Authorization", f"Bearer {TOKEN}")
       req.add_header("Accept", "application/json")
   
       attempt = 0
       while True:
           try:
               with urlopen(req, timeout=HTTP_TIMEOUT) as r:
                   return json.loads(r.read().decode("utf-8"))
           except HTTPError as e:
               # Respect Retry-After on 429/5xx
               if e.code in (429, 500, 502, 503, 504) and attempt < HTTP_RETRIES:
                   retry_after = 0
                   try:
                       retry_after = int(e.headers.get("Retry-After", RETRY_AFTER_DEFAULT))
                   except Exception:
                       retry_after = RETRY_AFTER_DEFAULT
                   time.sleep(max(1, retry_after))
                   attempt += 1
                   continue
               # Re-raise other HTTP errors
               raise
           except URLError:
               if attempt < HTTP_RETRIES:
                   time.sleep(RETRY_AFTER_DEFAULT)
                   attempt += 1
                   continue
               raise
   
   def _write_page(payload: dict, page_idx: int) -> str:
       ts  = time.strftime("%Y/%m/%d/%H%M%S", time.gmtime())
       key = f"{PREFIX}/{ts}-slack-audit-p{page_idx:05d}.json"
       body = json.dumps(payload, separators=(",", ":")).encode("utf-8")
       s3.put_object(Bucket=BUCKET, Key=key, Body=body, ContentType="application/json")
       return key
   
   def lambda_handler(event=None, context=None):
       state  = _get_state()
       cursor = state.get("cursor")
   
       params = {"limit": LIMIT}
       if ACTIONS:
           params["action"] = [a.strip() for a in ACTIONS.split(",") if a.strip()]
       if cursor:
           params["cursor"] = cursor
       else:
           # First run (or reset): fetch a recent window by time
           params["oldest"] = int(time.time()) - LOOKBACK_SEC
   
       pages = 0
       total = 0
       last_cursor = None
   
       while pages < MAX_PAGES:
           data = _http_get(params)
           _write_page(data, pages)
   
           entries = data.get("entries") or []
           total += len(entries)
   
           # Cursor for next page
           meta = data.get("response_metadata") or {}
           next_cursor = meta.get("next_cursor") or data.get("next_cursor")
           if next_cursor:
               params = {"limit": LIMIT, "cursor": next_cursor}
               if ACTIONS:
                   params["action"] = [a.strip() for a in ACTIONS.split(",") if a.strip()]
               last_cursor = next_cursor
               pages += 1
               continue
           break
   
       if last_cursor:
           _put_state({"cursor": last_cursor})
   
       return {"ok": True, "pages": pages + (1 if total or last_cursor else 0), "entries": total, "cursor": last_cursor}
   
   if __name__ == "__main__":
       print(lambda_handler())
   

2. [構成> 環境変数 > 編集 > 新しい環境変数を追加] に移動します。

3. 次の環境変数を入力し、実際の値に置き換えます。

   キー	値の例
   S3_BUCKET	slack-audit-logs
   S3_PREFIX	slack/audit/
   STATE_KEY	slack/audit/state.json
   SLACK_AUDIT_TOKEN	xoxp-***（auditlogs:read を含む組織レベルのユーザー トークン）
   LIMIT	200
   MAX_PAGES	20
   LOOKBACK_SECONDS	3600
   HTTP_TIMEOUT	60
   HTTP_RETRIES	3
   RETRY_AFTER_DEFAULT	2
   ACTIONS	（省略可、CSV） user_login,app_installed

4. 関数が作成されたら、そのページにとどまるか、[Lambda] > [関数] > [your-function] を開きます。

5. [CONFIGURATION] タブを選択します。

6. [全般設定] パネルで、[編集] をクリックします。

7. [Timeout] を [5 minutes (300 seconds)] に変更し、[Save] をクリックします。

EventBridge スケジュールを作成する

1. Amazon EventBridge > Scheduler > スケジュールの作成に移動します。
2. 次の構成の詳細を入力します。
   • 定期的なスケジュール: レート（1 hour）。
   • ターゲット: Lambda 関数 slack_audit_to_s3。
   • 名前: slack-audit-1h
3. [スケジュールを作成] をクリックします。

省略可: Google SecOps 用の読み取り専用の IAM ユーザーと鍵を作成する

1. AWS コンソールで、[IAM] > [Users] > [Add users] に移動します。
2. [ユーザーを追加] をクリックします。
3. 次の構成の詳細を入力します。
   • ユーザー: secops-reader。
   • アクセスタイプ: アクセスキー - プログラマティック アクセス。
4. [ユーザーを作成] をクリックします。
5. 最小限の読み取りポリシー（カスタム）を関連付ける: [ユーザー] > [secops-reader] > [権限] > [権限を追加] > [ポリシーを直接関連付ける] > [ポリシーを作成]。

6. JSON エディタで、次のポリシーを入力します。

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Effect": "Allow",
         "Action": ["s3:GetObject"],
         "Resource": "arn:aws:s3:::slack-audit-logs/*"
       },
       {
         "Effect": "Allow",
         "Action": ["s3:ListBucket"],
         "Resource": "arn:aws:s3:::slack-audit-logs"
       }
     ]
   }
   

7. 名前を secops-reader-policy に設定します。

8. [ポリシーの作成> 検索/選択> 次へ> 権限を追加] に移動します。

9. [セキュリティ認証情報] > [アクセスキー] > [アクセスキーを作成] に移動します。

10. CSV をダウンロードします（これらの値はフィードに入力されます）。

Slack Audit Logs を取り込むように Google SecOps でフィードを構成する

1. [SIEM 設定] > [フィード] に移動します。
2. [+ 新しいフィードを追加] をクリックします。
3. [フィード名] フィールドに、フィードの名前を入力します（例: Slack Audit Logs）。
4. [ソースタイプ] として [Amazon S3 V2] を選択します。
5. [ログタイプ] として [Slack Audit] を選択します。
6. [次へ] をクリックします。
7. 次の入力パラメータの値を指定します。
   • S3 URI: s3://slack-audit-logs/slack/audit/
   • Source deletion options: 必要に応じて削除オプションを選択します。
   • ファイルの最大経過日数: 指定した日数以内に変更されたファイルを含めます。デフォルトは 180 日です。
   • アクセスキー ID: S3 バケットにアクセスできるユーザー アクセスキー。
   • シークレット アクセスキー: S3 バケットにアクセスできるユーザーのシークレット キー。
   • アセットの名前空間: アセットの名前空間。
   • Ingestion labels: このフィードのイベントに適用されるラベル。
8. [次へ] をクリックします。
9. [Finalize] 画面で新しいフィードの設定を確認し、[送信] をクリックします。

UDM マッピング テーブル

さらにサポートが必要な場合 コミュニティ メンバーや Google SecOps のプロフェッショナルから回答を得ることができます。