Amazon ECRのイメージスキャン結果をSlackに通知する

2021/12/17 追記

Amazon Inspector による拡張スキャンに対応したバージョンを投稿しましたので、今後は以下の記事を参照していただければ思います。

はじめに

2019/10/28 Amazon ECR でコンテナイメージの脆弱製スキャン機能が利用可能になりました！
しかもスキャン機能自体の利用は無料です。最高。
https://aws.amazon.com/jp/about-aws/whats-new/2019/10/announcing-image-scanning-for-amazon-ecr/

いずれ AWS Chatbot が対応してくれそうですが、
それまでの繋ぎでスキャン結果を Slack に通知する Lambda 関数を作成しました。
Amazon EventBridge（CloudWatch Events）と組み合わせて利用します。

結果イメージ

イメージ名をクリックするとスキャン結果の詳細ページに飛びます。

ざっくり構成

Amazon EventBridge（CloudWatch Events）でイメージスキャン実行を検知し、Lambda関数を起動します。
Lambda関数は DescribeImages API でスキャン結果のサマリーを取得し、整形してSlackに通知します。

Lambda 関数

ランタイムは python 3.7 です。
スキャン結果のサマリーを取得するためには最新のAWS SDK(boto3)が必要です。
関数のデプロイパッケージに含めてもよいですが、Lambda Layersを使う方法 がオススメです。
Lambdaの実行ロールで ecr:DescribeImages を許可してください。
環境変数でSlackのWEBHOOK_URLと通知先チャンネル名の設定が必要です。

lambda_function.py

"""
This is a sample function to send ECR Image ScanFindings to slack.

Environment variables:
    CHANNEL: Slack channel name
    WEBHOOK_URL: Incoming Webhook URL
"""

from datetime import datetime
from logging import getLogger, INFO
from urllib.request import Request, urlopen
from urllib.error import URLError, HTTPError
import json
import os
from botocore.exceptions import ClientError
import boto3

logger = getLogger()
logger.setLevel(INFO)

def get_properties(finding_counts):
    """Returns the color setting of severity"""
    if finding_counts['CRITICAL'] != 0:
        properties = {'color': 'danger', 'icon': ':red_circle:'}
    elif finding_counts['HIGH'] != 0:
        properties = {'color': 'warning', 'icon': ':large_orange_diamond:'}
    else:
        properties = {'color': 'good', 'icon': ':green_heart:'}
    return properties

def get_params(scan_result):
    """Slack message formatting"""
    region = os.environ['AWS_DEFAULT_REGION']
    channel = os.environ['CHANNEL']
    severity_list = ['CRITICAL', 'HIGH', 'MEDIUM', 'LOW', 'INFORMAL', 'UNDEFINED']
    finding_counts = scan_result['imageScanFindingsSummary']['findingSeverityCounts']

    for severity in severity_list:
        finding_counts.setdefault(severity, 0)

    message = f"*ECR Image Scan findings | {region} | Account:{scan_result['registryId']}*"
    description = scan_result['imageScanStatus']['description']
    text_properties = get_properties(finding_counts)

    complete_at = datetime.strftime(
        scan_result['imageScanFindingsSummary']['imageScanCompletedAt'],
        '%Y-%m-%d %H:%M:%S %Z'
    )
    source_update_at = datetime.strftime(
        scan_result['imageScanFindingsSummary']['vulnerabilitySourceUpdatedAt'],
        '%Y-%m-%d %H:%M:%S %Z'
    )

    slack_message = {
        'username': 'Amazon ECR',
        'channels': channel,
        'icon_emoji': ':ecr:',
        'text': message,
        'attachments': [
            {
                'fallback': 'AmazonECR Image Scan Findings Description.',
                'color': text_properties['color'],
                'title': f'''{text_properties['icon']} {
                    scan_result['repositoryName']}:{
                    scan_result['imageTags'][0]}''',
                'title_link': f'''https://console.aws.amazon.com/ecr/repositories/{
                    scan_result['repositoryName']}/image/{
                    scan_result['imageDigest']}/scan-results?region={region}''',
                'text': f'''{description}\nImage Scan Completed at {
                    complete_at}\nVulnerability Source Updated at {source_update_at}''',
                'fields': [
                    {'title': 'Critical', 'value': finding_counts['CRITICAL'], 'short': True},
                    {'title': 'High', 'value': finding_counts['HIGH'], 'short': True},
                    {'title': 'Medium', 'value': finding_counts['MEDIUM'], 'short': True},
                    {'title': 'Low', 'value': finding_counts['LOW'], 'short': True},
                    {'title': 'Informational', 'value': finding_counts['INFORMAL'], 'short': True},
                    {'title': 'Undefined', 'value': finding_counts['UNDEFINED'], 'short': True},
                ]
            }
        ]
    }
    return slack_message

def get_findings(detail):
    """Returns the image scan findings summary"""
    ecr = boto3.client('ecr')
    try:
        response = ecr.describe_images(
            repositoryName=detail['repository-name'],
            imageIds=[
                {'imageDigest': detail['image-digest']}
            ]
        )
    except ClientError as err:
        logger.error("Request failed: %s", err.response['Error']['Message'])
    else:
        return response['imageDetails'][0]

def lambda_handler(event, context):
    """AWS Lambda Function to send ECR Image Scan Findings to Slack"""
    response = 1
    scan_result = get_findings(event['detail'])
    slack_message = get_params(scan_result)
    req = Request(os.environ['WEBHOOK_URL'], json.dumps(slack_message).encode('utf-8'))
    try:
        with urlopen(req) as res:
            res.read()
            logger.info("Message posted.")
    except HTTPError as err:
        logger.error("Request failed: %d %s", err.code, err.reason)
    except URLError as err:
        logger.error("Server connection failed: %s", err.reason)
    else:
        response = 0

    return response

イメージスキャンが完了すると以下のようなイベントが発生します。

{
    "version": "0",
    "id": "85fc3613-e913-7fc4-a80c-a3753e4aa9ae",
    "detail-type": "ECR Image Scan",
    "source": "aws.ecr",
    "account": "123456789012",
    "time": "2019-10-29T02:36:48Z",
    "region": "us-east-1",
    "resources": [
        "arn:aws:ecr:us-east-1:123456789012:repository/my-repo"
    ],
    "detail": {
        "scan-status": "COMPLETE",
        "repository-name": "my-repo",
        "image-digest": "sha256:7f5b2640fe6fb4f46592dfd3410c4a79dac4f89e4782432e0378abcd1234",
        "image-tags": []
    }
}

repository-name と image-digest元に boto3 の describe_images メソッドで
スキャン結果のサマリーを取得しています。

Boto 3 Documentation ECR
https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ecr.html#ECR.Client.describe_images

describe_images のレスポンス例は以下のような感じです。

{
    "imageDetails": [
        {
            "registryId": "123456789012",
            "repositoryName": "amazonlinux",
            "imageDigest": "sha256:7f5b2640fe6fb4f46592dfd3410c4a79dac4f89e4782432e0378abcd1234",
            "imageTags": [
                "2.0.20190115"
            ],
            "imageSizeInBytes": 61283455,
            "imagePushedAt": 1572489492.0,
            "imageScanStatus": {
                "status": "COMPLETE",
                "description": "The scan was completed successfully."
            },
            "imageScanFindingsSummary": {
                "imageScanCompletedAt": 1572489494.0,
                "vulnerabilitySourceUpdatedAt": 1572454026.0,
                "findingSeverityCounts": {
                    "HIGH": 9,
                    "LOW": 5,
                    "MEDIUM": 18
                }
            }
        }
    ]
}

EventBridge の設定

スキャン完了イベントのみを検知するために、
新規のルールの作成では以下のようにカスタムイベントパターンを指定します。

{
  "source": [
    "aws.ecr"
  ],
  "detail-type": [
    "ECR Image Scan"
  ]
}

あとは作成したLambda関数をターゲットに設定するだけでOKです。

参考

Image Scanning
https://docs.aws.amazon.com/AmazonECR/latest/userguide/image-scanning.html
Events and EventBridge
https://docs.aws.amazon.com/AmazonECR/latest/userguide/ecr-eventbridge.html

以上です。
参考になれば幸いです。