【F-Secure Radar API + CodePipeline】Webスキャンの実行と結果判定を自動化してみた

import os
import urllib.request
import boto3
import json

code_pipeline = boto3.client('codepipeline')

APIACCESSKEY = os.environ['ApiAccessKey']
APISECRETKEY = os.environ['ApiSecretKey']

WEBSCANID = os.environ['WebScanId']

THRESHOLD_HIGH = os.environ['ThresholdHigh']
THRESHOLD_MEDIUM = os.environ['ThresholdMedium']
THRESHOLD_LOW = os.environ['ThresholdLow']
THRESHOLD_INFORMATION = os.environ['ThresholdInformation']

def call_api(url, method, headers):
    # APIを呼び出す
    req = urllib.request.Request(
        url,
        method=method,
        headers=headers
        )
    with urllib.request.urlopen(req) as res:
        return res.read().decode('utf-8')

def continue_job(jobId, previousReportId):
    # continuationTokenを返却してジョブを継続
    print('Job continue')
    continuationTokenOut = json.dumps(
        {
            'previousJobId' : jobId,
            'previousReportId' : previousReportId
        }
    )
    code_pipeline.put_job_success_result(jobId=jobId, continuationToken=continuationTokenOut)

def judge_result(jobId, result):
    # スキャン結果を判定
    print('Judge Result')
    print(str(result))
    for item in result:
        print(str(item))
        if item['VulnRiskLevel'] == "High" and item['TotalCount'] > int(THRESHOLD_HIGH):
            code_pipeline.put_job_failure_result(jobId=jobId, failureDetails={'message': 'Failed due to number of vulnerabilities.(Type: High)', 'type': 'JobFailed'})
            return
        elif item['VulnRiskLevel'] == "Medium" and item['TotalCount'] > int(THRESHOLD_MEDIUM):
            code_pipeline.put_job_failure_result(jobId=jobId, failureDetails={'message': 'Failed due to number of vulnerabilities.(Type: Medium)', 'type': 'JobFailed'})
            return
        elif item['VulnRiskLevel'] == "Low" and item['TotalCount'] > int(THRESHOLD_LOW):
            code_pipeline.put_job_failure_result(jobId=jobId, failureDetails={'message': 'Failed due to number of vulnerabilities.(Type: Low)', 'type': 'JobFailed'})
            return
        elif item['VulnRiskLevel'] == "Information" and item['TotalCount'] > int(THRESHOLD_INFORMATION):
            code_pipeline.put_job_failure_result(jobId=jobId, failureDetails={'message': 'Failed due to number of vulnerabilities.(Type: Information)', 'type': 'JobFailed'})
            return

    # 各レベルに設定したしきい値を超えなければOK
    code_pipeline.put_job_success_result(jobId=jobId)

def lambda_handler(event, context):
    print('----------')
    print('Start Function')
    try:
        jobId = event['CodePipeline.job']['id']
        jobData = event['CodePipeline.job']['data']

        # スキャンID取得
        scanId = WEBSCANID

        # continuationTokenの有無をチェック
        if 'continuationToken' in jobData:
            # スキャン情報を取得
            url = "https://api.radar.f-secure.com/api/latest/vulnerabilityscans/" + scanId + "/info"
            headers = {
                "Content-Type": "application/json",
                "ApiAccessKey": APIACCESSKEY,
                "ApiSecretKey": APISECRETKEY
            }
            res = call_api(url, "GET", headers)

            # レスポンスから'LastReportId'を取り出す
            data = json.loads(res)
            reportId = data['LastReportId']

            # continuationTokenから'previousReportId'を取り出す
            continuationTokenIn = json.loads(jobData['continuationToken'])
            previousReportId = continuationTokenIn['previousReportId']

            print("reportId:" + str(reportId) + ", previousReportId:" + str(previousReportId))

            """LastReportIdとpreviousReportIdを比較
            スキャンが完了するとLastReportIdの値が更新されるため、
            LastReportIdとpreviousReportIdを比較して異なる場合はスキャンが完了したものと判定する
            """
            if previousReportId == reportId:
                # 値が同じ場合はジョブ継続
                continue_job(jobId, previousReportId)
            else:
                # スキャン結果を取得
                print('Get ScanResult')
                url = "https://api.radar.f-secure.com/api/latest/webscans/" + scanId + "/reports/" + reportId + "/statistics/vulnerabilitiesandfindings"
                headers = {
                    "Content-Type": "application/json",
                    "ApiAccessKey": APIACCESSKEY,
                    "ApiSecretKey": APISECRETKEY
                }
                res = call_api(url, "GET", headers)

                # 結果判定
                judge_result(jobId, json.loads(res))

                print('Finish Scan')
        else:
            print('Start Scan')
            # スキャン開始
            url = "https://api.radar.f-secure.com/api/latest/vulnerabilityscans/" + scanId + "/start"
            headers = {
                "Content-Type": "application/json",
                "ApiAccessKey": APIACCESSKEY,
                "ApiSecretKey": APISECRETKEY
            }

            res = call_api(url, "PUT", headers)

            data = json.loads(res)
            lastReportId = data['LastReportId']

            continue_job(jobId, lastReportId)

    except Exception as e:
        # 予期せぬエラー
        print(e)
        code_pipeline.put_job_failure_result(jobId=jobId, failureDetails={'message': 'Function failed due to exception.', 'type': 'JobFailed'})

    print('Finish Function')
    print('----------')
    return

システムスキャンの時とほぼ同じコードの為説明は割愛します。

Lambda を実行するロールに CodePipeline へのアクセスを許可するポリシーをつけるのをお忘れなく。

API キーや、結果判定時に使うための値は環境変数にセットしています。

パイプラインの編集

CodePipeline のコンソール画面より、[編集する]→[ステージを追加する]にて以下アクションを追加します。

以下の様にデプロイ後にスキャンが実行されるパイプラインになります。

動かしてみる

では、パイプラインを動かしていきます。

ソースを更新してプッシュすると、パイプラインが動きます。

少し待つと InvokeWebScan が進行中になります。スキャンには時間がかかるのでしばらく待ちましょう。

しばらくすると、InvokeWebScan が完了しました。

Lambda の環境変数に設定しているしきい値を超える脆弱性は検出されなかった様です。

メール通知の設定もしていたので、結果がメールで届いていました。Low が一個だけ検出された様です。

本筋とは関係ありませんが、設定するだけでメールでお手軽に結果が確認できるのは嬉しいポイントです。

上記コードで結果取得 API で取得した値をログに吐き出しているため、CloudWatch でも同様の内容が確認できました。

同じく Low が1つですね。

まとめ

今回はF-secure Radar の API を使った Web スキャンを試してみました。

アプリケーションをデプロイする度に脆弱性診断が自動実行できるのは非常に嬉しいポイントだと思いますので、興味ある方は是非パイプラインに組み込んでみては如何でしょうか。

F-Secure Radar API は非常に数があるため、触っていて面白いです。

本記事がどなたかのお役にたてば幸いです。

以上、AWS 事業本部の大前でした。

【F-Secure Radar API + CodePipeline】Webスキャンの実行と結果判定を自動化してみた

やること

使用するAPI

PUT /api/latest/vulnerabilityscans/{scanId}/start

GET /api/latest/vulnerabilityscans/{scanId}/info

GET /api/latest/webscans/{scanId}/reports/{reportId}/statistics/vulnerabilitiesandfindings

やってみた