2019.11.30 2022.10.28

Amazon ECRのイメージスキャン結果をSlack通知、Backlogに課題登録

gene

2019年10月に Amazon ECR の脆弱性のスキャン機能が実装された。

Announcing Image Scanning for Amazon ECR

今回は、下記の記事を参考に、

ECR へイメージをプッシュ
ECR のイメージスキャン
スキャン実行をトリガーに Lambda を実行して Slack 通知

という流れでやってみた。

【参考URL】

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

Contents

ECR のスキャン設定
Lambda Layers
IAM role 設定
Lambda 関数
トリガー設定
動作確認
Backlog に課題作成もやってみる
動作確認

ECR のスキャン設定

まずは、 ECR の設定を行う。
ECR のリポジトリ作成時、もしくは既存でリポジトリがある場合は設定を編集して、 Scan on push (プッシュ時にスキャン) を有効にする。
これを設定することで、 ECR にイメージをプッシュするたびにスキャンが実行される。

Lambda Layers

今回は、 Python で Lambda を実行する。
ECR のイメージのスキャン結果を取得するためには、最新の boto3 が必要になるため、 Lambda Layers を設定する。

　【参考URL】

Lambda Layers で最新の AWS SDK を使用する - Qiita

ローカル環境にて、下記のように boto3 を ./python ディレクトリ配下にインストールし、 zip で固める。

$ mkdir python
$ pip install -t ./python boto3
$ zip -r boto3-1.10.28.zip python

AWS Lambda のメニューより、 Layers をクリック。

Create Layer をクリックして、

先ほど作成した zip ファイルをアップロードして、 Layer を作成する。

IAM role 設定

ECR のスキャン結果取得のため、 Lambda を実行するロールに、 ecr:DescribeImages を付与する。

Lambda 関数

Lambda 関数は、参考にさせていただいた記事のものをベースに少し手を加えて、チャンネル名(CHANNEL)、会社名(COMPANY)、SlackのウェブフックURLの(WEBHOOK_URL)を環境変数から取得するようにした。
弊社では、 GitHub や CircleCI などの通知を集約している Slack チャンネルがあり、複数の開発や運用が同時で動いているため、どのプロジェクトのものか判別できるように対応したものである。

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(s_counts):
    """Returns the color setting of severity"""
    if s_counts['CRITICAL'] != 0:
        properties = {'color': 'danger', 'icon': ':red_circle:'}
    elif s_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 = 'ap-northeast-1'
    channel = os.environ['CHANNEL']
    company = os.environ['COMPANY']
    severity_list = ['CRITICAL', 'HIGH', 'MEDIUM', 'LOW', 'INFORMATIONAL', 'UNDEFINED']
    s_counts = scan_result['imageScanFindingsSummary']['findingSeverityCounts']

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

    message = f"*ECRイメージスキャン結果 | {company} | Account:{scan_result['registryId']}*"
    description = scan_result['imageScanStatus']['description']
    text_properties = get_properties(s_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}
スキャン完了日時: {
                    complete_at}
脆弱性情報更新日時: {source_update_at}''',
                'fields': [
                    {'title': 'Critical', 'value': s_counts['CRITICAL'], 'short': True},
                    {'title': 'High', 'value': s_counts['HIGH'], 'short': True},
                    {'title': 'Medium', 'value': s_counts['MEDIUM'], 'short': True},
                    {'title': 'Low', 'value': s_counts['LOW'], 'short': True},
                    {'title': 'Info', 'value': s_counts['INFORMATIONAL'], 'short': True},
                    {'title': 'Undefined', 'value': s_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

Lambda ファンクションに、先ほど作成した Layer を忘れずに設定する。

トリガー設定

Lambda 関数のトリガーとなる、 CloudWatch Events の Rule を作成する。

サービス名に、 Elastic Container Registry(ECR) を選択。

イベントタイプとして、 ECR Image Scan を設定。

ターゲットとして、先ほど作成した Lambda ファンクションをセット。

動作確認

試しに、脆弱性を含んだイメージをプッシュしてみると、

無事通知が行われた。

Backlog に課題作成もやってみる

Slack の通知ではなくて、

検出された脆弱性のレベルに CRITICAL が含まれていた場合は Backlog にて課題登録

ということもやってみる。

【参考URL】

Datadog で障害検知したら Slack と Backlog に通知しよう

Lambda ファンクションは下記のようにした。

from datetime import datetime
from logging import getLogger, INFO
import os
from botocore.exceptions import ClientError
import boto3
import requests

logger = getLogger()
logger.setLevel(INFO)


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 create_backlog_issue(scan_result):
    """Create Backlog Issue"""
    company = os.environ['COMPANY']
    severity = 2
    region = 'ap-northeast-1'
    title = "ECRイメージスキャンにて脆弱性が検出されました"
    url = f"https://{os.environ['BACKLOG_SPACE_ID']}.backlog.com/api/v2/issues"
    repo = f"{scan_result['repositoryName']}:{scan_result['imageTags'][0]}"

    summary = scan_result['imageScanFindingsSummary']['findingSeverityCounts']
    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'
    )
    result_url = f'''https://console.aws.amazon.com/ecr/repositories/{
                    scan_result['repositoryName']}/image/{
                    scan_result['imageDigest']}/scan-results?region={region}'''

    description = """
会社名: %s
イメージ: %s
スキャン完了日時: %s
脆弱性情報更新日時: %s
スキャン結果: %s

URL: %s
    """ % (company, repo, complete_at, source_update_at, summary, result_url)

    payload = {
        'projectId' : os.environ['BACKLOG_PROJECT_ID'],
        'issueTypeId' : os.environ['BACKLOG_ISSUE_TYPE_ID'],
        'priorityId' : severity,
        'summary' : title,
        'description' : description,
        'apiKey' : os.environ['BACKLOG_API_KEY']
    }
    response = requests.post(url, params=payload)
    return response.json()


def lambda_handler(event, context):
    """ECR Image Scan Findings to Backlog"""
    logger.info(event)
    response = 1
    scan_result = get_findings(event['detail'])

    # create backlog issue if 'CRITICAL' found
    counts = scan_result['imageScanFindingsSummary']['findingSeverityCounts']
    logger.info(counts)
    if 'CRITICAL' in counts:
        logger.info("CRITICAL found. Creating Backlog Issue.")
        response = create_backlog_issue(scan_result)

    return response

※ 環境変数に下記の変数設定をする必要があるので注意。

COMPANY
BACKLOG_SPACE_ID
BACKLOG_PROJECT_ID
BACKLOG_ISSUE_TYPE_ID
BACKLOG_API_KEY

※ また、このファンクションでは、 requests を使用しているので、 Lambda Layer に requests を含めてやる必要があるので注意。

$ mkdir python
$ pip install -t ./python boto3 requests
$ zip -r boto3-1.10.28.zip python

動作確認

脆弱性を含んだイメージをプッシュしてみると、無事 Backlog 上に課題が登録された。

上記の処理を組み合わせて、開発時に ECR へプッシュするたびにスキャン実行、スキャン結果を Slack へ通知、もし脆弱性のレベルに CRITICAL が含まれていたら Backlog へ課題登録する、という流れができる。
リンク先の URL でスキャン結果の詳細が確認できるようになっているので、確認および対応をすることができる。