Python自动化NVD漏洞监控:CPE解析与API实战指南
1. 项目概述:为什么我们需要一个自动化的NVD漏洞监控器?
在安全运维和漏洞管理的日常工作中,我经常需要关注特定软件或组件的最新安全漏洞。手动去NVD(National Vulnerability Database)官网翻找,不仅效率低下,还容易遗漏。特别是当我们需要精准定位到带有CPE(Common 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/activate
3.2 安装必要的库
我们主要需要以下几个库:
-
requests: 用于发送HTTP请求到NVD API,这是最核心的库。 -
pandas: 非必须,但强烈推荐。用于将获取到的漏洞数据整理成表格,方便筛选、分析和导出为Excel/CSV。 -
python-dotenv: 非必须,但推荐。用于管理API密钥等敏感配置,避免将其硬编码在脚本中。
使用pip一键安装:
pip install requests pandas python-dotenv
3.3 配置文件管理
创建一个
.env
文件来存储你的NVD API密钥(如果你有的话)和其他配置。永远不要将密钥提交到代码仓库。
# .env 文件内容示例
NVD_API_KEY=your_actual_api_key_here
TIME_WINDOW_HOURS=24
RESULTS_PER_PAGE=2000
然后在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_index=0):
"""
构建查询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}"
f"&lastModEndDate={end_date}"
f"&startIndex={start_index}")
# 如果有API密钥,通常作为请求头传递,但NVD API也支持作为参数 `apiKey`
# 更常见的做法是放在请求头 `apiKey` 中,但NVD文档建议用参数。这里按参数处理。
if API_KEY:
url += f"&apiKey={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, headers=headers, timeout=30)
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(f"HTTP错误发生: {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_part=None, filter_vendor=None, filter_product=None):
"""
过滤出带有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)
# 通常我们只关心标记为易受攻击(vulnerable=True)的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则表示是不受影响的版本(例如修复后的版本)。我们的脚本通常只关心vulnerable=True的条目。 -
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_format='json'):
"""
将过滤后的漏洞数据保存到文件。
参数:
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 = f"cve_with_cpe_{timestamp}.json"
with open(filename, 'w', encoding='utf-8') as f:
# 处理无法序列化的对象(如果需要的话,这里parsed_cpes是dict列表,可序列化)
json.dump(filtered_cves, f, indent=2, ensure_ascii=False)
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序号'] = i+1
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 = f"cve_with_cpe_{timestamp}.csv"
df.to_csv(filename, index=False, encoding='utf-8-sig') # utf-8-sig支持Excel中文
print(f"结果已保存为 CSV 文件: {filename}")
else: # excel
filename = f"cve_with_cpe_{timestamp}.xlsx"
df.to_excel(filename, index=False)
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_back=24, filter_product=None):
"""
主函数:执行完整的漏洞抓取、过滤和保存流程。
参数:
hours_back (int): 查询多少小时之前到现在的数据。
filter_product (str): 可选,过滤产品名称,如 'tomcat'
"""
print("=== NVD CVE 监控脚本开始运行 ===")
# 1. 计算时间范围
end_time = datetime.utcnow()
start_time = end_time - timedelta(hours=hours_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_product=filter_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_back=24, filter_product='tomcat')
实现自动化
:
在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 2>&1
5. 高级技巧与避坑指南
5.1 处理复杂的CPE匹配逻辑
NVD返回的CPE信息有时不是简单的单个字符串,而是嵌套的、带有逻辑运算符(AND, OR)的复杂节点树。我们的
filter_and_parse_cves
函数中的遍历逻辑(遍历
nodes
和
cpeMatch
)已经能处理大部分情况。但对于更复杂的
CVE-2021-44228
(Log4Shell) 这样的漏洞,其影响范围可能通过复杂的逻辑组合来定义。一个健壮的解析器可能需要递归遍历整个节点树,并解析
operator
字段。对于大多数监控场景,我们目前的扁平化提取
vulnerable=True
的CPE已经足够,但了解这一点对于深度分析很重要。
5.2 优化性能与应对API限制
- 增量查询 : 不要每次都查询过去24小时的全量数据。可以将上次成功运行的时间戳记录在一个文件中,下次查询时,只查询从那个时间戳到现在的新数据。这能大幅减少请求数据量,也符合API调用的最佳实践。
- 缓存机制 : 对于需要频繁查询的CVE详情(比如在生成报告时需要更完整的描述),可以考虑将结果缓存在本地数据库(如SQLite)中,避免重复请求。
- 错误重试与退避 : 我们之前实现了对429错误的简单重试。在生产环境中,应该实现更完善的退避算法(如指数退避),并对连接超时、服务器错误(5xx)等进行重试。
5.3 数据丰富与关联
单纯的CPE信息有时不够。你可以考虑将获取到的CVE数据与其它源关联:
- 关联Exploit DB : 检查是否有公开的漏洞利用代码(Exploit)。
- 关联安全公告 : 获取厂商的官方安全公告链接。
-
内部资产关联
: 这是最关键的一步。将解析出的
vendor和product与你内部的CMDB(配置管理数据库)或资产清单进行匹配,就能立即知道这个漏洞影响了你哪些具体的服务器或应用,实现从“漏洞预警”到“风险定位”的跨越。
5.4 常见问题排查(FAQ)
-
Q: 脚本运行后返回“未找到任何漏洞更新”,但我确定应该有新漏洞。
-
A:
首先,检查你使用的
lastModStartDate和lastModEndDate时间格式是否正确(必须是UTC的ISO格式)。其次,确认时间窗口是否合理。NVD数据处理有延迟,刚披露的漏洞可能几分钟到几小时后才会出现在API中。可以尝试将hours_back参数调大一些(比如48小时)。
-
A:
首先,检查你使用的
-
Q: 获取到的CPE信息中,
vulnerable字段全是False,导致过滤后没有数据。-
A:
这可能是正常的。有些漏洞条目在初始收录时,可能只包含了“不受影响”的CPE配置。或者,你查询的漏洞其影响范围是通过更复杂的逻辑定义的,而不是简单的
cpeMatch列表。可以尝试暂时取消vulnerable=True的过滤条件,查看原始数据结构和operator字段。
-
A:
这可能是正常的。有些漏洞条目在初始收录时,可能只包含了“不受影响”的CPE配置。或者,你查询的漏洞其影响范围是通过更复杂的逻辑定义的,而不是简单的
-
Q: 请求总是失败,返回403或429错误。
-
A:
403错误可能是请求头或参数格式问题。429错误肯定是触发了速率限制。
- 确认你是否使用了API密钥,并且密钥有效。
-
检查你的请求频率。即使有API密钥,也要遵守每分钟/每小时的限制。在循环请求中添加
time.sleep()。 -
检查
User-Agent是否被设置,有些公开API会拒绝空User-Agent的请求。
-
A:
403错误可能是请求头或参数格式问题。429错误肯定是触发了速率限制。
-
Q: 解析CPE字符串时出错,提示索引超出范围。
-
A:
并非所有CPE URI都严格遵循
cpe:2.3:a:vendor:product:version:...的格式。有些可能版本字段为空(:),或者包含其他变体。在parse_cpe_uri函数中增加更严格的长度检查和异常捕获(try...except),对解析失败的CPE记录日志而不是直接崩溃。
-
A:
并非所有CPE URI都严格遵循
-
Q: 输出的CSV文件在Excel中打开中文乱码。
-
A:
确保使用
utf-8-sig编码保存CSV文件。pandas的to_csv方法使用encoding='utf-8-sig'参数即可解决。
-
A:
确保使用
6. 从脚本到工具:构建一个简单的Web仪表板
当脚本稳定运行后,你可以考虑将其升级为一个内部工具。使用轻量级的Web框架如
Flask
或
FastAPI
,可以快速搭建一个仪表板。
# 这是一个非常简化的 Flask 应用示例
from flask import Flask, render_template, jsonify
import sqlite3
app = Flask(__name__)
def get_recent_vulns_from_db(limit=50):
# 假设你把结果存入了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(limit=100)
# 按严重等级统计
stats = {'CRITICAL': 0, 'HIGH': 0, 'MEDIUM': 0, 'LOW': 0}
for v in vulns:
stats[v.severity] += 1
return render_template('dashboard.html', vulnerabilities=vulns, stats=stats)
@app.route('/api/vulns')
def api_vulns():
vulns = get_recent_vulns_from_db(limit=200)
return jsonify(vulns)
if __name__ == '__main__':
app.run(debug=True)
这样,团队成员就可以通过浏览器查看最新的漏洞情报,并且你可以集成搜索、按产品过滤、订阅提醒等功能,让漏洞监控工作变得更加协同和高效。这个脚本的核心价值在于自动化了信息收集的苦力活,让你能更专注于分析和响应这些漏洞带来的实际风险。
更多推荐
所有评论(0)