用 Python 打造漏洞情报流水线：组件清单与 CVE 交叉匹配实战

场景

NVD 每日新增条目数量会变化。安全服务的实际问题从来不是”没有漏洞信息”，而是：

1. 当天新增的几百条 CVE 里，哪些与客户在用组件有关？
2. 命中之后，哪些是真实命中，哪些是关键词误报？
3. 筛完之后，怎么快速形成可报送的通报？

这篇文章分享一套我在驻场期间打磨的流水线：多源拉取 → 组件交叉匹配 → 去噪 → 人工复核 → 通报输出。

一、多源聚合

单一数据源不够用。各数据源的角色分工：

数据源	用途
NVD 2.0 API	主查询源：CVSS、CPE、CWE、参考链接
CVE.org	详情入口链接
FIRST EPSS	利用概率参考（0-1，越高越可能被利用）
CISA KEV	是否在已知被在野利用目录中
阿里云 AVD	中文详情页
奇安信 TI	漏洞库入口

NVD 2.0 API 查询要点：

# NVD API 按发布时间过滤
# 注意时区：北京时间当天 = UTC 前一天 16:00 起
import requests

def query_nvd(start_utc, end_utc):
    """演示分页逻辑；生产使用时还需限速、重试、去重与审计记录。"""
    url = "https://services.nvd.nist.gov/rest/json/cves/2.0"
    params = {
        "pubStartDate": start_utc,   # 如 "2026-08-20T16:00:00.000"
        "pubEndDate": end_utc,
        "resultsPerPage": 2000,
        "startIndex": 0,
    }
    items = []
    with requests.Session() as session:
        while True:
            r = session.get(url, params=params, timeout=30)
            r.raise_for_status()
            data = r.json()
            batch = data.get("vulnerabilities", [])
            total = int(data["totalResults"])
            if not batch:
                if params["startIndex"] < total:
                    raise RuntimeError("分页提前结束，不能作为完整查询结果")
                break
            items.extend(batch)
            params["startIndex"] += len(batch)
            if params["startIndex"] >= total:
                break
    return items

两个必踩的坑：

1. 分页截断：不能假定单次响应就是全部结果。应根据 totalResults、startIndex 和实际返回数量循环翻页，并处理接口限速和错误；即使设置较大的 resultsPerPage，也仍需检查是否完整。
2. 时区换算：北京时间”当天”对应 UTC 是前一天 16:00 起。日报口径必须在文档里同时写明北京时间窗口和实际使用的 UTC 参数，否则复盘时对不上数。

二、组件清单 × CVE 交叉匹配

匹配输入是经授权整理的在用组件清单，覆盖数据库/搜索、容器云原生、Java 中间件、语言运行时等。公开示例不包含客户资产明细或真实清单规模。

匹配字段取三个：CVE 的 description（描述）、CPE、affected product。三字段命中任意一个即视为候选。

下面的 match 接收的是预先归一化的记录，不是 NVD 原始响应。需要先从 cve.descriptions 提取描述，从 cve.configurations 中递归提取 CPE 与版本条件，并按数据源结构整理产品字段；不能把原始嵌套 JSON 直接传入这个函数。

def match(cve, components):
    text = " ".join([
        cve.get("description", ""),
        cve.get("cpe", ""),
        cve.get("affected_product", ""),
    ]).lower()
    return [c for c in components if c.lower() in text]

去噪：比匹配更重要的环节

直接跑上面的代码，会被短词关键词干掉准确性。实测的坑：

• 组件清单里的 R 会命中所有描述里含字母 r 的 CVE
• go 会命中 “Google””MongoDB””Cargo”……
• Git 会命中 “Digital””Legit”

处理方式：

1. 短词白名单跳过：长度 ≤ 3 的组件名（R、go、gcc、php 等）不做子串匹配，改用词边界正则或直接人工核对；
2. 大小写与词形：react 要匹配 “React” 但不能匹配 “reaction”——用 \b 词边界；
3. 结果复核：对命中统计做分布检查，某个组件命中数异常高（如一次命中 30+ 条），大概率是短词误报。

三、输出格式示例

以下数字为虚构演示值，只说明报表结构，不代表真实项目资产或某个时间窗口的查询结果：

NVD 时间窗口总 CVE：500 条
命中组件相关 CVE：20 条
其中 CRITICAL 2 条，HIGH 8 条

命中分布（前几名）：
示例组件 A：8、示例组件 B：5、示例组件 C：3、其他：4

输出按 CVSS 降序排列的优先关注清单，每条包含：命中组件、发布时间、CVSS 分值与向量、摘要、NVD/CVE.org/AVD/TI 链接、参考链接。

四、人工复核：脚本筛完 ≠ 结论

这是整个流水线里最不能省的一步。自动命中数量不等于需要处置的漏洞数量。复核要点：

1. 核对真实产品名：CVE 描述里的 “Kubernetes” 可能是某个依赖 Kubernetes 的第三方项目（如 Prowler 的 kubeconfig 处理问题），不是 Kubernetes 本身的漏洞——不能只看命中的关键词标签。
2. 核对版本范围：CPE 的 affected range 与客户实际版本比对。例如某次 MongoDB 相关情报命中 30+ 条，逐条核对版本后确认客户版本不在受影响范围，无需紧急处置——这一条结论比 30 条命中更有价值。
3. 可利用性判断：结合资产重要性、可达性、攻击前提和补偿控制排序。未认证远程利用通常优先；DoS、需要用户交互或特定条件的漏洞不能机械降级，关键业务可用性和真实使用场景同样重要。
4. EPSS 与 KEV 加权：EPSS 高分或在 CISA KEV 目录中的，处置优先级上调。

五、筛选标准

只保留”有实战威胁”的条目，不凑数：

保留：未认证/低门槛认证的 RCE、命令注入、反序列化、认证绕过、严重越权、影响数据 SQL 注入、高危文件上传/读取、可打内网的 SSRF。

降低优先级或转专项复核：纯版本信息、需要用户交互、DoS 或特定场景利用等条目，应核对实际影响后决定。不得仅按漏洞类型排除，也不能把“本轮不优先通报”等同于“不受影响”。

六、产出与闭环

最终形成三件东西：

1. Markdown 情报清单（结构化：组件/时间/CVSS/摘要/链接），可直接进日报；
2. Word/PDF 通报（面向客户的正式版本）；
3. 留档数据：原始查询结果 + 匹配脚本 + 匹配结果，保证每条结论可溯源回查。

流程闭环：情报收集 → 影响判断 → 结果报送。第二天复盘时能回答”这条 CVE 为什么报/为什么不报”，这才是情报工作的可信度来源。

总结

• 情报流水线的难点不在”拉数据”，在去噪和人工复核；
• 时区和分页是两个最隐蔽的技术坑；
• 命中数不是 KPI，结论准确才是——30 条命中核对后”不受影响”的结论，比 30 条原文粘贴有价值得多。

参考资料

• NVD 漏洞 API 文档
• NVD API 使用建议