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信息”的漏洞,主要会用到两个核心接口:

  1. 漏洞摘要接口 ( /rest/json/cves/2.0 ) : 这是获取漏洞列表的主要入口。它支持通过多种参数进行筛选,例如发布时间范围、漏洞严重等级(CVSS分数)、关键词等。我们将主要利用它的 lastModStartDate lastModEndDate 参数来获取在特定时间段内修改(包括新发布)的漏洞。
  2. 单个漏洞详情接口 ( /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 整体架构设计

脚本的核心工作流可以概括为以下几步:

  1. 确定时间窗口 : 计算需要查询的时间范围,通常是“从昨天凌晨到现在”或者最近24小时。
  2. 调用摘要API : 向NVD API发起请求,获取该时间段内所有更新的漏洞摘要列表。
  3. 初步过滤 : 遍历列表,筛选出 configurations 字段不为空的条目(即含有CPE信息)。
  4. 数据解析与丰富 : 对筛选出的漏洞,解析其CPE信息,并可以根据需要,选择性地调用详情API获取更完整的描述、参考链接等。
  5. 结果输出与持久化 : 将结果以结构化的格式(如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)

  1. Q: 脚本运行后返回“未找到任何漏洞更新”,但我确定应该有新漏洞。

    • A: 首先,检查你使用的 lastModStartDate lastModEndDate 时间格式是否正确(必须是UTC的ISO格式)。其次,确认时间窗口是否合理。NVD数据处理有延迟,刚披露的漏洞可能几分钟到几小时后才会出现在API中。可以尝试将 hours_back 参数调大一些(比如48小时)。
  2. Q: 获取到的CPE信息中, vulnerable 字段全是 False ,导致过滤后没有数据。

    • A: 这可能是正常的。有些漏洞条目在初始收录时,可能只包含了“不受影响”的CPE配置。或者,你查询的漏洞其影响范围是通过更复杂的逻辑定义的,而不是简单的 cpeMatch 列表。可以尝试暂时取消 vulnerable=True 的过滤条件,查看原始数据结构和 operator 字段。
  3. Q: 请求总是失败,返回403或429错误。

    • A: 403错误可能是请求头或参数格式问题。429错误肯定是触发了速率限制。
      • 确认你是否使用了API密钥,并且密钥有效。
      • 检查你的请求频率。即使有API密钥,也要遵守每分钟/每小时的限制。在循环请求中添加 time.sleep()
      • 检查 User-Agent 是否被设置,有些公开API会拒绝空User-Agent的请求。
  4. Q: 解析CPE字符串时出错,提示索引超出范围。

    • A: 并非所有CPE URI都严格遵循 cpe:2.3:a:vendor:product:version:... 的格式。有些可能版本字段为空( : ),或者包含其他变体。在 parse_cpe_uri 函数中增加更严格的长度检查和异常捕获( try...except ),对解析失败的CPE记录日志而不是直接崩溃。
  5. Q: 输出的CSV文件在Excel中打开中文乱码。

    • A: 确保使用 utf-8-sig 编码保存CSV文件。 pandas to_csv 方法使用 encoding='utf-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(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)

这样,团队成员就可以通过浏览器查看最新的漏洞情报,并且你可以集成搜索、按产品过滤、订阅提醒等功能,让漏洞监控工作变得更加协同和高效。这个脚本的核心价值在于自动化了信息收集的苦力活,让你能更专注于分析和响应这些漏洞带来的实际风险。

更多推荐