1. 项目概述为什么我们需要一个自动化的NVD漏洞监控器在安全运维和漏洞管理的日常工作中我经常需要关注特定软件或组件的最新安全漏洞。手动去NVDNational Vulnerability Database官网翻找不仅效率低下还容易遗漏。特别是当我们需要精准定位到带有CPECommon Platform Enumeration信息的漏洞时手动筛选更是耗时费力。CPE信息就像是漏洞的“身份证”它精确地告诉我们这个漏洞影响的是哪个厂商、哪个产品、哪个版本这对于资产梳理和精准修复至关重要。于是我决定用Python写一个脚本让它自动调用NVD的官方API每天帮我抓取最新发布的、并且带有CPE信息的漏洞数据。这不仅仅是写几行请求代码那么简单它涉及到对NVD API接口的深入理解、对返回的复杂JSON数据的解析、对CPE字符串的拆解以及如何设计一个稳定、高效且易于维护的自动化流程。这个脚本后来成了我们团队内部的一个小工具今天我就把它的核心实现思路、踩过的坑以及一些实用的技巧分享出来希望能给同样有漏洞监控需求的朋友们一些参考。2. 核心思路与NVD API深度解析2.1 NVD API你的漏洞数据金矿NVD提供了免费的RESTful API供开发者使用。对于我们的目标——获取“最新发布”且“带有CPE信息”的漏洞主要会用到两个核心接口漏洞摘要接口 (/rest/json/cves/2.0): 这是获取漏洞列表的主要入口。它支持通过多种参数进行筛选例如发布时间范围、漏洞严重等级CVSS分数、关键词等。我们将主要利用它的lastModStartDate和lastModEndDate参数来获取在特定时间段内修改包括新发布的漏洞。单个漏洞详情接口 (/rest/json/cves/2.0/{cveId}): 获取单个CVE编号的完整详细信息。我们通过第一个接口拿到CVE ID列表后如果需要更详细的信息虽然摘要接口返回的数据通常已包含CPE可以再调用此接口。关键设计决策为什么选择“最后修改日期”而非“发布时间”NVD的漏洞数据是动态更新的。一个漏洞可能先被披露几天后NVD才收录并分配CPE信息。如果我们只按“发布时间”筛选可能会漏掉那些发布时没有CPE、但后续更新时补上了CPE的漏洞。使用lastModStartDate则可以抓取到所有在指定时间段内发生过数据更新的漏洞记录确保我们能捕获到CPE信息被添加或更新的情况这比单纯看“新发布”更全面、更可靠。2.2 CPE信息从字符串到结构化数据CPE 2.3格式的字符串看起来像这样cpe:2.3:a:microsoft:internet_explorer:11.0.9600.16384:*:*:*:*:*:*:*。 这串字符是有固定结构的各部分以冒号分隔分别代表了cpe:2.3 标准标识和版本。a 应用层其他还有o操作系统h硬件。microsoft 厂商。internet_explorer 产品名称。11.0.9600.16384 版本号。后续的*分别代表了更新版本、版本类型等更多属性。我们的脚本不仅要能判断一个漏洞是否包含CPE信息即configurations字段不为空更需要能从中解析出具体的厂商、产品、版本这样才能与我们自己的资产清单进行匹配。例如我们可能只关心a:apache:tomcat或a:openssl:openssl相关的漏洞。2.3 整体架构设计脚本的核心工作流可以概括为以下几步确定时间窗口 计算需要查询的时间范围通常是“从昨天凌晨到现在”或者最近24小时。调用摘要API 向NVD API发起请求获取该时间段内所有更新的漏洞摘要列表。初步过滤 遍历列表筛选出configurations字段不为空的条目即含有CPE信息。数据解析与丰富 对筛选出的漏洞解析其CPE信息并可以根据需要选择性地调用详情API获取更完整的描述、参考链接等。结果输出与持久化 将结果以结构化的格式如JSON文件、CSV表格或存入数据库保存下来并可以设置邮件、钉钉/飞书机器人等通知机制。注意API速率限制。NVD的公开API在没有API密钥的情况下每小时有请求次数限制通常为每分钟5次每小时50次。对于批量查询强烈建议申请一个免费的API密钥并在请求头中带上它这样可以将限制提升到每分钟50次完全满足日常监控需求。申请过程很简单在NVD官网的“Developers”板块即可完成。3. 环境准备与核心依赖库3.1 Python环境与包管理这个项目对Python版本要求不高Python 3.6及以上即可。我习惯使用venv创建独立的虚拟环境避免包冲突。# 创建项目目录并进入 mkdir nvd-cve-monitor cd nvd-cve-monitor # 创建虚拟环境 python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate3.2 安装必要的库我们主要需要以下几个库requests: 用于发送HTTP请求到NVD API这是最核心的库。pandas: 非必须但强烈推荐。用于将获取到的漏洞数据整理成表格方便筛选、分析和导出为Excel/CSV。python-dotenv: 非必须但推荐。用于管理API密钥等敏感配置避免将其硬编码在脚本中。使用pip一键安装pip install requests pandas python-dotenv3.3 配置文件管理创建一个.env文件来存储你的NVD API密钥如果你有的话和其他配置。永远不要将密钥提交到代码仓库。# .env 文件内容示例 NVD_API_KEYyour_actual_api_key_here TIME_WINDOW_HOURS24 RESULTS_PER_PAGE2000然后在Python脚本中这样加载from dotenv import load_dotenv import os load_dotenv() # 加载 .env 文件中的变量到环境变量 API_KEY os.getenv(NVD_API_KEY)4. 分步实现构建你的漏洞监控脚本4.1 步骤一构造精准的API请求首先我们需要一个函数来构建请求URL和头部信息。NVD API的Base URL是https://services.nvd.nist.gov。import requests from datetime import datetime, timedelta import time def build_cve_query_url(start_date, end_date, start_index0): 构建查询CVE列表的URL。 参数: start_date (str): 开始日期格式 YYYY-MM-DDTHH:MM:SSZ end_date (str): 结束日期格式同上 start_index (int): 分页起始索引默认0 返回: str: 完整的API请求URL base_url https://services.nvd.nist.gov/rest/json/cves/2.0 # 关键参数lastModStartDate 和 lastModEndDate url (f{base_url}?lastModStartDate{start_date} flastModEndDate{end_date} fstartIndex{start_index}) # 如果有API密钥通常作为请求头传递但NVD API也支持作为参数 apiKey # 更常见的做法是放在请求头 apiKey 中但NVD文档建议用参数。这里按参数处理。 if API_KEY: url fapiKey{API_KEY} return url def make_api_request(url): 发送HTTP GET请求并处理可能的错误和速率限制。 参数: url (str): 请求的URL 返回: dict: 解析后的JSON响应数据 headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36 } try: response requests.get(url, headersheaders, timeout30) response.raise_for_status() # 如果状态码不是200抛出HTTPError异常 return response.json() except requests.exceptions.HTTPError as http_err: # 特别注意429状态码请求过多 if response.status_code 429: retry_after int(response.headers.get(Retry-After, 60)) print(f触发速率限制等待 {retry_after} 秒后重试...) time.sleep(retry_after) return make_api_request(url) # 简单重试一次 else: print(fHTTP错误发生: {http_err}) print(f响应内容: {response.text}) return None except requests.exceptions.RequestException as err: print(f请求异常: {err}) return None关键点解析日期格式 NVD API要求UTC时间的ISO 8601格式即YYYY-MM-DDTHH:MM:SSZ。T是日期和时间的分隔符Z代表UTC时区。User-Agent 虽然NVD API没有强制要求但设置一个合理的User-Agent是良好的网络公民行为。错误处理 重点处理了429状态码Too Many Requests。NVD API会在响应头中通过Retry-After告诉你需要等待多少秒。我们的代码捕获这个异常并自动等待后重试一次。API密钥使用 这里展示了将API密钥作为查询参数传递的方式。请务必遵循NVD官方文档的最新说明。4.2 步骤二处理分页与获取全部数据NVD API一次返回的数据量有限制默认20条可通过参数调整但最大可能为2000。我们需要处理分页直到获取所有结果。def fetch_all_cves_within_timeframe(start_dt, end_dt): 获取指定时间段内所有的CVE记录处理分页。 参数: start_dt (datetime): 开始时间对象 end_dt (datetime): 结束时间对象 返回: list: 包含所有CVE摘要信息的字典列表 all_cves [] start_index 0 results_per_page 2000 # 每页最大数量 date_format %Y-%m-%dT%H:%M:%SZ start_date_str start_dt.strftime(date_format) end_date_str end_dt.strftime(date_format) print(f正在查询从 {start_date_str} 到 {end_date_str} 的漏洞更新...) while True: url build_cve_query_url(start_date_str, end_date_str, start_index) print(f正在获取索引 {start_index} 开始的数据...) data make_api_request(url) if not data: print(获取数据失败终止。) break vulnerabilities data.get(vulnerabilities, []) if not vulnerabilities: print(未找到更多漏洞数据。) break all_cves.extend(vulnerabilities) total_results data.get(totalResults, 0) results_this_page len(vulnerabilities) start_index results_this_page print(f已获取 {len(all_cves)} / {total_results} 条记录。) # 如果本次返回的数量小于请求的数量或者已获取数量等于总数说明已到最后一页 if results_this_page results_per_page or len(all_cves) total_results: break # 礼貌性暂停避免给API服务器造成压力即使有API Key time.sleep(1) print(f查询完成共获取 {len(all_cves)} 条CVE记录。) return all_cves实操心得分页逻辑 NVD API使用startIndex和返回的totalResults来控制分页。循环请求每次将startIndex增加已获取的记录数直到获取的记录数达到totalResults或单次返回的记录数小于请求数。礼貌性暂停 即使在速率限制内在请求间添加一个短暂的休眠如time.sleep(1)也是一个好习惯体现了对公共API资源的尊重。4.3 步骤三筛选与解析CPE信息这是整个脚本的核心逻辑。我们需要遍历获取到的所有CVE检查其是否包含配置信息CPE并从中提取出我们关心的部分。def parse_cpe_uri(cpe_uri): 解析CPE 2.3格式的URI提取关键组件。 参数: cpe_uri (str): 例如 cpe:2.3:a:microsoft:internet_explorer:11.0.9600.16384:*:*:*:*:*:*:* 返回: dict: 包含解析后字段的字典如 {vendor: microsoft, product: internet_explorer, version: 11.0.9600.16384} if not cpe_uri.startswith(cpe:2.3:): return None parts cpe_uri.split(:) # 确保有足够的部分 if len(parts) 6: return None # 根据CPE 2.3规范索引位置是固定的 cpe_dict { part: parts[2], # a, o, h vendor: parts[3], product: parts[4], version: parts[5], update: parts[6] if len(parts) 6 else *, # 可以继续解析更多字段... } # 处理可能存在的转义字符如下划线 for key in [vendor, product]: cpe_dict[key] cpe_dict[key].replace(\\, ) return cpe_dict def filter_and_parse_cves(vulnerabilities, filter_cpe_partNone, filter_vendorNone, filter_productNone): 过滤出带有CPE信息的漏洞并解析CPE详情。 参数: vulnerabilities (list): fetch_all_cves_within_timeframe 返回的漏洞列表 filter_cpe_part (str): 可选按CPE类型过滤如 a (应用), o (操作系统) filter_vendor (str): 可选按厂商过滤如 apache filter_product (str): 可选按产品过滤如 tomcat 返回: list: 过滤并丰富后的漏洞信息列表 filtered_cves [] for vuln in vulnerabilities: cve_item vuln.get(cve, {}) cve_id cve_item.get(id) # 1. 检查是否存在配置信息即CPE configurations cve_item.get(configurations, []) if not configurations: continue # 没有CPE信息跳过 # 2. 提取所有CPE all_cpes [] for config in configurations: nodes config.get(nodes, []) for node in nodes: cpe_matches node.get(cpeMatch, []) for match in cpe_matches: cpe_uri match.get(criteria) vulnerable match.get(vulnerable, False) # 通常我们只关心标记为易受攻击vulnerableTrue的CPE if cpe_uri and vulnerable: all_cpes.append(cpe_uri) if not all_cpes: continue # 虽然有configurations节点但没有标记为vulnerable的CPE也跳过 # 3. 解析CPE并应用过滤条件 parsed_cpes [] for cpe_uri in all_cpes: parsed parse_cpe_uri(cpe_uri) if not parsed: continue # 应用过滤条件 if filter_cpe_part and parsed[part] ! filter_cpe_part: continue if filter_vendor and filter_vendor.lower() not in parsed[vendor].lower(): continue if filter_product and filter_product.lower() not in parsed[product].lower(): continue parsed_cpes.append(parsed) # 如果过滤后没有CPE了则跳过此CVE if not parsed_cpes: continue # 4. 收集该CVE的其他有用信息 descriptions cve_item.get(descriptions, []) # 优先取英文描述 en_description next((desc[value] for desc in descriptions if desc.get(lang) en), ) metrics cve_item.get(metrics, {}) cvss_metric_v3 metrics.get(cvssMetricV31, metrics.get(cvssMetricV30, metrics.get(cvssMetricV2, []))) base_score None severity None if cvss_metric_v3: cvss_data cvss_metric_v3[0].get(cvssData, {}) base_score cvss_data.get(baseScore) severity cvss_data.get(baseSeverity) # 5. 构建最终输出条目 filtered_cve { cve_id: cve_id, published_date: cve_item.get(published, ), last_modified_date: cve_item.get(lastModified, ), description: en_description[:500], # 截取前500字符 cvss_score: base_score, severity: severity, cpe_count: len(parsed_cpes), parsed_cpes: parsed_cpes, # 保存解析后的CPE对象列表 cpe_uris: all_cpes, # 保存原始的CPE URI列表 references: [ref.get(url) for ref in cve_item.get(references, [])][:5] # 取前5个参考链接 } filtered_cves.append(filtered_cve) print(f过滤完成共找到 {len(filtered_cves)} 个带有CPE信息且符合过滤条件的漏洞。) return filtered_cves深度解析vulnerable字段 在cpeMatch中vulnerable字段至关重要。它为True表示该CPE条目是受此漏洞影响的版本为False则表示是不受影响的版本例如修复后的版本。我们的脚本通常只关心vulnerableTrue的条目。CPE解析的健壮性parse_cpe_uri函数需要处理各种边界情况比如字符串格式不正确、部分字段缺失等。这里展示的是一个基础解析在实际生产中你可能需要处理更复杂的CPE表达式比如带有逻辑运算符如cpe:2.3:a:*:mysql:*:*:*:*:*:*:*:*这样的通配符。过滤逻辑 过滤可以在获取数据后如本函数也可以通过API参数在请求时进行NVD API支持cpeName参数。对于监控特定产品如Apache Tomcat在请求时过滤效率更高能减少不必要的数据传输。但对于需要分析多种产品的场景先获取再过滤更灵活。4.4 步骤四数据输出与持久化获取并处理完数据后我们需要将其保存下来。这里提供几种常见方式。import pandas as pd import json from datetime import datetime def save_results(filtered_cves, output_formatjson): 将过滤后的漏洞数据保存到文件。 参数: filtered_cves (list): filter_and_parse_cves 返回的列表 output_format (str): 输出格式支持 json, csv, excel if not filtered_cves: print(没有数据需要保存。) return timestamp datetime.now().strftime(%Y%m%d_%H%M%S) if output_format json: filename fcve_with_cpe_{timestamp}.json with open(filename, w, encodingutf-8) as f: # 处理无法序列化的对象如果需要的话这里parsed_cpes是dict列表可序列化 json.dump(filtered_cves, f, indent2, ensure_asciiFalse) print(f结果已保存为 JSON 文件: {filename}) elif output_format in [csv, excel]: # 为了输出到表格需要将嵌套的列表结构扁平化 flat_data [] for cve in filtered_cves: base_info { CVE ID: cve[cve_id], 发布时间: cve[published_date], 最后修改: cve[last_modified_date], 描述摘要: cve[description], CVSS分数: cve[cvss_score], 严重等级: cve[severity], CPE数量: cve[cpe_count] } # 将每个CPE展开为一行 for i, cpe in enumerate(cve[parsed_cpes]): row base_info.copy() row[CPE序号] i1 row[CPE类型] cpe.get(part) row[厂商] cpe.get(vendor) row[产品] cpe.get(product) row[版本] cpe.get(version) row[更新版本] cpe.get(update) flat_data.append(row) df pd.DataFrame(flat_data) if output_format csv: filename fcve_with_cpe_{timestamp}.csv df.to_csv(filename, indexFalse, encodingutf-8-sig) # utf-8-sig支持Excel中文 print(f结果已保存为 CSV 文件: {filename}) else: # excel filename fcve_with_cpe_{timestamp}.xlsx df.to_excel(filename, indexFalse) print(f结果已保存为 Excel 文件: {filename}) else: print(f不支持的输出格式: {output_format})技巧分享数据扁平化 为了便于在Excel或CSV中查看我们将一个CVE对应多个CPE的情况“展开”每个CPE占一行并重复CVE的基础信息。这在数据分析时非常直观。文件名时间戳 使用时间戳命名文件可以避免覆盖历史数据也便于追溯。UTF-8编码 保存CSV时使用utf-8-sig编码可以确保在Microsoft Excel中直接打开时中文字符能正常显示。4.5 步骤五组装主程序与设置定时任务最后我们将所有功能整合到一个主函数中并可以配置为定时运行。def main(hours_back24, filter_productNone): 主函数执行完整的漏洞抓取、过滤和保存流程。 参数: hours_back (int): 查询多少小时之前到现在的数据。 filter_product (str): 可选过滤产品名称如 tomcat print( NVD CVE 监控脚本开始运行 ) # 1. 计算时间范围 end_time datetime.utcnow() start_time end_time - timedelta(hourshours_back) # 2. 获取所有CVE all_vulns fetch_all_cves_within_timeframe(start_time, end_time) if not all_vulns: print(在指定时间段内未找到任何漏洞更新。) return # 3. 过滤并解析带CPE的漏洞 filtered filter_and_parse_cves(all_vulns, filter_productfilter_product) # 4. 保存结果 if filtered: # 可以同时保存多种格式 save_results(filtered, json) save_results(filtered, csv) # 这里可以添加通知逻辑例如发送邮件 # send_email_notification(filtered) else: print(f在 {hours_back} 小时内未发现带有CPE信息且符合过滤条件的新漏洞。) print( 脚本运行结束 ) if __name__ __main__: # 示例监控过去24小时内与‘tomcat’相关的漏洞 main(hours_back24, filter_producttomcat)实现自动化 在Linux服务器上你可以使用cron定时任务在Windows上可以使用“任务计划程序”。一个典型的cron配置每天上午9点运行0 9 * * * cd /path/to/your/script /path/to/your/venv/bin/python nvd_monitor.py /path/to/log/cve_monitor.log 215. 高级技巧与避坑指南5.1 处理复杂的CPE匹配逻辑NVD返回的CPE信息有时不是简单的单个字符串而是嵌套的、带有逻辑运算符AND, OR的复杂节点树。我们的filter_and_parse_cves函数中的遍历逻辑遍历nodes和cpeMatch已经能处理大部分情况。但对于更复杂的CVE-2021-44228(Log4Shell) 这样的漏洞其影响范围可能通过复杂的逻辑组合来定义。一个健壮的解析器可能需要递归遍历整个节点树并解析operator字段。对于大多数监控场景我们目前的扁平化提取vulnerableTrue的CPE已经足够但了解这一点对于深度分析很重要。5.2 优化性能与应对API限制增量查询 不要每次都查询过去24小时的全量数据。可以将上次成功运行的时间戳记录在一个文件中下次查询时只查询从那个时间戳到现在的新数据。这能大幅减少请求数据量也符合API调用的最佳实践。缓存机制 对于需要频繁查询的CVE详情比如在生成报告时需要更完整的描述可以考虑将结果缓存在本地数据库如SQLite中避免重复请求。错误重试与退避 我们之前实现了对429错误的简单重试。在生产环境中应该实现更完善的退避算法如指数退避并对连接超时、服务器错误5xx等进行重试。5.3 数据丰富与关联单纯的CPE信息有时不够。你可以考虑将获取到的CVE数据与其它源关联关联Exploit DB 检查是否有公开的漏洞利用代码Exploit。关联安全公告 获取厂商的官方安全公告链接。内部资产关联 这是最关键的一步。将解析出的vendor和product与你内部的CMDB配置管理数据库或资产清单进行匹配就能立即知道这个漏洞影响了你哪些具体的服务器或应用实现从“漏洞预警”到“风险定位”的跨越。5.4 常见问题排查FAQQ: 脚本运行后返回“未找到任何漏洞更新”但我确定应该有新漏洞。A:首先检查你使用的lastModStartDate和lastModEndDate时间格式是否正确必须是UTC的ISO格式。其次确认时间窗口是否合理。NVD数据处理有延迟刚披露的漏洞可能几分钟到几小时后才会出现在API中。可以尝试将hours_back参数调大一些比如48小时。Q: 获取到的CPE信息中vulnerable字段全是False导致过滤后没有数据。A:这可能是正常的。有些漏洞条目在初始收录时可能只包含了“不受影响”的CPE配置。或者你查询的漏洞其影响范围是通过更复杂的逻辑定义的而不是简单的cpeMatch列表。可以尝试暂时取消vulnerableTrue的过滤条件查看原始数据结构和operator字段。Q: 请求总是失败返回403或429错误。A:403错误可能是请求头或参数格式问题。429错误肯定是触发了速率限制。确认你是否使用了API密钥并且密钥有效。检查你的请求频率。即使有API密钥也要遵守每分钟/每小时的限制。在循环请求中添加time.sleep()。检查User-Agent是否被设置有些公开API会拒绝空User-Agent的请求。Q: 解析CPE字符串时出错提示索引超出范围。A:并非所有CPE URI都严格遵循cpe:2.3:a:vendor:product:version:...的格式。有些可能版本字段为空:或者包含其他变体。在parse_cpe_uri函数中增加更严格的长度检查和异常捕获try...except对解析失败的CPE记录日志而不是直接崩溃。Q: 输出的CSV文件在Excel中打开中文乱码。A:确保使用utf-8-sig编码保存CSV文件。pandas的to_csv方法使用encodingutf-8-sig参数即可解决。6. 从脚本到工具构建一个简单的Web仪表板当脚本稳定运行后你可以考虑将其升级为一个内部工具。使用轻量级的Web框架如Flask或FastAPI可以快速搭建一个仪表板。# 这是一个非常简化的 Flask 应用示例 from flask import Flask, render_template, jsonify import sqlite3 app Flask(__name__) def get_recent_vulns_from_db(limit50): # 假设你把结果存入了SQLite数据库 conn sqlite3.connect(cve_data.db) cursor conn.cursor() cursor.execute(SELECT * FROM cve_records ORDER BY last_modified_date DESC LIMIT ?, (limit,)) rows cursor.fetchall() conn.close() # 将行转换为字典列表... return rows app.route(/) def dashboard(): vulns get_recent_vulns_from_db(limit100) # 按严重等级统计 stats {CRITICAL: 0, HIGH: 0, MEDIUM: 0, LOW: 0} for v in vulns: stats[v.severity] 1 return render_template(dashboard.html, vulnerabilitiesvulns, statsstats) app.route(/api/vulns) def api_vulns(): vulns get_recent_vulns_from_db(limit200) return jsonify(vulns) if __name__ __main__: app.run(debugTrue)这样团队成员就可以通过浏览器查看最新的漏洞情报并且你可以集成搜索、按产品过滤、订阅提醒等功能让漏洞监控工作变得更加协同和高效。这个脚本的核心价值在于自动化了信息收集的苦力活让你能更专注于分析和响应这些漏洞带来的实际风险。