目的

リソース変更を随時Slackに通知することで、不正な操作や予期しない動きを検知する

 

シーケンス

AWS Config -> Event Bridge Rule -> SNS -> Lambda -> Slack

 

SlackのWeb hook URLを生成

 

「incoming-webhooks」で検索

 

AWS Configを有効にする

有効にする

これでAWSリソースの変更が記録できる

 

SNSの作成

SNSを作成する

 

Event Bridge Ruleの作成

AWS ConfigとSNSを紐づける

 

 

 

Lambda

 

• Lambdaの実行権限はbasicなもので良い
• SLACK_INCOMING_URLにインカミングURL入れてね
• Python3.9

import json
import urllib.request
import logging
from collections import OrderedDict
import pprint

# Slack通知先incoming url
SLACK_INCOMING_URL = "🐱 入力してね"

print('Loading function')


def lambda_handler(event, context):
    #print("Received event: " + json.dumps(event, indent=2))
    message = event['Records'][0]['Sns']['Message']
    print("From SNS: " + message)
    post_slack(message)
    return message


def post_slack(message) -> bool:
    message_array = json.loads(message)
    
    send_data = {
        "attachments": [
            {
                "fallback": "AWS Resource Change Notify",
                "color": "#36a64f",
                "pretext": "AWS Resource Change Notify",
                "title": message_array['detail']['configurationItem']['resourceType'] + " " + message_array['detail']['configurationItemDiff']['changeType'] if 'changeType' in message_array['detail']['configurationItemDiff'] else "",
                "fields":[
                    {
                        "title": "ResourceId",
                        "value": message_array['detail']['configurationItem']['resourceId'],
                        "short": "true"
                    },
                    {
                        "title": "resourceName",
                        "value": message_array['detail']['configurationItem']['resourceName'] if 'resourceName' in message_array['detail']['configurationItem'] else "",
                        "short": "true"
                    },
                    {
                        "title": "Region",
                        "value": message_array['detail']['configurationItem']['awsRegion'],
                        "short": "true"
                    },
                    {
                        "title": "awsAccountId",
                        "value": message_array['detail']['configurationItem']['awsAccountId'],
                        "short": "true"
                    },
                    {
                        "title": "configurationItemStatus",
                        "value": message_array['detail']['configurationItem']['configurationItemStatus'],
                        "short": "true"
                    },
                ]
            },
            {
                "fallback": "raw message",
                "pretext": "raw message",
                "text":pretty_json(message)
            }
        ],
    }
    send_text = json.dumps(send_data)
    request = urllib.request.Request(
        SLACK_INCOMING_URL, 
        data=send_text.encode('utf-8'), 
        method="POST"
    )
    with urllib.request.urlopen(request) as response:
        response_body = response.read().decode('utf-8')
    return True

# JSONを整形
def pretty_json(json_text):
    temp = json.loads(json_text)
    return json.dumps(temp, indent=4, sort_keys=True)

 

リソースを変更して通知テスト

 

できたね🐱