Amazon Inspectorの診断結果をメール通知する

この記事は公開されてから半年以上経過しています。情報が古い可能性がありますので、ご注意ください。

はじめに

Inspectorの診断結果をメール通知する方法が無いかな、と思ったら
以下のような記事が2016年に投稿されていました。全然知らなかった…。

Amazon Inspector でセキュリティ脆弱性テストを拡大

Inspectorの診断結果をLambdaに渡して、見やすい形でメール通知させる、というものです。

本記事では、ここで紹介されているLambdaのコード(Python)を少しだけ変えて紹介したいと思います。
Inspectorの利用方法までは細かく解説していません。

設定方法

Inspectorが診断結果をLambdaに渡す用のSNSトピック作成

Inspectorが診断結果をLambdaに渡す用のSNSトピックを作成します。
診断結果を通知したいメールアドレスを登録するわけではないので注意ください。

トピック名はinspector-lambdaとしました。

アクセスポリシーはJSONを貼り付けて設定します。

# 貼り付けるJSONの内容
{
  "Version": "2008-10-17",
  "Id": "inspector-sns-publish-policy",
  "Statement": [
    {
      "Sid": "inspector-sns-publish-statement",
      "Effect": "Allow",
      "Principal": {
        "Service": "inspector.amazonaws.com"
      },
      "Action": "SNS:Publish",
      "Resource": "arn:aws:sns:*"
    }
  ]
}

# 貼り付けるJSONの内容

{

"Version": "2008-10-17",

"Id": "inspector-sns-publish-policy",

"Statement": [

{

"Sid": "inspector-sns-publish-statement",

"Effect": "Allow",

"Principal": {

"Service": "inspector.amazonaws.com"

"Action": "SNS:Publish",

"Resource": "arn:aws:sns:*"

}

]

}

それ以外は特に設定せずトピックを作成します。

Lambda用のIAMロール作成

以下内容のIAMポリシーを作成し、

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "logs:CreateLogGroup",
                "logs:CreateLogStream",
                "logs:PutLogEvents"
            ],
            "Resource": "arn:aws:logs:*:*:*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "inspector:DescribeFindings",
                "SNS:CreateTopic",
                "SNS:Subscribe",
                "SNS:ListSubscriptionsByTopic",
                "SNS:Publish"
            ],
            "Resource": "*"
        }
    ]
}

{

"Version": "2012-10-17",

"Statement": [

{

"Effect": "Allow",

"Action": [

"logs:CreateLogGroup",

"logs:CreateLogStream",

"logs:PutLogEvents"

"Resource": "arn:aws:logs:*:*:*"

{

"Effect": "Allow",

"Action": [

"inspector:DescribeFindings",

"SNS:CreateTopic",

"SNS:Subscribe",

"SNS:ListSubscriptionsByTopic",

"SNS:Publish"

"Resource": "*"

}

]

}

このポリシーをアタッチしたロールを作成します。

Lambda関数の作成

sns-message-pythonという設計図を利用します。

IAMロールや、関数実行のトリガーとなるSNSトピックを設定し、関数を作成します。

関数が作成されたら、以下のコードを適用します。
黄色部分は適宜変更ください。

from __future__ import print_function
import boto3
import json
import datetime
 
sns = boto3.client('sns')
inspector = boto3.client('inspector')
 
# SNS topic - will be created if it does not already exist
SNS_TOPIC = "Inspector-Finding-Delivery" <span style="color: #ffff00;"># メール通知用のSNSトピック名</span>
 
# Destination email - will be subscribed to the SNS topic if not already
DEST_EMAIL_ADDR = "xxxx@skyarch.net" <span style="color: #ffff00;"># メール通知の宛先</span>

# inspector finding titles excluded from mail notification
exception = [
        "Unsupported Operating System or Version",
        "No potential security issues found" <span style="color: #ffff00;"># 通知不要なタイトルをリストで記述</span>
        ]

# quick function to handle datetime serialization problems
enco = lambda obj: (
    obj.isoformat()
    if isinstance(obj, datetime.datetime)
    or isinstance(obj, datetime.date)
    else None
)
 
def lambda_handler(event, context):
 
    # extract the message that Inspector sent via SNS
    message = event['Records'][0]['Sns']['Message']
 
    # get inspector notification type
    notificationType = json.loads(message)['event']
 
    # skip everything except report_finding notifications
    if notificationType != "FINDING_REPORTED":
        print('Skipping notification that is not a new finding: ' + notificationType)
        return 1
   
    # extract finding ARN
    findingArn = json.loads(message)['finding']
 
    # get finding and extract detail
    response = inspector.describe_findings(findingArns = [ findingArn ], locale='EN_US')
    print(response)
    try:
        finding = response['findings'][0]
    except OSError as err:
        print("OS error: {0}".format(err))
    except:
        print("Unexpected error:", sys.exc_info()[0])
        raise
       
    # skip uninteresting findings
    title = finding['title']
    if title in exception:
        print('Skipping finding: ', title)
        return 1

    # get the information to send via email
    subject = title[:100] # truncate @ 100 chars, SNS subject limit
    messageBody = "Title:\n" + title + "\n\nDescription:\n" + finding['description'] + "\n\nRecommendation:\n" + finding['recommendation']
   
    # un-comment the following line to dump the entire finding as raw json
    # messageBody = json.dumps(finding, default=enco, indent=2)
 
    # create SNS topic if necessary
    response = sns.create_topic(Name = SNS_TOPIC)
    snsTopicArn = response['TopicArn']
 
    # check to see if the subscription already exists
    subscribed = False
    response = sns.list_subscriptions_by_topic( TopicArn = snsTopicArn )
 
    # iterate through subscriptions array in paginated list API call
    while True:
        for subscription in response['Subscriptions']:
            if ( subscription['Endpoint'] == DEST_EMAIL_ADDR ):
                subscribed = True
                break
       
        if 'NextToken' not in response:
            break
       
        response = sns.list_subscriptions_by_topic(
            TopicArn = snsTopicArn,
            NextToken = response['NextToken']
            )
       
    # create subscription if necessary
    if ( subscribed == False ):
        response = sns.subscribe(
            TopicArn = snsTopicArn,
            Protocol = 'email',
            Endpoint = DEST_EMAIL_ADDR
            )
 
    # publish notification to topic
    response = sns.publish(
        TopicArn = snsTopicArn,
        Message = messageBody,
        Subject = subject
        )
 
    return 0

100

101

102

103

104

105

106

107

from __future__ import print_function

import boto3

import json

import datetime

sns = boto3.client('sns')

inspector = boto3.client('inspector')

# SNS topic - will be created if it does not already exist

SNS_TOPIC = "Inspector-Finding-Delivery" # メール通知用のSNSトピック名

# Destination email - will be subscribed to the SNS topic if not already

DEST_EMAIL_ADDR = "xxxx@skyarch.net" # メール通知の宛先

# inspector finding titles excluded from mail notification

exception = [

"Unsupported Operating System or Version",

"No potential security issues found" # 通知不要なタイトルをリストで記述

]

# quick function to handle datetime serialization problems

enco = lambda obj: (

obj.isoformat()

if isinstance(obj, datetime.datetime)

or isinstance(obj, datetime.date)

else None

)

def lambda_handler(event, context):

# extract the message that Inspector sent via SNS

message = event['Records'][0]['Sns']['Message']

# get inspector notification type

notificationType = json.loads(message)['event']

# skip everything except report_finding notifications

if notificationType != "FINDING_REPORTED":

print('Skipping notification that is not a new finding: ' + notificationType)

return 1

# extract finding ARN

findingArn = json.loads(message)['finding']

# get finding and extract detail

response = inspector.describe_findings(findingArns = [ findingArn ], locale='EN_US')

print(response)

try:

finding = response['findings'][0]

except OSError as err:

print("OS error: {0}".format(err))

except:

print("Unexpected error:", sys.exc_info()[0])

raise

# skip uninteresting findings

title = finding['title']

if title in exception:

print('Skipping finding: ', title)

return 1

# get the information to send via email

subject = title[:100] # truncate @ 100 chars, SNS subject limit

messageBody = "Title:\n" + title + "\n\nDescription:\n" + finding['description'] + "\n\nRecommendation:\n" + finding['recommendation']

# un-comment the following line to dump the entire finding as raw json

# messageBody = json.dumps(finding, default=enco, indent=2)

# create SNS topic if necessary

response = sns.create_topic(Name = SNS_TOPIC)

snsTopicArn = response['TopicArn']

# check to see if the subscription already exists

subscribed = False

response = sns.list_subscriptions_by_topic( TopicArn = snsTopicArn )

# iterate through subscriptions array in paginated list API call

while True:

for subscription in response['Subscriptions']:

if ( subscription['Endpoint'] == DEST_EMAIL_ADDR ):

subscribed = True

break

if 'NextToken' not in response:

break

response = sns.list_subscriptions_by_topic(

TopicArn = snsTopicArn,

NextToken = response['NextToken']

)

# create subscription if necessary

if ( subscribed == False ):

response = sns.subscribe(

TopicArn = snsTopicArn,

Protocol = 'email',

Endpoint = DEST_EMAIL_ADDR

)

# publish notification to topic

response = sns.publish(

TopicArn = snsTopicArn,

Message = messageBody,

Subject = subject

)

return 0

Inspector側の設定

評価テンプレートの設定で、診断結果を今回作成したSNSトピックに通知するように設定します。

診断実施

Inspectorによる診断を実施します。
初回は通知先アドレスにSNSのサブスクリプション確認メールが届くはずなので、承認してください。

2回目では診断の結果が1タイトルごとに1通届きますが、中には通知しなくてもよいタイトルがあると思います。
例えば下記はグローバルからインスタンスにHTTPSでアクセスできるよ、という警告なのですが
LB無しでWEBサービスを提供していれば当たり前のことなので、次回診断以降は通知しないようにしたいところです。

ちなみに、メールの内容をよく見ると「HTTPSをListenしているプロセスが無いのに、
セキュリティグループでHTTPSのアクセスを許可しているよ」と書かれています。
実際はHTTPSをListenしているのですが、診断時間1時間が推奨されるところを
15分に設定して診断しているので、精度が低くなっているのかも知れません。

通知不要なタイトルは、Lambdaのコードの以下の部分に追記してください。
これによって、次回以降はこのタイトルがメールで通知されなくなります。

# inspector finding titles excluded from mail notification
exception = [
        "Unsupported Operating System or Version",
        "No potential security issues found"<span style="color: #ffff00;">,
        "On instance i-xxxxxxxxxxxxxxxxx, TCP port 443 which is associated with 'HTTPS' is reachable from the internet"</span>
        ]

# inspector finding titles excluded from mail notification

exception = [

"Unsupported Operating System or Version",

"No potential security issues found",

"On instance i-xxxxxxxxxxxxxxxxx, TCP port 443 which is associated with 'HTTPS' is reachable from the internet"

]