Python で緊急対応時の Slack 操作を自動化してみた

このブログはこんな方におすすめ

• 緊急対応の初動を早めたい
• リモートワーク中心になり、緊急対応中の作業分担や進捗確認が難しい
• Google Apps Script で緊急対応時の Slack 操作を自動化してみた の Python 版がほしい

はじめに

クラスメソッドでは Slack API を活用することで、いくつかある緊急対応の初動が約30分早くなりました。 では、どのように活用して初動対応を迅速化したのでしょうか。 本ブログでは、クラスメソッドの緊急対応の一つである、AWS 不正利用対応を例に紹介します。

AWS 不正利用とは、AWS アクセスキーの漏洩などが原因で発生します。 最悪の場合、情報漏洩や経済的損失などの甚大な被害を受ける可能性があります。 クラスメソッドでは、クラスメソッドメンバーズのお客様に対して、ベストエフォート型でこちらの対応支援をしています。

AWS 不正利用の詳細については、下記を参照ください。

• AWS アカウントの不正なアクティビティに気付いた場合、どうすればよいですか?
• [AWS利用者必読] アクセスキー漏洩による不正利用について

初動対応について

AWS からの不正利用に関する通知を受信した場合、主に以下の初動対応を実施しています。

• AWS からの通知をお客様へ転送
• AWS への一次返信
• 専用の Slack チャンネル作成
• チャンネル情報の更新
• 社内関係者をチャンネル招待
• 他チームへの作業停止依頼
• 影響範囲調査
• お客様へ電話連絡

Slack API を活用する前は、冒頭の2つを除いた作業は手動対応していました。 こちらを Slack API を活用して自動化・効率化することで初動を30分短くできました。 下記では、改善内容を2つに分けて説明します。

自動化

緊急対応の中で、人手でやらなくてもよく、地味に時間がかかる作業はありませんか？ そういった作業にこちらの方法は効果があります。 具体的には、Slack API を利用して記載の作業を全て自動化しました。 これによって、AWS 不正利用を検知したらすぐに、記載内容が全て実行されるようになりました。

• 専用のチャンネル作成(conversations.create)
• チャンネル情報の更新(conversations.setTopic)
• 社内関係者をチャンネル招待(conversations.invite)
• 他チームへの作業停止依頼(chat.postMessage)

効率化

緊急対応の中でもどうしても人手が必要な作業ってありませんか？ これらの作業にこちらの方法は効果があります。 具体的には、Slack API の chat.postMessage を利用して作成されたチャンネルに TODO 形式で次の対応を指示するようにしました。 これによって、以下を実現しました。

• 手順書を探す手間と、手順書を往復するロスを削減
• 対応中に各メッセージにリアクションをつけることで、関係者が着手状況をすぐに把握可能
• 各メッセージのスレッドに対応内容を残すことで、関係者がリアルタイムな対応状況をすぐ把握可能
• 手が空いた・途中参加した対応者が、次に何をすれば良いかすぐに判断可能

下記が改善後のアーキテクチャです。

関数

アーキテクチャの中で記載されている各関数を紹介します。

• API の制限については 各 Slack API のドキュメントを参照ください
• 機微な情報は AWS Secrets Manager などに安全に保存して、そこから取得してください
  参考： Serverless Framework でクレデンシャル情報を扱う Lambda ファンクションを作成する
• 利用する関数に応じて、下記のように適切なモジュールをインポートしてください

import os
import requests
import json
import boto3
import re
import logging
import time
from datetime import datetime, timedelta, timezone

logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

チャンネルを作成する

conversations.create method | Slack API

def create_channel(event, context):
    url = "https://slack.com/api/conversations.create"
    payload = {
        "token": SLACK_TOKEN,
        "name": CHANNEL_NAME,
    }
    try:
        res = requests.post(url, params=payload)
        json = res.json()
    except Exception as e:
        logger.exception("create_channel {}".format(e))
        raise
    else:
        return {
            "channel_id": json["channel"]["id"],
        }

チャンネルのトピックを設定する

conversations.setTopic method | Slack

def set_topic(event, context):
    url = "https://slack.com/api/conversations.setTopic"
    payload = {
        "token": SLACK_TOKEN,
        "channel": event["channel_id"],
        "topic": TOPIC,
    }
    try:
        res = requests.post(url, params=payload)
        json = res.json()
    except Exception as e:
        logger.exception("set_topic {}".format(e))
        raise
    else:
        return {
            "channel_id": event["channel_id"],
        }

ユーザグループからメンバーIDを取得する

usergroups.users.list method | Slack

def fetch_userids_from_usergroupid(event, context):
    url = "https://slack.com/api/usergroups.users.list"
    payload = {
        "token": SLACK_TOKEN,
        "usergroup": USER_GROUPID,
    }
    try:
        res = requests.get(url, params=payload)
        json = res.json()
    except Exception as e:
        logger.exception("fetch_memberid_from_usergroupid {}".format(e))
        raise
    else:
        return {
            "channel_id": event["channel_id"],
            "userids": json["users"],
        }

メンバーを招待する

conversations.invite method | Slack

def invite_users(event, context):
    url = "https://slack.com/api/conversations.invite"
    userids = ','.join(event["userids"])
    payload = {
        "token": SLACK_TOKEN,
        "channel": event["channel_id"],
        "users": userids,
    }
    try:
        res = requests.post(url, params=payload)
        json = res.json()
    except Exception as e:
        logger.exception("inviteUsers {}".format(e))
        raise
    else:
        return {
            "channel_id": event["channel_id"],
        }

チャンネルにメッセージを投稿する

chat.postMessage method | Slack

def post_channel(channel_id, message):
    url = "https://slack.com/api/chat.postMessage"
    payload = {
        "token": SLACK_TOKEN,
        "channel": channel_id,
        "text": message,
    }
    try:
        res = requests.post(url, params=payload)
        json = res.json()
    except Exception as e:
        logger.exception("post_slack {}".format(e))
        raise
    else:
        return

# メンションありの場合
def post_mention(event, context):
    post_channel(event["channel_id"], MENTION_TEST)
    return {
        "channel_id": event["channel_id"],
    }
    
    
# メンションなし場合
def post_message(event, context):
    post_channel(event["channel_id"], MSG_TEST)
    return {
        "channel_id": event["channel_id"],
    }

• メンションありの関数(post_mention)は、メンション対象のユーザをチャンネルに招待してから実行してください。 メンバーを招待していない状態で実行すると、対象ユーザにメンション通知がされないように見受けられました。
• 以下はメンションメッセージ例です。 ユーザとユーザグループの場合で指定方法が異なります。 %%%は適切な ユーザID かユーザグループID に置き換えてください。

MENTION_TEST = (
    "-----------------------------------------------------------------------\n"
    ":warning: テストチーム\n"
    "<@%%%>\n" #ユーザを指定する場合
    "<!subteam^%%%>\n" #ユーザグループを指定する場合
    "こちらはテスト投稿です。"
)

おわりに

実運用では、Serveless Framework を使って、アーキテクチャ図の AWS サービスをデプロイしています。 紹介した関数を組み合わせることで、色々な場面で利用できるかと思います。 API 制限などに注意しつつご利用ください。

更新履歴

• HTTP メソッドを見直しました