スラッシュコマンド(SlackAPI) ⇄ API Gateway ⇄ Lambda でSSMセッション情報を取得する

こんにちは。ざわかける！のざわ(@zw_kakeru)です。
Slackでスラッシュコマンドを入力することで、SlackAPI経由でAWS API Gatewayをたたき、AWS Lambdaを実行し、SSM(現AWS Systems Manager Session Manager)セッション情報を取得します。
セッション情報の中でも特に、今現在セッションを繋いでいるユーザー名をSlackに返すようにしていきます。
EC2にセッションを誰が繋いでるのかをSlackから簡単に把握したいというのがモチベーションです。

スラッシュコマンド(SlackAPI) ⇄ API Gateway ⇄ Lambdaの仕組みを作成
IAMロールによる権限の付与
Lambdaスクリプトの作成
結果
おわりに

スラッシュコマンド(SlackAPI) ⇄ API Gateway ⇄ Lambdaの仕組みを作成

SlackからAPI Gateway経由でLambdaにアクセスする仕組みについては別の記事で詳しく述べたので今回は解説しません。
知りたい方は以下のページをご覧ください。

今回の記事はこの記事で解説したシステムの上に乗っけていく想定をしています。
以下に続いていく内容の理解が追いつかない場合は、↑の記事から読むと分かりやすいかもしれません。
なお上の記事ではLambda上のNode環境を用いてJavaScriptで記述していますが、今回はJSではなくPython環境によるPythonでの実装で進めていこうと思います。

IAMロールによる権限の付与

必要となる権限はdescribe-sessionsなので、Lambdaにアタッチされているロールに対してこの権限を付与してあげましょう。
(もしかしたら他にも権限が必要かもしれませんが、後の実装時に足りなければ適宜付与してください。)

Lambdaスクリプトの作成

AWS Lambda上で関数を作成できるようになったら、自由にスクリプトを書いていきましょう。
例えば次のような感じです。

import logging
import boto3
from botocore.exceptions import ClientError

def describe_ssm_sessions():
    client = boto3.client("ssm")
    try:
        response = client.describe_sessions(State="Active")
        user_list = [
            session["Owner"].split("/")[-1] for session in response["Sessions"]
        ]
        text = "Users currently on the server are...\n"
        user = "\n".join(set(user_list)) if user_list else "no one"

        result = {"result": text + user, "data": response}
    except ClientError as err:
        result = {"result": "ClientErrorが発生しました", "data": str(err)}
    return result

def get_command_response(message, result):
    return {
        "response_type": "in_channel",
        "attachments": [
            {
                "color": "#32cd32",
                "color": green,
                "title": "Success",
                "text": message,
            },
            {"title": "Result", "text": result["result"]},
        ],
    }

def handler(event, context):
    event_name = event["text"]
    logger.info(f"Event: {str(event_name)}")

    match event_name:
        case "session":
            result = describe_ssm_sessions()
            logger.info(f"Session result: {str(result)}")
            return get_command_response("Checking ssm sessions...", result)
        case _:
            return get_error_response("Unknown parameters!")

client.describe_sessions(State=”Active”)を使えばサクッとセッション情報が取れます。
上記スクリプトでいうところのresponse[“Sessions”]変数の中身がこれにあたるわけですが、これは実際には次のような感じです。

[{'SessionId': 'hogeuser-XXXXXXXXXXXXXXXXXXXX',
  'Target': 'i-XXXXXXXXXXXXXXXX',
  'Status': 'Connected',
  'StartDate': datetime.datetime(2023, 10, 12, 16, 33, 18, 100000, tzinfo=tzlocal()),
  'DocumentName': 'AWS-StartSSHSession',
  'Owner': 'arn:aws:iam::XXXXXXXXXXXXX:user/hogeuser',
  'Details': '',
  'OutputUrl': {'S3OutputUrl': '', 'CloudWatchOutputUrl': ''}},
 {'SessionId': 'fugauser-XXXXXXXXXXXXXXXXXXXXX',
  'Target': 'i-XXXXXXXXXXXXXXXXXXX',
  'Status': 'Connected',
  'StartDate': datetime.datetime(2023, 10, 12, 16, 19, 27, 373000, tzinfo=tzlocal()),
  'DocumentName': 'AWS-StartSSHSession',
  'Owner': 'arn:aws:iam::XXXXXXXXXXXXXXX:user/fugauser',
  'Details': '',
  'OutputUrl': {'S3OutputUrl': '', 'CloudWatchOutputUrl': ''}},
  ...

ここから必要な情報だけ抜き出して返すだけですね。
しかも今回はユーザー名を取得していくだけなので簡単です。

ユーザーリストを取得している部分は↓です。

        user_list = [
            session["Owner"].split("/")[-1] for session in response["Sessions"]
        ]

Ownerの項目から、必要となるユーザー名の部分だけ抜き出してリストにまとめています。
今回は省略しましたが、複数のEC2インスタンスに対してどのユーザーがセッションを繋いでいるか、なども出し分けることができそうですね。

結果

実行してみた結果は次のとおりです。

セッションを接続しているユーザーの一覧が無事に取得できました。
これでインスタンスを再起動するときなどに、ログイン中のユーザーがいないか事前に確認できますね。

おわりに

かなり説明を端折った気がしますが、ひとまずこんな感じでした。
こういう便利な自動化や仕組みは結構楽しいので、引き続きやっていきたいものですね。
以上です。