implementing-epss-score-for-vulnerability-prioritization

# 实施 EPSS 评分进行漏洞优先排序（Implementing EPSS Score for Vulnerability Prioritization）

## 概述

漏洞利用预测评分系统（Exploit Prediction Scoring System，EPSS）是由 FIRST（事件响应和安全团队论坛）开发的数据驱动模型，用于估算 CVE 在未来 30 天内在真实环境中被利用的概率。EPSS 使用基于真实世界漏洞利用数据训练的机器学习模型，产生 0.0 到 1.0（即 0% 到 100%）的评分。与衡量严重性的 CVSS 不同，EPSS 衡量的是被利用的可能性，这使其成为基于风险的漏洞优先排序的关键工具。

## 前置条件

- Python 3.9+，包含 `requests`、`pandas`、`matplotlib`
- 访问 FIRST EPSS API（https://api.first.org/data/v1/epss）
- 含 CVE 标识符的漏洞扫描结果
- 可选：用于 CVSS 丰富化的 NVD API 密钥

## EPSS API 用法

### 查询单个 CVE
```bash
# 获取特定 CVE 的 EPSS 评分
curl -s "https://api.first.org/data/v1/epss?cve=CVE-2024-3400" | python3 -m json.tool

# 响应示例：
# {
#   "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
```bash
# 最多批量查询 100 个 CVE
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}% 利用概率（百分位数：{item['percentile']}）\")
"
```

### 下载完整 EPSS 数据集
```bash
# 下载完整的每日 EPSS 评分（CSV 格式）
curl -s "https://epss.cyentia.com/epss_scores-current.csv.gz" | gunzip > epss_scores_current.csv

# 检查大小和预览
wc -l epss_scores_current.csv
head -5 epss_scores_current.csv
```

### 查询历史 EPSS 评分
```bash
# 获取特定日期的 EPSS 评分
curl -s "https://api.first.org/data/v1/epss?cve=CVE-2024-3400&date=2024-04-12"

# 获取时间序列数据
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%**：低利用概率；在正常补丁周期中处理

## 实现

```python
import requests
import pandas as pd
from datetime import datetime

def fetch_epss_scores(cve_list):
    """从 FIRST API 批量获取 CVE 列表的 EPSS 评分。"""
    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):
    """用 EPSS 评分丰富扫描结果并分配优先级。"""
    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"[+] 已对 {len(df)} 个漏洞进行优先排序 -> {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 趋势分析

```python
def fetch_epss_timeseries(cve_id):
    """获取历史 EPSS 评分用于趋势分析。"""
    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):
    """检测 EPSS 评分显著上升，指示新兴威胁。"""
    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"[!] 检测到 {cve_id} 的 EPSS 激增：{previous:.3f} -> {latest:.3f} (+{increase:.3f})")
        return True
    return False
```

## 参考资料

- [FIRST EPSS 官网](https://www.first.org/epss/)
- [EPSS API 文档](https://www.first.org/epss/api)
- [EPSS 模型文档](https://www.first.org/epss/model)
- [EPSS 数据下载](https://epss.cyentia.com/)
- [Cyentia Institute 研究](https://www.cyentia.com/epss/)