Dredger × FutureVuls 連携ガイド：外部公開資産の"発見"から"脆弱性管理"までワンストップで効率化する

pip install requests
python main.py <JSONエクスポートファイルのパス>
または
uv run main.py <JSONエクスポートファイルのパス>

# /// script
# requires-python = ">=3.10"
# dependencies = [
#     "requests>=2.32.3",
# ]
# ///
import argparse
import json
from typing import TypedDict
import requests

VULS_ACCESS_TOKEN = "{FutureVulsのAPIアクセストークン}"

MAPPING_FILE = "domain_and_id_mapping.txt"
VULS_BASE_URL = "https://rest.vuls.biz"


class DredgerAsset(TypedDict):
    domain_name: str
    cves: list[str]


def add_pseudo_server(server_name: str):
    """
    FutureVuls APIを経由して疑似サーバを追加する関数
    """
    headers = {
        "Accept": "application/json",
        "Authorization": VULS_ACCESS_TOKEN,
        "Content-Type": "application/json",
    }
    data = {"serverName": server_name}
    response = requests.post(
        VULS_BASE_URL + "/v1/server/pseudo", headers=headers, data=json.dumps(data)
    )

    return response.json()


def add_scan_import_data(server_id: int, dredger_asset: DredgerAsset):
    """
    FutureVuls APIを経由して外部スキャンデータをインポートする関数
    """
    headers = {
        "Accept": "application/json",
        "Authorization": VULS_ACCESS_TOKEN,
        "Content-Type": "application/json",
    }
    data = {
        "cveList": dredger_asset["cves"],
        "isExternalScan": True,
        "scanImportName": dredger_asset["domain_name"],
        "serverID": server_id,
    }
    response = requests.post(
        VULS_BASE_URL + "/v1/scanImport", headers=headers, data=json.dumps(data)
    )

    return response.json()


def update_scan_import_data(scan_import_id: int, dredger_asset: DredgerAsset):
    """
    FutureVuls APIを経由して外部スキャンデータを更新する関数
    """
    headers = {
        "Accept": "application/json",
        "Authorization": VULS_ACCESS_TOKEN,
        "Content-Type": "application/json",
    }
    data = {"cveList": dredger_asset["cves"]}
    response = requests.put(
        VULS_BASE_URL + f"/v1/scanImport/{scan_import_id}",
        headers=headers,
        data=json.dumps(data),
    )

    return response.json()


def parse_dredger_json(filepath: str) -> list[DredgerAsset]:
    """
    DredgerのJSONファイルをパースしてドメイン名とCVE番号を抽出する関数
    """
    result_assets: list[DredgerAsset] = []
    with open(filepath) as f:
        json_dict: list[dict] = json.load(f)
        for asset in json_dict:
            domain_name = asset["domain_name"]
            cves: set[str] = set()
            for risk in asset["assets_on_risks"]:
                if risk["risk"]["categoryName"] == "CVE-REGISTERED-VULNERABILITY":
                    cves.add(risk["risk"]["id"])

            result_assets.append(
                {
                    "domain_name": domain_name,
                    "cves": sorted(list(cves)),
                }
            )
    return result_assets


def save_id_to_file(response: dict, domain_name: str):
    """
    POSTリクエストの結果からIDを取得し、ファイルに保存
    """
    scan_import_id = response.get("id")
    server_id = response.get("serverID")

    if isinstance(scan_import_id, int) and isinstance(server_id, int):
        with open(MAPPING_FILE, "a") as file:
            file.write(f"{domain_name} {scan_import_id} {server_id}")
        print(
            f"ドメイン： {domain_name} スキャン結果ID： {scan_import_id} サーバID： {server_id}をファイルに追記しました"
        )
    else:
        print("スキャン結果の登録に失敗しました")
        raise ValueError("スキャン結果の登録に失敗しました")


def read_ip_and_scan_import_id_from_file():
    """
    ファイルからドメイン名と外部スキャン結果のIDのペアを読み取る
    """
    ip_id_pairs: dict[str, int] = {}
    try:
        with open(MAPPING_FILE, "r") as file:
            for line in file:
                parts = line.strip().split()
                if len(parts) == 3:
                    domain_name, scan_import_id, server_id = parts
                    if scan_import_id.isdigit():
                        ip_id_pairs[domain_name] = int(scan_import_id)
                    else:
                        print(
                            f"無効な外部スキャン結果のIDが見つかりました： {scan_import_id}"
                        )
    except FileNotFoundError:
        pass
    except Exception as e:
        raise e

    return ip_id_pairs


def main():
    parser = argparse.ArgumentParser(
        description="DredgerのアセットJSONエクスポートデータから、FutureVulsの疑似サーバを登録し外部スキャンデータをインポートするスクリプト"
    )
    parser.add_argument(
        "export_file", help="DredgerのアセットJSONエクスポートファイルのパス"
    )

    args = parser.parse_args()

    filename: str = args.export_file

    assets = parse_dredger_json(filename)
    ip_id_pairs = read_ip_and_scan_import_id_from_file()
    for asset in assets:
        if scan_import_id := ip_id_pairs.get(asset["domain_name"]):
            # 実行履歴があった場合
            print(f"疑似サーバ{asset['domain_name']}の外部スキャンデータを更新します")
            update_scan_import_data(scan_import_id, asset)
            print("外部スキャンデータを更新しました")
        else:
            # 実行履歴が無かった場合
            print(f"{asset['domain_name']}の疑似サーバを作成します")
            add_pseudo_server_response = add_pseudo_server(asset["domain_name"])
            print("疑似サーバを作成しました")
            server_id = add_pseudo_server_response["id"]
            print("外部スキャンデータをインポートします")
            add_scan_import_data_response = add_scan_import_data(server_id, asset)
            print("外部スキャンデータをインポートしました")
            print("実行履歴データを記録します")
            save_id_to_file(add_scan_import_data_response, asset["domain_name"])
            print("実行履歴データを記録しました")


if __name__ == "__main__":
    main()