Lambda & CloudWatchEvents & Slackで 長期的に利用されていないAccessKeyを検知・通知する

概要

「Lambda & CloudWatchEvents & Slackで 長期的にログインしていないIAMユーザを検知・通知する」の第二弾として、AccessKeyの利用状況を調べて、長期的に利用されていないAccessKeyがあったら、Slackに通知できるServerlessソリューションを共有したいと思っています。

そもそも何でこれをやりたいのか

　AWSを利用されている方（特にAWSを社内利用されている方）が分かると思いますが、AccessKeyをやたらに作られて、放置されていることがありますでしょうか？
　自分の場合ですと、商用目的でのAWSアカウントはまだましですが、検証目的でのAWSアカウントには,
たくさんのAccessKeyが作られていて、長期的に利用されていないものもあります。
　AccessKeyを漏洩していなければ、別にいいでしょうという考え方もありますが、乱立しているAccessKeyを見たら、やっぱりどうしても少し気持ちが悪いなと感じました。しかし、一々手動でAccessKeyの利用状況を確認して、整理するのも気が遠くなりますので、長期的に利用されていないAccessKeyを検知・通知するLambdaを作ることに至りました。

使えるまでの流れと各サービスの設定

前回書いたもの
　↓
「Lambda & CloudWatchEvents & Slackで 長期的にログインしていないIAMユーザを検知・通知する」
とほぼほぼ同じですので、違うところ（コード部分）だけを共有したいと思います。

下記のコードでは、60日以上利用されていAccessKeyを検知し、通知する例です。
日数に関しては、変数Daysに設定すればよいです。

lambda_function.py

# -*- coding: utf-8 -*-
from __future__ import print_function
import boto3
import time
import calendar
import slackweb

print('Loading function')

# Webhook
Slack = slackweb.Slack(url="https://hooks.slack.com/services/T0HTZK5S4/B0***************w")

client = boto3.client('iam')

# 日数を指定
Days = 60
Interval = 60 * 60 * 24 * Days

def add_info(Users,Attachments):

    for User in Users:
        Name = User["UserName"]
        AccessKeyId = User["AccessKeyId"]
        LastUsed = str(User["LastUsed"])
        Text = "```AccessKeyId: " + AccessKeyId + "\n" + \
                "UserName: " + Name + "\n" + \
                "LastUsed: " + LastUsed + "```"

        Attachment = {
            "text": Text,
            "color": "danger",
            "mrkdwn_in": ["text"]
        }
        Attachments.append(Attachment)

def lambda_handler(event, context):
    # Slack Message Attachments
    Attachments = []

    # 指定日数以上利用されていないAccessKeyの情報
    AccessKeyInfo = []

    # 作成されてから、利用されていないAccessKeyの情報
    UnusedAccessKeyInfo = []

    # 現在の時刻を取得
    Now = time.time()
    print(Now)

    UsersList = client.list_users()
    for User in UsersList["Users"]:
        AccessKeysList = client.list_access_keys(
            UserName = User['UserName']
        )

        for AccessKeyMetadata in AccessKeysList["AccessKeyMetadata"]:
            print(AccessKeyMetadata["AccessKeyId"])
            Key = client.get_access_key_last_used(
                AccessKeyId = AccessKeyMetadata["AccessKeyId"]
            )

            # Keyが利用された場合
            if str(Key["AccessKeyLastUsed"].get("LastUsedDate")) != 'None':

                # Keyが最後に利用された時間(Unix Time)
                AccessKeyLastUsedUnixTime = calendar.timegm(Key['AccessKeyLastUsed']['LastUsedDate'].utctimetuple())
                print(AccessKeyLastUsedUnixTime)

                if Now - AccessKeyLastUsedUnixTime > Interval:
                    Info = {
                        "UserName" : User['UserName'],
                        "AccessKeyId" : AccessKeyMetadata["AccessKeyId"],
                        "LastUsed" : Key['AccessKeyLastUsed']['LastUsedDate'],
                    }
                    AccessKeyInfo.append(Info)

            # Keyが未使用の場合
            else:
                Info = {
                    "UserName" : User['UserName'],
                    "AccessKeyId" : AccessKeyMetadata["AccessKeyId"],
                    "LastUsed" : "Null"
                }
                UnusedAccessKeyInfo.append(Info)
                print(str(Key["AccessKeyLastUsed"].get("LastUsedDate")))

    # 指定日数以上利用されていないAccessKeyがない場合
    if len(AccessKeyInfo) == 0 and len(UnusedAccessKeyInfo) == 0 :
        Attachments = [
            {
                "pretext": "AccessKeyの棚卸しを行いました、問題がありません.",
            }
        ]

    # 指定日数以上利用されていないAccessKeyがある場合
    else :
        Attachments = [
            {
                "pretext": "AccessKeyの棚卸しを行いました.\n" + \
                            str(Days) + "日以上利用されていないAccessKeyがあります!",
            }
        ]

        print(str(Days) + "日利用されていないAccessKeyは下記の通り")
        print(len(AccessKeyInfo))
        add_info(AccessKeyInfo,Attachments)

        print("作成されてから、利用されていないAccessKeyは下記の通り")
        print(len(UnusedAccessKeyInfo))
        add_info(UnusedAccessKeyInfo,Attachments)

    # Slackに送信
    Slack.notify(attachments = Attachments)

　

実行結果

　上記のコードをLambdaで実行すると、Slackには下記のようなメッセージが指定したSlackのチャンネルに送信されます。

最後

　pythonの学習も兼ねて、作ってみたものですので、コード自体が綺麗ではありませんが、すこし皆さんに共有できたらいいなと思います。

個人的には、LambdaやAPI Gatewayが出てから、ServerlessアーキテクチャとかChatOpsとかに結構興味がありますので、今後もどんどん共有していきたいです。