きっかけ

社内のコミュニケーションツールとして、Slackを利用していますが、未使用のAPIがあることに気づきました。それがきっかけで、この度Slackの監査ログAPIを試してみることにしました。

Slack 監査ログAPIとは？

Slack 監査ログAPIは、Slack Enterprise Grid（Slackのエンタープライズ向けプラン）ユーザーが利用することができるAPIで、組織全体のさまざまなイベントやアクティビティのログデータにアクセスできます。

管理者はこのAPIを使って、Slack内のユーザーアクション、ログイン状況、アプリの利用状況などの詳細な監査情報を取得することができます。

使用方法

1. 認証と設定

Slack 監査ログAPIを使用するためには、まず認証情報を取得し、適切な権限を設定する必要があります。

Slack APIの公式サイトにアクセスし、新しいアプリを作成します。
「OAuth & Permissions」ページに移動し、「Bot Token Scopes」にauditlogs:readスコープを追加します。
アプリをインストールし、生成されたBot User OAuth Tokenを保存します。

2. 監査ログAPIの実行

次にAPIを使用して、特定のイベントログを取得します。例えば、最近作られたプライベートチャンネルの情報を得ることができます。

サンプル（curl）:

$  curl --location --request GET 'https://api.slack.com/audit/v1/logs?action=private_channel_created&&limit=1' --header 'Authorization: Bearer "監査ログAPI用のトークン'
{"entries":[{"id":"XXXXXXX","date_create":1724976696,"action":"private_channel_created","actor":{"type":"user","user":{"id":"ユーザID","name":"XXXXX","email":"登録メールアドレス","team":"XXXXX"}},"entity":{"type":"channel","channel":{"id":"XXXXX","privacy":"private","name":"XXXXXXX","is_shared":false,"is_org_shared":false,"is_salesforce_channel":false}},"context":{"location":{"type":"workspace","id":"XXXX","name":"XXXXXXX","domain":"XXXXXXX"},"ua":"Mozilla\/5.0 (Windows NT 10.0.22631; Win64; x64) AppleWebKit\/537.36 (KHTML, like Gecko) Slack\/4.39.95 Chrome\/126.0.6478.127 Electron\/31.2.1 Safari\/537.36 OS_Product\/Workstation Servicing_Channel\/SAC Sonic Slack_SSB\/4.39.95","ip_address":"XX.XX.XX.XX","session_id":XXXXXX}}],"response_metadata":{"next_cursor":""}}

監査ログイベントのSlack通知の実験

さらに、監査ログを定期的にリアルタイムで確認するための試みとして、Pythonスクリプトを用いてSlackチャンネルへの通知を試してみました。

事前準備

Python がインストール済みであること
Slack のメッセージ投稿用のAPIキーを取得済みであること

実施方法

cronを使ってPythonスクリプトを5分ごとに実行します。

crontab設定例

# CrontabによるPythonスクリプトのスケジューリング
0/5 * * * * /usr/bin/python3 /path/to/your_script.py

スクリプトの主な機能

特定のイベント（例：ファイルダウンロード）を監査ログから取得し、指定のSlackチャンネルへ通知します。

サンプルコード（Python）:

import requests
import datetime
import time
import json  # JSONデータの整形に使用

def get_unix_time(minutes=5):
    # UTCを基準にして、指定された分数（デフォルトは5分）前の時間をJSTでUNIXタイムスタンプとして計算する
    past_time = (datetime.datetime.utcnow() + datetime.timedelta(hours=9)) - datetime.timedelta(minutes=minutes)
    past_unix_time = int(time.mktime(past_time.timetuple()))
    print(f"Calculated Past Time (JST): {past_time}")
    print(f"Calculated Past UNIX Time: {past_unix_time}")
    return past_unix_time

def fetch_file_download_events(token, oldest):
    # Slack監査ログAPIからファイルダウンロードイベントを取得する
    headers = {
        'Authorization': f'Bearer {token}'
    }
    params = {
        'action': 'file_downloaded',  # アクションタイプ
        'oldest': oldest,  # この時間からのイベントをフィルタリング
        'limit': 100  # 取得するイベントの最大数
    }
    response = requests.get('https://api.slack.com/audit/v1/logs', headers=headers, params=params)
    return response.json() if response.status_code == 200 else f'Error: {response.status_code} - {response.text}'

def format_date(timestamp):
    # UNIXタイムスタンプを日本時間に変換し、フォーマットして返す
    return (datetime.datetime.utcfromtimestamp(timestamp) + datetime.timedelta(hours=9)).strftime('%Y/%m/%d %H:%M:%S')

def notify_slack(token, channel, message):
    # 特定のSlackチャンネルにメッセージを投稿する
    url = "https://slack.com/api/chat.postMessage"
    headers = {
        "Authorization": "Bearer " + token,
        "Content-Type": "application/json"
    }
    payload = {
        "channel": channel,
        "text": message
    }
    response = requests.post(url, headers=headers, json=payload)
    return response.json() if response.status_code == 200 else f'Error sending message to Slack: {response.text}'

def main():
    # メインの実行関数
    slack_data_token = '監査ログAPIのトークン'
    slack_notify_token = 'Slack投稿用のAPIトークン'
    slack_channel = '投稿先のチャンネル名'

    oldest_time = get_unix_time()  # APIクエリ用の時間を取得
    file_download_events = fetch_file_download_events(slack_data_token, oldest_time)

    entries = file_download_events.get('entries', [])
    
    for entry in entries:
        context = entry.get('context', {})
        ip_address = context.get('ip_address', '')

        # 日時をフォーマットし、メッセージを準備する
        date_created = format_date(entry.get('date_create'))
        details = f"IP Address: {ip_address}\n" \
                  f"Date Created: {date_created}\n" \
                  f"Action: {entry.get('action')}\n" \
                  f"Email: {entry.get('actor', {}).get('user', {}).get('email', '')}\n" \
                  f"File Name: {entry.get('entity', {}).get('file', {}).get('name', '')}\n" \
                  f"File Type: {entry.get('entity', {}).get('file', {}).get('filetype', '')}\n" \
                  f"Location: {context.get('location', {}).get('name', '')}\n" \
                  f"User Agent: {context.get('ua', '')}\n" \
                  f"URL: {entry.get('details', {}).get('url_private', '')}"
        
        # Slackへの通知を実施
        notify_result = notify_slack(slack_notify_token, slack_channel, details)
        print(notify_result)

if __name__ == '__main__':
    main()