implementing-epss-score-for-vulnerability-prioritization

脆弱性の優先順位付けのための EPSS スコアの実装

概要

Exploit Prediction Scoring System（EPSS）は、FIRST（Forum of Incident Response and Security Teams）が開発したデータ駆動型のモデルで、今後 30 日以内に CVE が実際に悪用される確率を推定します。EPSS は実世界の悪用データで訓練された機械学習を用いて、0.0 から 1.0（0% から 100%）のスコアを算出します。深刻度を測定する CVSS とは異なり、EPSS は悪用の可能性を測定するため、リスクベースの脆弱性優先順位付けに不可欠です。

使用するタイミング

• 環境において、脆弱性の優先順位付けのための EPSS スコア実装機能をデプロイまたは設定するとき
• コンプライアンス要件に沿ったセキュリティ統制を確立するとき
• この領域のセキュリティアーキテクチャを構築または改善するとき
• この実装が必要なセキュリティアセスメントを実施するとき

前提条件

• requests、pandas、matplotlib がインストールされた Python 3.9 以降
• FIRST EPSS API（https://api.first.org/data/v1/epss）へのアクセス
• CVE 識別子を含む脆弱性スキャン結果
• 任意：CVSS の情報を補完するための NVD API キー

EPSS API の使用方法

単一の CVE を照会する

# Get EPSS score for a specific CVE
curl -s "https://api.first.org/data/v1/epss?cve=CVE-2024-3400" | python3 -m json.tool

# Response:
# {
#   "status": "OK",
#   "status-code": 200,
#   "version": "1.0",
#   "total": 1,
#   "data": [
#     {
#       "cve": "CVE-2024-3400",
#       "epss": "0.95732",
#       "percentile": "0.99721",
#       "date": "2024-04-15"
#     }
#   ]
# }

複数の CVE を照会する

# Batch query up to 100 CVEs
curl -s "https://api.first.org/data/v1/epss?cve=CVE-2024-3400,CVE-2024-21887,CVE-2023-44228" | \
  python3 -c "
import sys, json
data = json.load(sys.stdin)
for item in data['data']:
    pct = float(item['epss']) * 100
    print(f\"{item['cve']}: {pct:.2f}% exploitation probability (percentile: {item['percentile']})\")
"

EPSS データセット全体をダウンロードする

# Download complete daily EPSS scores (CSV format)
curl -s "https://epss.cyentia.com/epss_scores-current.csv.gz" | gunzip > epss_scores_current.csv

# Check size and preview
wc -l epss_scores_current.csv
head -5 epss_scores_current.csv

EPSS の過去スコアを照会する

# Get EPSS score for a specific date
curl -s "https://api.first.org/data/v1/epss?cve=CVE-2024-3400&date=2024-04-12"

# Get time series data
curl -s "https://api.first.org/data/v1/epss?cve=CVE-2024-3400&scope=time-series"

優先順位付けの戦略

EPSS と CVSS の組み合わせアプローチ

EPSS スコア	CVSS スコア	優先度	対応
> 0.7	>= 9.0	P0 - 即時	24 時間以内に修正
> 0.7	>= 7.0	P1 - 緊急	48 時間以内に修正
> 0.4	>= 7.0	P2 - 高	7 日以内に修正
> 0.1	>= 4.0	P3 - 中	30 日以内に修正
<= 0.1	>= 7.0	P3 - 中	30 日以内に修正
<= 0.1	< 7.0	P4 - 低	90 日以内に修正

EPSS パーセンタイルのしきい値

• 上位 1%（パーセンタイル >= 0.99）：悪用される可能性が極めて高いため、重大として扱う
• 上位 5%（パーセンタイル >= 0.95）：悪用確率が高いため、修正を優先する
• 上位 10%（パーセンタイル >= 0.90）：リスクが高まっているため、近いうちに修正を計画する
• 下位 50%：悪用確率は低いため、通常のパッチサイクルで対応する

実装

import requests
import pandas as pd
from datetime import datetime

def fetch_epss_scores(cve_list):
    """Fetch EPSS scores for a list of CVEs from FIRST API."""
    scores = {}
    batch_size = 100
    for i in range(0, len(cve_list), batch_size):
        batch = cve_list[i:i + batch_size]
        resp = requests.get(
            "https://api.first.org/data/v1/epss",
            params={"cve": ",".join(batch)},
            timeout=30
        )
        if resp.status_code == 200:
            for entry in resp.json().get("data", []):
                scores[entry["cve"]] = {
                    "epss": float(entry["epss"]),
                    "percentile": float(entry["percentile"]),
                    "date": entry.get("date", ""),
                }
    return scores

def prioritize_vulnerabilities(scan_results_csv, output_csv):
    """Enrich scan results with EPSS scores and assign priorities."""
    df = pd.read_csv(scan_results_csv)
    cve_list = df["cve_id"].dropna().unique().tolist()

    epss_data = fetch_epss_scores(cve_list)

    df["epss_score"] = df["cve_id"].map(lambda c: epss_data.get(c, {}).get("epss", 0))
    df["epss_percentile"] = df["cve_id"].map(lambda c: epss_data.get(c, {}).get("percentile", 0))

    def assign_priority(row):
        epss = row.get("epss_score", 0)
        cvss = row.get("cvss_score", 0)
        if epss > 0.7 and cvss >= 9.0:
            return "P0"
        if epss > 0.7 and cvss >= 7.0:
            return "P1"
        if epss > 0.4 and cvss >= 7.0:
            return "P2"
        if epss > 0.1 or cvss >= 7.0:
            return "P3"
        return "P4"

    df["priority"] = df.apply(assign_priority, axis=1)
    df = df.sort_values(["priority", "epss_score"], ascending=[True, False])
    df.to_csv(output_csv, index=False)
    print(f"[+] Prioritized {len(df)} vulnerabilities -> {output_csv}")
    print(f"    P0: {len(df[df['priority']=='P0'])}")
    print(f"    P1: {len(df[df['priority']=='P1'])}")
    print(f"    P2: {len(df[df['priority']=='P2'])}")
    print(f"    P3: {len(df[df['priority']=='P3'])}")
    print(f"    P4: {len(df[df['priority']=='P4'])}")
    return df

EPSS トレンド分析

def fetch_epss_timeseries(cve_id):
    """Get historical EPSS scores for trend analysis."""
    resp = requests.get(
        "https://api.first.org/data/v1/epss",
        params={"cve": cve_id, "scope": "time-series"},
        timeout=30
    )
    if resp.status_code == 200:
        return resp.json().get("data", [])
    return []

def detect_epss_spikes(cve_id, threshold=0.3):
    """Detect significant EPSS score increases indicating emerging threats."""
    timeseries = fetch_epss_timeseries(cve_id)
    if len(timeseries) < 2:
        return False
    sorted_data = sorted(timeseries, key=lambda x: x.get("date", ""))
    latest = float(sorted_data[-1].get("epss", 0))
    previous = float(sorted_data[-2].get("epss", 0))
    increase = latest - previous
    if increase >= threshold:
        print(f"[!] EPSS spike detected for {cve_id}: {previous:.3f} -> {latest:.3f} (+{increase:.3f})")
        return True
    return False

参考資料

• FIRST EPSS 公式
• EPSS API ドキュメント
• EPSS モデルのドキュメント
• EPSS データのダウンロード
• Cyentia Institute の研究