개요

"Lambda & CloudWatchEvents & Slack에서 장기적으로 로그인하지 않은 IAM 사용자를 감지 및 통지" 의 제2탄으로서 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등에 상당히 흥미가 있으므로, 앞으로도 점점 공유해 가고 싶습니다.