1. 项目概述:爬取NVD漏洞数据的实用价值
美国国家漏洞数据库(NVD)作为全球最权威的漏洞信息库之一,收录了超过20万条CVE漏洞记录。对于安全研究人员、企业IT管理者和开发者而言,定期获取这些数据具有多重价值:
- 漏洞情报监控:实时掌握新披露漏洞的严重程度和影响范围
- 风险评估依据:通过CVSS评分快速判断漏洞修复优先级
- 安全工具开发:为漏洞扫描器、SIEM系统提供基础数据支持
- 学术研究素材:分析漏洞趋势、攻击模式演变的量化依据
传统手动查询方式效率低下,而NVD提供的API存在调用频率限制(官方建议每分钟不超过5次请求)。通过Python爬虫实现自动化采集,可以灵活获取结构化数据,特别适合需要定制化字段或历史数据归档的场景。
2. 技术方案设计思路
2.1 网页爬取 vs API调用
NVD同时提供网页端和API两种数据获取方式,我们选择网页爬取方案主要基于以下考量:
- 字段完整性:网页包含漏洞描述、解决方案等API未提供的细节信息
- 历史数据:可获取早年间未通过API标准化的老旧漏洞记录
- 灵活性:不受API版本变更和字段限制影响
- 成本控制:避免企业级API的调用配额管理
注意:虽然网页爬取更灵活,但必须严格遵守NVD的robots.txt规定。实测发现NVD允许爬虫访问/cve目录,但要求请求间隔至少6秒。
2.2 目标数据结构设计
我们需要提取的字段包括:
{ "cve_id": "CVE-2023-1234", "severity": "CRITICAL", "cvss_score": 9.8, "published_date": "2023-01-15", "last_modified": "2023-03-22", "description": "在XX组件中发现缓冲区溢出漏洞...", "affected_products": ["Windows 10", "Windows Server 2019"] }2.3 技术栈选型
- 请求库:Requests + Retrying(比urllib3更友好的重试机制)
- 解析工具:BeautifulSoup4 + lxml(比正则表达式更稳定的选择)
- 数据存储:csv/json双格式输出(兼顾可读性和程序可处理性)
- 反反爬:随机User-Agent + 动态延迟(模拟人类操作模式)
3. 核心实现细节解析
3.1 请求层实现要点
from retrying import retry import random import time class NVDCrawler: def __init__(self): self.base_url = "https://nvd.nist.gov/vuln/search/results" self.delay = 6 + random.uniform(0, 3) # 基础6秒+随机扰动 @retry(stop_max_attempt_number=3, wait_exponential_multiplier=1000) def fetch_page(self, page_num): params = { "form_type": "Basic", "results_type": "overview", "search_type": "all", "startIndex": (page_num - 1) * 20 } headers = { "User-Agent": self._get_random_ua() } time.sleep(self.delay) response = requests.get(self.base_url, params=params, headers=headers) response.raise_for_status() return response.text def _get_random_ua(self): ua_list = [ "Mozilla/5.0 (Windows NT 10.0; Win64; x64)...", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)..." ] return random.choice(ua_list)关键设计说明:
- 指数退避重试:网络异常时自动重试,每次间隔时间倍增
- 动态延迟:在6秒基础上增加0-3秒随机扰动,避免固定周期被识别
- 分页参数:startIndex每页增加20(NVD的默认分页大小)
3.2 解析层关键技术
NVD页面采用动态生成的HTML结构,需要通过实际DOM分析确定选择器:
from bs4 import BeautifulSoup def parse_page(html): soup = BeautifulSoup(html, 'lxml') items = [] for row in soup.select('table[data-testid="vuln-results-table"] tr'): cve_id = row.select_one('th a').text.strip() severity = row.select_one('span[data-testid="vuln-cvss3-panel-score"]').text cvss_score = float(severity.split()[0]) date_cols = row.select('td')[2].find_all('span') pub_date = date_cols[0].text mod_date = date_cols[1].text if len(date_cols) > 1 else pub_date desc = row.select_one('p[data-testid="vuln-summary-"]').text items.append({ "cve_id": cve_id, "severity": severity, "cvss_score": cvss_score, "published_date": pub_date, "last_modified": mod_date, "description": desc }) return items解析难点处理:
- 动态属性:使用data-testid而非class作为选择器锚点
- 空值处理:未修改的漏洞last_modified等于published_date
- 文本清洗:CVSS评分需要从"9.8 CRITICAL"中分离数值
4. 数据存储与导出方案
4.1 多格式输出实现
import csv import json from datetime import datetime class DataExporter: @staticmethod def to_csv(data, filename): with open(f"{filename}.csv", 'w', newline='', encoding='utf-8') as f: writer = csv.DictWriter(f, fieldnames=data[0].keys()) writer.writeheader() writer.writerows(data) @staticmethod def to_json(data, filename): with open(f"{filename}.json", 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) @staticmethod def to_excel(data, filename): # 使用pandas简化Excel导出 import pandas as pd df = pd.DataFrame(data) df.to_excel(f"{filename}.xlsx", index=False)4.2 字段类型转换处理
原始数据需要规范化:
def normalize_data(raw_items): for item in raw_items: # 日期标准化 item['published_date'] = datetime.strptime( item['published_date'], '%m/%d/%Y').strftime('%Y-%m-%d') # 严重等级枚举化 item['severity'] = item['severity'].split()[-1].upper() # 描述文本清洗 item['description'] = ' '.join( item['description'].split()).strip() return raw_items5. 实战问题排查指南
5.1 常见问题解决方案
| 问题现象 | 可能原因 | 解决方案 |
|---|---|---|
| 返回403状态码 | 请求头特征明显 | 1. 更换User-Agent 2. 添加Referer头 |
| 解析到空列表 | HTML结构变更 | 1. 检查最新DOM结构 2. 改用更宽松的选择器 |
| 日期解析失败 | 格式区域差异 | 1. 强制指定locale 2. 使用dateutil解析器 |
| 内存占用过高 | 未分页处理 | 1. 分批处理数据 2. 使用生成器替代列表 |
5.2 性能优化建议
- 增量采集:
# 记录最后采集的CVE_ID last_cve = get_last_record() if item['cve_id'] == last_cve: break- 异步请求(需控制并发数):
import aiohttp async def fetch_all(pages): async with aiohttp.ClientSession() as session: tasks = [fetch_page(session, p) for p in range(1, pages+1)] return await asyncio.gather(*tasks)- 分布式扩展:
- 使用Redis存储已采集ID集合
- 通过Celery分发采集任务
6. 合规使用注意事项
- 访问频率控制:
- 单IP请求间隔≥6秒
- 每日采集量建议≤1000页
- 避免商业时段高峰期操作
- 数据使用限制:
- 禁止重新发布原始数据
- 衍生分析需注明数据来源
- 不得用于漏洞利用工具开发
- 技术伦理建议:
- 添加明显的用户代理标识
- 提供可联系的爬虫负责人邮箱
- 遵循最少必要原则采集数据
在实际运行中,建议先小规模测试(如采集前10页),确认无误后再进行全量采集。对于企业级应用场景,更推荐申请官方API密钥获取数据。