Amazon Inspector(+Athena)を使って脆弱性管理

時間やお金を節約しつつ、情報セキュリティを維持（あわよくば向上）したい、という期待に基づく入門編です。システムのセキュリティ対策にはキリがありませんが、定期的に実施することが重要になるため、やはり費用は抑えたいのが基本方針です。

今回は、脆弱性スキャンの自動化サービスであるAmazon Inspectorを使って、CVEに基づく評価結果を収集してみます。

＋CVEとは
＋Amazon Inspector仕様
＋今回の実装内容
＋Inspectorの評価結果をAthenaで管理
＋最後に

CVEとは

Inspectorのまえに、CVEについておさらい。 IPAのWebサイトには「共通脆弱性識別子CVE概説」があります。

直接的にはMITRE社が提供しているCVEですが、脆弱性データベースであるNVDはアメリカ合衆国の機関であるNISTが管理しています。 NISTでは脆弱性タイプを仕様として標準化しており、それをCVEに統合しているようです。

ITエンジニアが実際によく使うのは、CVSSというCVEの評価スコアです。 IPAには、概要なのか疑わしいほど詳しい「共通脆弱性評価システムCVSS v3概説」もあります。「時間の経過や利用環境(システム)によって左右されない」という基準をベースとして、多面的に評価されたうえ、脆弱性の深刻度が0(低)～10.0(高)の数値で表されます。

AWSのFAQにも以下のような記載があります。

一般的な脆弱性や暴露 (CVE) ルールは、公に知られている情報セキュリティの脆弱性およびリスクにさらされていないかどうかをチェックします。 CVE ルールについては、National Vulnerability Database (NVD) から一般向け情報が入手できます。NVD の Common Vulnerability Scoring System (CVSS) が、緊急度情報の優先ソースとして使用されます。ある CVE のスコアが NVD によって設定されていないが、Amazon Linux AMI Security Advisory (ALAS) 内に CVE が存在する場合、Amazon Linux アドバイザリからの緊急度が使用されます。ある CVE について両方からスコアを入手できない場合、その CVE は所見として報告されません。NVD および ALAS からの最新情報が日次でチェックされ、情報に応じてルールパッケージが更新されます。

Inspector緊急度	CVSSベーススコア	Amazon Linux AMI Security Advisory緊急度
高 = 9.0	>=5	クリティカルまたは重要
中 = 6.0	< 5 and >= 2.1	中
低 = 3.0	< 2.1 and >= 0.8	低
情報 = 0.0	< 0.8	該当なし

Amazon Inspector仕様

AWS提供のエージェントソフトを導入する必要がある
エージェントからはインターネット通信が可能なこと　※テレメトリと呼ばれるスキャン結果をInspectorに送信するため
エージェント形式(push型)ソフトのため、対象サーバが停止していると脆弱性スキャンの対象外となる。
スキャン状態(start/processing/complete等)の通知はできるが、スキャンの評価結果を通知する機能はない。
スケジュール実行できる。

Inspectorのスキャンで使用できる評価内容は、下記ページをご参照下さい。

Amazon Inspector のルールパッケージとルール

今回の実装内容

スクリプト以外で事前にやっておくことは、次の点です。

評価ターゲットを決める：　今回はEC2タグを条件としました。
評価テンプレートを作成する：　AWSが提供するルールパッケージから選択します(今回はCVE)。
評価テンプレートのスケジュールを決める：　AWSコンソール「Inspector評価テンプレート」画面から設定できます(中身はCloudWatchイベント)。
評価テンプレートの通知用SNSを作成する：　「Amazon Inspector 通知用の SNS トピックを設定するには」の通り、設定します。

設定済みの評価テンプレートは、このようになります。

Amamzon Inspector 評価テンプレート

Inspectorによるスキャン結果を通知する際は、Inspectorの通知用SNSをトリガーにしたLambda関数に任せます。スキャンから通知までの流れは次の通りです。

Inspector評価テンプレート実行
「実行完了」ステータスでInspectorからSNS通知
SNS通知からLambda関数を起動
Lambda関数でInspector評価結果をパース →取得データはcsvファイルとしてS3に保存し、SNSでメール通知

import boto3
import csv
import json
from datetime import datetime as d
 
inspec = boto3.client('inspector')
s3 = boto3.client('s3')
s3r = boto3.resource('s3')
sns = boto3.client('sns')
 
topic = 'arn:aws:sns:ap-northeast-1:~'
subject = '[Amazon Inspector] Monthly report'
 
BUCKET = '~'
 
base_line = 7
 
def mailing(f_name):
    presigne = s3.generate_presigned_url(
        ClientMethod = 'get_object',
        Params = {'Bucket' : BUCKET, 'Key' : f_name},
        ExpiresIn = 604800, # unit: second
        HttpMethod = 'GET')
 
    s3r.Object(BUCKET,f_name).download_file('/tmp/tmp3.csv')
 
    with open('/tmp/tmp3.csv') as f:
        reader = csv.reader(f)
        next(reader)
 
        host = ''
        score = 0
        desc = ''
        msg = ''
        message = ''
        count = 0
        no_count = 0
 
        for row in reader:
            host = row[0]
            score = float(row[1])
            desc = row[2]
 
            if score >= base_line:
                count += 1
            elif score < base_line:
                no_count += 1
 
        message = '''
Severity %i 以上は%iあります.
Severity %i 未満は%iあります.
下記リンクから詳細情報を確認してください.
''' % (base_line,count,base_line,no_count)
 
    mail = {'TopicArn': topic ,'Message':  message + presigne,'Subject': subject}
    sns.publish(**mail)
 
 
def lambda_handler(event, context):
 
    print(event)
    event_m = json.loads(event['Records'][0]['Sns']['Message'])
    print(event_m['run'])
    run_ARN = event_m['run']
 
    next_t = {}
    while True:
        res = inspec.list_findings(
            assessmentRunArns=[
                run_ARN
            ],
            maxResults=500,
            **next_t
        )
 
        finds = res['findingArns']
 
        rows = []
        for i in finds:
            find = {}
            find = inspec.describe_findings(
                findingArns=[
                    i,
                ],
                locale='EN_US'
            )['findings']
 
            for f in find:
                host = f['assetAttributes']['hostname']
#                score = f['attributes'][0].get('value')
                score = f['numericSeverity']
                title = f['title']
                desc = f['description']
                comment = f['recommendation']
 
                row = [host,score,title,desc,comment]
                rows.append(row)
 
        with open('/tmp/tmp.csv','a',newline='') as f:
            cw = csv.writer(f, delimiter=',')
            cw.writerows(rows)
 
        if 'nextToken' in res:
            next_t['nextToken'] = res['nextToken']
        else:
            break
 
 
    head = ['hostname','value','title','description','recommendation']
    with open("/tmp/tmp.csv") as rf, open("/tmp/tmptmp.csv","w",newline='') as wf:
        cr = csv.reader(rf)
        cw = csv.writer(wf)
        cw.writerow(head)
        cw.writerows(cr)
 
    now = d.now()
    now_str = now.strftime('%Y-%m-%d-%H%M%S')
    f_name = 'inspector' + '/' + now_str + '.csv'
 
    s3.upload_file('/tmp/tmptmp.csv', BUCKET, f_name)
 
    mailing(f_name)

上記の通り、Lambda関数ではlist_findings -> describe_findingsのうち、以下の値だけ取得しています。

  host = f['assetAttributes']['hostname']
  score = f['numericSeverity']
  title = f['title']
  desc = f['description']
  comment = f['recommendation']

上記の項目以外にも、Inspectorの評価結果は色々と情報を持っています。気になる方はAWSコンソール「評価の実行」画面から、元データのjsonをご覧ください。

今回のメール通知内容は以下の通りになりました。

Inspectorの評価結果をAthenaで管理

Lambda関数でわざわざcsv出力したのは、Inspectorによる評価結果をAthenaから見たかったからです。 Athenaから脆弱性情報できれば、時系列やホスト毎の集計も簡単にできます。維持費はS3の保存データだけ、というのも魅力です。

試しにAWS GlueのクローラでLambda関数がcsvファイルをアップロードしたS3パスを指定すると、Athenaテーブルを自動生成してくれました。今回、Athenaからの脆弱性情報はこのようなかたちで参照できます。

Athenaからの脆弱性情報

最後に

今回は脆弱性診断のスキャン結果をメール通知することが目的の1つだったため、S3上のcsvファイルに対する署名付きURLをメール送付しています。脆弱性診断の評価結果を自動通知する仕組みは便利ですが、「システムの脆弱性情報」なので機密情報に相当する程度には扱いに配慮が必要です。そのため、実運用では元データに対するアクセス制御等の保護が必須です。

セキュリティ対策は実施しても、公にせず、むしろ情報は内部にとどめておくことも多々あります。快適なadmin生活の助けになるツールを今後もつくっていきたいところです。

このブログの著者