做安全工作这么多年真正让我觉得“数据驱动”这四个字有重量的时刻不是SOC里堆了多少贵的平台而是当你把全球公开的漏洞信息全部抓下来、按企业资产的维度整理好之后每天早晨那封预警邮件带来的确定性。这套基于Python爬虫搭建的企业级漏洞数据采集系统核心就是解决CVE数据的增量爬取、智能分类、威胁评估、实时预警和可视化分析问题数据源锁定在NVD和CVE Details两个公开渠道最后输出一份结构化的威胁情报报告让安全团队不用再靠人工去翻公告。如果你是安全工程师、运维负责人或者正在做安全数据产品的开发者这篇文章值得从头看完。它不会只丢给你一段能跑的代码而是把整个系统的设计思路、数据源差异、增量同步原理、分类与评分逻辑、预警接入方式以及我踩过的坑全部拆开讲。项目本身不需要特别高深的算法主要靠requests、lxml、APScheduler、pyecharts这类常规库就能落地但对工程细节的要求比较高——比如增量怎么判断、接口限流怎么绕开、CVE编号怎么去重这些才是决定系统能不能长期稳定跑下去的关键。1. 整体方案设计两个数据源怎么选、怎么搭1.1 核心需求拆解与模块划分标题里提到的功能点很多但拆开看其实就是一条流水线采集、清洗、入库、分析、预警、展示。最忌讳一上来就写爬虫脚本爬完发现字段对不上、重复数据一堆、调度又没着落。我建议先按模块拆采集层负责对接NVD API和CVE Details页面控制请求频率做最基本的HTML/JSON解析。存储层设计CVE主表、分类标签表、预警记录表增量爬取的元数据表也放在这里。分析层根据CWE分类、CVSS指标、引用链接做威胁评级产出结构化威胁情报。预警层按规则把新增的严重漏洞推送到邮件、钉钉或企业微信机器人。展示层用Flaskpyecharts搭一个轻量Dashboard输出漏洞趋势、供应商排行等可视化图表。这样拆的好处是每个模块都能独立调试。比如NVD接口挂了采集层单独重试不会影响分析层和展示层预警规则要改直接改分析层的输出逻辑就行。实际开发中我强烈建议用类来封装每一层不要写成几百行的脚本堆在一个文件里。1.2 NVD与CVE Details数据源对比选择数据源是第一个容易踩坑的地方。NVDNational Vulnerability Database提供官方JSON API数据字段完整、更新及时还带CVSS评分和CPE受影响产品信息是主数据源的不二之选。CVE Details则是一个聚合平台页面结构适合爬虫抓取它有按年、按月、按厂商的列表页历史数据回溯非常方便适合用来补全NVD中偶尔缺失的CWE编号或漏洞类型描述。对比维度NVD APICVE Details数据格式官方JSON结构化好需要解析HTML表格更新频率实时更新CVSS数据完整依赖聚合来源更新略慢历史数据支持分页时间窗口查询按年月归档翻页方便反爬策略接口有限流无封禁风险页面频率高会触发风控适用场景增量同步、全量初始化数据补全、按厂商统计真实项目中我的用法是初始化全量数据时两个源都跑一遍互相校验CVE条目数量日常增量同步以NVD为主一旦NVD某个时间窗口返回异常或者发现缺失记录再触发CVE Details的补偿抓取。这种双源融合策略能显著提高数据完整性但也引入了去重问题——同一个CVE编号在两个源的描述字段可能有细微差别处理方式放在后面数据库设计部分细说。2. 增量爬取不重不漏的关键实现2.1 增量策略的思路所谓增量爬取本质是回答一个问题上次跑到哪了这次从哪开始很多初学爬虫的人会把所有数据全量抓一遍然后用CVE编号去重。这在数据量小时可行但NVD目前公开的CVE条目已经超过二十万条每天还在新增全量拉取不仅慢还非常容易被接口限流。我采用的方案是“时间窗口本地游标”双保险。NVD API 2.0版本支持lastModStartDate和lastModEndDate参数可以按修改时间过滤CVE。每次调度任务启动时读取数据库中记录的last_sync_time作为本次窗口的起点终点是当前时间。然后调用API翻页拉取这个窗口内的所有CVE记录。因为NVD的lastModified字段会随任何元数据变更而更新所以这种方式天然能捕捉到“新增”和“修改”两类变化比单纯按公开日期判断更可靠。2.2 数据库表设计与CVE去重CVE主表是整系统的核心设计上需要平衡查询性能和灵活性。我实际使用的表结构大概是这样CREATE TABLE cve_records ( id INTEGER PRIMARY KEY AUTOINCREMENT, cve_id TEXT UNIQUE NOT NULL, -- CVE-2024-12345 description TEXT, -- 漏洞描述英文 published_date TEXT, last_modified_date TEXT, cvss_v31_score REAL, cvss_severity TEXT, -- LOW / MEDIUM / HIGH / CRITICAL cvss_vector TEXT, -- CVSS:3.1/AV:N/AC:L/... cwe_id TEXT, -- CWE-79 vuln_type TEXT, -- 智能分类后的漏洞类型 affected_vendor TEXT, affected_product TEXT, exploit_references TEXT, -- 是否含exploit-db等链接 threat_level INTEGER, -- 综合威胁评估分数 source TEXT, -- nvd / cvedetails / merged raw_hash TEXT, -- 原始内容哈希用于去重 created_at TEXT DEFAULT CURRENT_TIMESTAMP );去重逻辑有个容易忽略的细节不能只看cve_id因为同一CVE在NVD和CVE Details的描述可能被截断或改写。我的做法是先按cve_id做唯一索引插入时用INSERT OR IGNORE如果CVE已存在则对比raw_hash字段对descriptionlast_modified_date取MD5哈希不一致说明数据有更新执行UPDATE。这样既避免重复插入又不会漏掉字段修正。2.3 增量任务调度与失败续跑调度我用的APSchedulercron表达式每4小时执行一次增量任务。这里要提醒一个实操问题增量任务必须支持断点续跑。比如窗口内总共有500条数据API要求每次最多取2000条看起来一次能拉完但企业网络环境或者服务偶发超时会让任务中断。如果每次中断后都从头开始前面处理过的数据等于白跑。解决方案是维护一个crawl_meta表记录每个时间窗口的progress偏移量。伪代码如下def sync_nvd_window(window_start, window_end): offset get_meta(nvd_offset, 0) total None while True: params { lastModStartDate: window_start.isoformat() .000, lastModEndDate: window_end.isoformat() .000, startIndex: offset, resultsPerPage: 2000 } data nvd_api_request(params) if total is None: total data[totalResults] save_cve_records(data[vulnerabilities]) offset data[resultsPerPage] update_meta(nvd_offset, offset) if offset total: break这个方案实测下来很稳。哪怕中途断掉下次启动时offset还在继续拉剩余部分即可。等窗口跑完再把last_sync_time更新成新的window_end归档本次进度。3. 爬虫核心细节NVD API与页面解析实战3.1 NVD API的请求封装与限流规避NVD API免费Key的速率限制大约是每30秒最多几个请求不同时段可能有调整。没Key的匿名访问限制更严。所以第一件事去NVD官网申请一个免费API Key然后封装请求函数时做好退避重试。import requests import time from datetime import datetime, timedelta NVD_API https://services.nvd.nist.gov/rest/json/cves/2.0 API_KEY your-free-api-key def nvd_api_request(params, max_retries5): headers {apiKey: API_KEY, User-Agent: SecurityCrawler/1.0} for attempt in range(max_retries): try: resp requests.get(NVD_API, paramsparams, headersheaders, timeout30) if resp.status_code 200: return resp.json() elif resp.status_code 403: # 限流了指数退避等待 wait_time 30 * (attempt 1) print(f[NVD] Rate limited, wait {wait_time}s) time.sleep(wait_time) else: resp.raise_for_status() except requests.exceptions.Timeout: print(f[NVD] Timeout, retry {attempt1}/{max_retries}) time.sleep(10 * (attempt 1)) return None有一个坑NVD返回的JSON里vulnerabilities数组里每条记录的cve对象结构很复杂不能只拿顶层字段。实际解析时要深入到cve.metrics.cvssMetricV31[0].cvssData取baseScore和vectorString再从cve.descriptions里筛选lang en的description。这些字段层级如果没搞清楚解析代码会频繁抛KeyError。3.2 CVE Details页面解析xpath与text函数的实战用法CVE Details的列表页没有官方API只能爬HTML。以漏洞列表页为例每一行包含CVE编号、CWE编号、漏洞类型、公开日期、更新日期、CVSS评分、厂商、产品等列。用requests拉取页面后配合lxml的xpath解析这里有一个非常实用的技巧用xpath的text()函数配合normalize-space()清洗单元格内容。from lxml import html import requests def parse_cvedetails_page(year, month): url fhttps://www.cvedetails.com/vulnerability-list-{year}-{month}.html resp requests.get(url, headers{User-Agent: Mozilla/5.0}, timeout20) tree html.fromstring(resp.text) rows tree.xpath(//*[idpagetable]/tr) for row in rows[1:]: # 跳过表头 cve_id row.xpath(./td[1]/a/text()) cwe_id row.xpath(./td[2]/a/text()) vuln_type row.xpath(./td[3]/text()) date_pub row.xpath(./td[4]/text()) cvss row.xpath(./td[6]/text()) product row.xpath(./td[9]/a/text()) # 用text()取出来的内容经常带空白符必须清洗 clean lambda x: x[0].strip() if x and x[0].strip() else None yield { cve_id: clean(cve_id), cwe_id: clean(cwe_id), vuln_type: clean(vuln_type), published: clean(date_pub), cvss_score: clear_cvss(clean(cvss)), product: clean(product) }很多教程会忽略text()返回的是一个列表而且原始HTML里单元格内可能有换行符、多个空格直接取第一项会得到脏数据。我习惯写一个clean函数统一清洗同时因为CVSS评分一列里可能包含“7.5 5.0”这种多个值的情况需要单独处理取第一个数值即可缺失的置为None等待NVD数据补齐。CVE Details适合做NVD的补充而不是替代因为它的描述字段会截断而且某些CVE记录的CVSS版本可能混乱。3.3 双源融合与冲突解决策略双源融合是项目的加分项也是复杂度来源。我的策略是以NVD为准主表数据CVE Details只补充cwe_id和vuln_type。合并逻辑运行在数据入库后的清洗阶段避免爬虫代码和分析逻辑耦合。def merge_cvedetails_record(main_record, supplement_record): if not main_record.get(cwe_id) and supplement_record.get(cwe_id): main_record[cwe_id] supplement_record[cwe_id] if not main_record.get(vuln_type): main_record[vuln_type] supplement_record[vuln_type] main_record[source] merged if supplement_record.get(cwe_id) else nvd return main_record这里有个决策值得聊聊为什么不直接用CVE Details的漏洞类型字段做分类因为它的类型粒度太粗很多条目标的是“XSS”或者“Sql Injection”但也有大量“Others”“Unknown”的脏值。我后面做智能分类时主要依靠CWE编号映射表CVE Details的类型文本只作为兜底参考。4. 智能分类与威胁评估从原始数据到威胁等级4.1 基于CWE的漏洞类型映射CVE数据里的CWE编号比如CWE-89、CWE-79是对漏洞根因的标准分类。但CWE编号层多达上千个对安全运营来说粒度太细了。智能分类要做的是把CWE编号映射到业务视角的大类比如注入类、跨站脚本类、信息泄露类、权限提升类、拒绝服务类、资源管理类等。做法很直接维护一张映射字典再配合关键词兜底。CWE_MAPPING { CWE-89: SQL注入, CWE-78: 命令注入, CWE-79: 跨站脚本, CWE-22: 路径遍历, CWE-502: 反序列化, CWE-200: 信息泄露, CWE-269: 权限提升, CWE-287: 认证绕过, CWE-400: 拒绝服务, # ... 更多映射 } def classify_vulnerability(cwe_id, description): # 优先使用CWE映射 if cwe_id in CWE_MAPPING: return CWE_MAPPING[cwe_id] # 兜底从描述中匹配关键词 desc_lower (description or ).lower() for keyword, category in KEYWORD_CATEGORY_MAP.items(): if keyword in desc_lower: return category return 未分类KEYWORD_CATEGORY_MAP里我可以放类似sql映射到“SQL注入”、cross-site映射到“跨站脚本”这样的关键词。这套方案不需要训练模型纯规则就能覆盖70%以上的CVE剩下归类为“未分类”的可以每周人工复核一次把新出现的漏洞模式补充进映射表。4.2 CVSS评分指标拆解与威胁等级计算智能分类解决“是什么”威胁评估解决“有多危险”。CVSS v3.1评分本身已经给出基础等级0.1-3.9为Low4.0-6.9为Medium7.0-8.9为High9.0-10.0为Critical。但这只是基础分不足以反映企业视角下的真实威胁程度。我在基础分之上叠加了三个维度可利用性是否存在公开PoC或Exploit-DB链接、时效性最近7天内新发布或更新的漏洞加权、影响范围CVSS向量中AV网络可达且PR较低说明利用门槛低。综合评分按如下规则计算维度权重判断条件加成CVSS基础分0.6baseScore直接参与保留原始分可利用性0.2references包含exploit加分2.0时效性0.1发布/修改时间距今天数 7加分1.0利用门槛0.1AV:N与PR:N且UI:N加分1.0实际实现时我用一个函数把CVSS向量里的AV、PR、UI等字段解析出来组合成“威胁附加分”。最终威胁等级大于等于8.5且CVSS基础分大于等于7.0的进入严重预警队列威胁附加分大于等于2的即使基础分只有6.0也会触发中危提醒。这套规则的目的是避免纯粹依赖CVSS导致漏掉那些“分数不高但已经有公开利用工具”的漏洞。4.3 结构化威胁情报报告生成报告输出是整个系统的最终落脚点。我的做法不是生成一篇写死的HTML而是按标准JSON schema输出方便对接企业内部的工单系统或SIEM。报告里包含本次采集周期内的新增漏洞总数、按严重等级分布、受影响供应商TOP10、高危漏洞清单每个漏洞的CVE编号、描述、CVSS向量、参考链接、修复建议、以及分类图谱。{ report_id: TI-20250110-001, window: {start: 2025-01-09T00:00:00, end: 2025-01-10T00:00:00}, cve_count_new: 98, severity_stats: {critical: 12, high: 31, medium: 38, low: 17}, vendor_top10: [Microsoft, Google, Apple, ...], critical_list: [ { cve_id: CVE-2025-12345, description: ..., base_score: 9.8, threat_extra_score: 2.8, refs: [https://..., https://www.exploit-db.com/...] } ] }生成JSON之后再写一个模板渲染函数把JSON转成Markdown或PDF。实际运营中运维同事更喜欢看Markdown因为它可以直接粘贴到知识库或聊天工具里。5. 实时预警与可视化让数据跑起来5.1 预警通知渠道接入预警是“采集系统”里最能体现价值的一环。没有预警数据入库再多也是死的。我的预警模块设计成多通道分发邮件、钉钉群机器人、企业微信群机器人。核心是定义统一的告警事件结构然后各渠道根据结构渲染消息。钉钉机器人的接入非常简单只需要在钉钉群添加自定义机器人拿到webhook地址然后POST一段JSONdef send_dingtalk(message_dict): webhook https://oapi.dingtalk.com/robot/send?access_tokenxxxx payload { msgtype: markdown, markdown: { title: 新增高危漏洞预警, text: f### 新增高危漏洞 {message_dict[cve_id]}\n\n f- CVSS评分{message_dict[base_score]}\n f- 威胁等级{message_dict[threat_level]}\n f- 描述{message_dict[description][:100]}\n\n f[查看详情]({message_dict[ref_url]}) } } requests.post(webhook, jsonpayload, timeout10)企业微信机器人也是类似套路只是字段名换成{msgtype: markdown}格式稍微不同。邮件则用smtplib发送HTML正文。有个细节容易被忽略预警必须做摘要和合并。如果窗口内新增100个高危漏洞一条条发会瞬间刷屏最终被管理员设置免打扰。我设置了合并规则窗口内按严重等级聚合每30分钟最多推一条包含TOP5高危漏洞和完整清单的CSV附件。5.2 可视化Dashboard搭建可视化我采用的是Flask pyecharts的组合因为pyecharts可以直接在Python里生成ECharts的前端代码不用单独写JS。Dashboard按“概览-趋势-分布”三层组织概览卡片今日新增漏洞数、严重漏洞数、待人工复核数。趋势折线图最近30天每日新增CVE数量以及严重漏洞趋势线。分布图漏洞类型饼图、受影响供应商TOP15柱状图、CVSS评分分布散点图。from pyecharts.charts import Line, Pie, Bar from pyecharts import options as opts def build_trend_chart(days, counts): line ( Line() .add_xaxis(days) .add_yaxis(新增CVE数, counts, is_smoothTrue, markline_optsopts.MarkLineOpts(data[opts.MarkLineItem(type_average)])) .set_global_opts(title_optsopts.TitleOpts(title近30天漏洞趋势)) ) return line.render_embed() # 生成HTML片段这里有一个坑Flask渲染pyecharts时用render_embed()可以避免磁盘上的临时文件但每次请求都要重新计算图表数据量大时响应会慢。我的优化方案是在数据入库后直接提前生成静态HTML片段缓存到内存或RedisDashboard接口只负责读缓存。这样打开页面时基本无感。5.3 预警与可视化的衔接预警和可视化不是独立的很多团队把两者割裂了导致预警过来还要手动去系统里查详情。我的做法是预警消息里带上一个/cve/{cve_id}的链接点击进去是单漏洞详情页包含完整描述、CVSS向量解析、参考链接、关联的CWE分类。这样运营人员从收到告警到完成判定的路径最短一篇文章里能讲清楚整个链路也是系统展示层真正“企业级”的体现。6. 踩坑实录与常用排查技巧6.1 高频问题速查表这套系统上线半年我维护了一份问题排查清单在这里完整分享现象可能原因排查方法NVD API返回403限流或API Key失效检查请求频率增加退避时间确认Key仍有配额爬CVE Details偶尔空列表页面结构改版或反爬验证对比浏览器实际HTML更新xpath路径加Cookie头数据库cve_id冲突主键重复或大小写不一致统一转换为大写再去重使用INSERT OR IGNORE增量任务重复跑时间窗口未持久化检查crawl_meta中的last_sync_time是否提交事务任务中途崩掉导致数据不全没有断点续跑机制维护offset游标启动时恢复进度邮件预警发不出去SMTP服务25端口被运营商封禁换465端口SSL加密或使用企业邮箱APIDashboard图表不刷新缓存未失效入库任务完成后主动清除缓存键6.2 我列得出来的几个重要经验第一个经验不要把爬虫和业务逻辑写在一起。爬虫负责拿数据清洗入库是另一个模块这样NVD接口变动时影响的只是采集层不会拖垮整个系统。我吃过亏初版代码把解析逻辑直接写在requests请求后面结果CVE Details一次页面结构调整整个任务脚本崩了三天。第二个经验请求频率一定要做全局统一控制。不要只在NVD封装里控制limitCVE Details页面也要限速。我的方案是用一个TokenBucket类统一控制所有出站请求的速率默认每分钟60个请求NVD和CVE Details共用这个令牌桶。否则NVD限流刚解决CVE Details的反爬检测又触发。第三个经验测试环境必须和生产环境隔离。我专门搭了一套SQLite版本用于日常联调生产则用PostgreSQL或MySQL。因为SQLite在并发写入上限制太多一旦预警模块和采集模块同时操作数据库会出现数据库被锁的报错。虽然SQLite支持WAL模式能缓解但当数据量到几十万条后性能确实扛不住。第四个经验日志要留够上下文。很多爬虫脚本出问题后一脸懵就是因为日志只打了“请求失败”没记录请求参数和响应状态。我在采集层每个关键节点都记录结构化日志至少包含目标URL、请求参数、HTTP状态码、耗时、异常类型。出问题时第一件事是翻日志而不是重新跑任务。6.3 一个小技巧用测试CVE编号做全链路验证系统上线前可以用几个已知的CVE编号做端到端验证。比如用CVE-2023-1234、CVE-2024-5678这些特定编号在增量窗口内故意包含它们然后确认整条链路采集→入库→分类→预警→Dashboard展示。如果某个环节没跑通马上就能定位。这个方法帮我节省了无数调试时间强烈建议你在开发阶段就建立一套验证样例。整体做完这个项目我个人最大的感受是安全数据系统真正难的地方从来不是爬虫本身而是如何让数据在你需要的时候以正确的方式被使用。增量爬取只是敲门砖后面的清洗规则、分类映射、评分策略每一个决策都会直接影响漏洞响应的效率和准确度。如果你也打算搭建类似的系统我建议先用SQLite和单机脚本把整条链路跑通确认数据流没有任何问题之后再逐步引入消息队列、分布式采集这些重型组件。在数据量还没起来之前过度设计才是团队最大的敌人。
