简介这是一份面向网络安全初学者与Python爬虫爱好者的实战项目包围绕CNVD、CNNVD、CVE三大权威漏洞库的数据采集展开解决漏洞信息分散、难以本地统一管理的问题。项目将爬虫抓取、MySQL持久化存储与GUI可视化预览串联成完整链路适合作为课程设计、毕业设计或安全分析入门的练手素材。压缩包共72个文件约137.95MB以py爬虫与GUI脚本、toc与spec打包配置、csv与xlsx漏洞数据表、exe可执行程序及pyc缓存为主另含xml、txt、html等辅助文件结构上区分了爬取、构建与发布目录。目前已有1584人学习下载。读者可从中获得requests与BeautifulSoup解析思路、mysql-connector-python入库写法、增量与完整爬取策略以及PyQt/Tkinter界面搭建和chromedriver调用等可复用经验并理解延时、代理等反爬处理与合规边界。1. 从「python爬取漏洞库.zip」说起一份漏洞情报到底该怎么攒很多做安全运营或者漏洞管理的同行第一次看到「python爬取漏洞库.zip」这类标题脑子里冒出来的问题其实很朴素我手上有一堆资产每天要盯的漏洞公告散在好几个站点人工刷根本刷不过来能不能用 Python 把公开漏洞库的数据定时抓下来落到本地做检索、比对和告警这个方向是成立的而且落地成本比想象中低——一个稳定的抓取脚本加上一张结构合理的表就能把「今天有哪些新漏洞和我有关」这件事从人工变成自动。但这里有个反直觉的结论爬漏洞库最难的不是写 requests而是数据源选型、字段归一化和增量去重。漏洞编号有 CVE、CNVD、CNNVD 等多套体系同一漏洞在不同库里的描述、评分、影响版本写法都不一样抓下来不归一化后面根本没法用。这篇文章面向的是想自己搭一套漏洞情报管道的工程师从数据源怎么挑、脚本怎么写、字段怎么对齐一路讲到定时任务和踩坑新手能照着跑通熟手能直接拿去改。2. 数据源选型与字段设计先想清楚要抓什么2.1 三类漏洞数据源的取舍公开漏洞数据源大致分三类选型直接决定你后面脚本的复杂度。第一类是结构化 API 型典型代表是 NVD 提供的 REST 接口。它返回 JSON字段规范有分页参数适合做主力数据源。缺点是更新有延迟且对请求频率有要求需要申请 API Key 才能提高配额。第二类是网页列表型很多国家级漏洞库以 HTML 列表加详情页的形式发布。这类源信息全但需要解析 HTML页面结构一变脚本就翻车维护成本高。常见做法是用它做补充而不是主力。第三类是聚合镜像型一些开源项目会把多个源的数据合并后提供 JSON 或 SQLite 文件。好处是省事坏处是字段口径由别人定你没法控制更新节奏。我一般会这样组合NVD API 做主力网页源做补充聚合镜像做冷启动的历史数据回填。这样既有实时性又有历史深度。数据源类型代表形式优点缺点建议角色结构化 APIREST/JSON字段规范、易分页有频率限制主力网页列表HTML 页面信息全、更新快结构易变补充聚合镜像JSON/SQLite省事、含历史口径不可控冷启动2.2 字段归一化把不同库的字段对齐到一张表不管你抓几个源最后都要落到同一张表里。我建议的最小字段集是这样的cve_id统一编号没有 CVE 的用源内编号加前缀source来源标识方便回溯published发布时间统一成 ISO8601cvss_score评分统一到 0-10 浮点severity等级统一成 CRITICAL/HIGH/MEDIUM/LOWdescription描述文本affected受影响产品与版本尽量结构化references参考链接数组raw原始 JSON 备份出问题时的后悔药这里的关键是severity的映射。不同源用的等级词不一样有的用「严重/高危/中危/低危」有的用英文有的干脆只给分数。我的做法是以分数为准反推等级映射规则写死在一个字典里避免每个源各写一套判断逻辑。# severity 映射以 CVSS 分数为准统一等级口径 SEVERITY_MAP [ (9.0, CRITICAL), (7.0, HIGH), (4.0, MEDIUM), (0.1, LOW), (0.0, NONE), ] def score_to_severity(score): 把 0-10 的分数映射成统一等级None 归为 UNKNOWN if score is None: return UNKNOWN for threshold, label in SEVERITY_MAP: if score threshold: return label return UNKNOWN这段逻辑说明SEVERITY_MAP按分数从高到低排列遍历时第一个满足score threshold的就是结果所以顺序不能乱。参数上阈值取 9.0/7.0/4.0/0.1 是业界通用口径和 CVSS v3 的等级划分一致。如果某个源只给了等级没给分数就反过来查表补一个中间值保证字段完整。2.3 数据库表结构一张主表加一张变更表落库我推荐 SQLite 起步单文件、零运维几万条数据毫无压力。表结构分两张vulns存最新状态vuln_changes存每次抓取发现的字段变化。-- 主表每个漏洞一行存最新状态 CREATE TABLE IF NOT EXISTS vulns ( cve_id TEXT PRIMARY KEY, source TEXT NOT NULL, published TEXT, cvss_score REAL, severity TEXT, description TEXT, affected TEXT, refs TEXT, -- JSON 数组字符串 raw TEXT, -- 原始 JSON 备份 updated_at TEXT ); -- 变更表记录字段级变化用于增量告警 CREATE TABLE IF NOT EXISTS vuln_changes ( id INTEGER PRIMARY KEY AUTOINCREMENT, cve_id TEXT, field TEXT, old_value TEXT, new_value TEXT, changed_at TEXT ); CREATE INDEX IF NOT EXISTS idx_vulns_published ON vulns(published); CREATE INDEX IF NOT EXISTS idx_vulns_severity ON vulns(severity);逻辑说明cve_id做主键天然去重INSERT OR REPLACE就能实现「有则更新无则插入」。vuln_changes单独一张表是为了做告警——比如某个漏洞的评分从 5.0 涨到 9.8这种变化比新漏洞更值得关注。索引建在published和severity上因为日常查询基本都按时间和等级过滤。参数上refs和raw用 TEXT 存 JSON 字符串SQLite 的 JSON 函数可以直接查不用额外拆表。3. 抓取脚本落地从单次请求到增量入库3.1 带重试和限速的请求封装直接裸写requests.get是新手最容易翻车的地方——网络抖动、限流、超时全都没处理跑一晚上可能一半数据是空的。我一般会封装一个带重试和限速的会话对象。import time import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def build_session(api_keyNone): 构造带重试策略的会话统一超时和请求头 session requests.Session() retry Retry( total3, # 最多重试 3 次 backoff_factor1.5, # 退避系数间隔 1.5s、3s、6s status_forcelist[429, 500, 502, 503, 504], allowed_methods[GET], ) adapter HTTPAdapter(max_retriesretry) session.mount(https://, adapter) session.headers.update({ User-Agent: vuln-collector/1.0, Accept: application/json, }) if api_key: session.headers[apiKey] api_key return session def fetch_json(session, url, paramsNone, min_interval6.0): 请求 JSON 接口强制最小请求间隔避免触发限流 time.sleep(min_interval) resp session.get(url, paramsparams, timeout30) resp.raise_for_status() return resp.json()逻辑说明Retry的backoff_factor决定退避节奏1.5 意味着第一次重试等 1.5 秒第二次 3 秒第三次 6 秒避免短时间内反复冲击对方接口。status_forcelist里放的是值得重试的状态码429 是限流5xx 是服务端问题4xx 里的 404 不该重试所以没放。min_interval是主动限速NVD 无 Key 时建议 6 秒一次有 Key 可以降到 0.6 秒。参数上timeout30是经验值漏洞接口偶尔响应慢设太短会误判失败。3.2 分页拉取与时间窗口控制漏洞库动辄十几万条不可能每次全量拉。正确做法是按发布时间窗口增量拉取每次只取上次成功时间之后的数据。from datetime import datetime, timedelta, timezone def fetch_nvd_window(session, start, end, api_keyNone): 按时间窗口分页拉取返回原始条目列表 base https://services.nvd.nist.gov/rest/json/cves/2.0 page_size 2000 offset 0 results [] while True: params { pubStartDate: start.strftime(%Y-%m-%dT%H:%M:%S.000), pubEndDate: end.strftime(%Y-%m-%dT%H:%M:%S.000), resultsPerPage: page_size, startIndex: offset, } data fetch_json(session, base, params, min_interval0.6 if api_key else 6.0) items data.get(vulnerabilities, []) results.extend(items) total data.get(totalResults, 0) offset page_size if offset total or not items: break return results逻辑说明时间窗口用pubStartDate和pubEndDate卡住格式必须是 ISO8601 带毫秒这是接口的硬要求格式错了直接返回 404。分页靠startIndex递增resultsPerPage上限 2000。循环终止条件有两个偏移量超过总数或者本页返回空双保险防止死循环。参数上窗口建议按天切一天的数据量在几千条量级单次跑完不会太久也方便失败后重跑某一天。3.3 解析、归一化与增量入库拿到原始条目后要把它拍平成前面设计的字段再写库。增量入库的核心是「先查旧值有变化就记变更」。import json import sqlite3 from datetime import datetime, timezone def parse_item(item): 把 NVD 原始条目拍平成统一字段 cve item.get(cve, {}) cve_id cve.get(id) published cve.get(published) desc for d in cve.get(descriptions, []): if d.get(lang) en: desc d.get(value, ) break score None metrics cve.get(metrics, {}) for key in (cvssMetricV31, cvssMetricV30, cvssMetricV2): if metrics.get(key): score metrics[key][0][cvssData][baseScore] break return { cve_id: cve_id, source: nvd, published: published, cvss_score: score, severity: score_to_severity(score), description: desc, affected: json.dumps(cve.get(configurations, []), ensure_asciiFalse), refs: json.dumps([r.get(url) for r in cve.get(references, [])], ensure_asciiFalse), raw: json.dumps(item, ensure_asciiFalse), } def upsert_vuln(conn, row): 写入主表并把字段变化记入变更表 cur conn.cursor() cur.execute(SELECT cvss_score, severity, description FROM vulns WHERE cve_id?, (row[cve_id],)) old cur.fetchone() now datetime.now(timezone.utc).isoformat() if old: for field, old_val in zip((cvss_score, severity, description), old): if str(old_val) ! str(row[field]): cur.execute( INSERT INTO vuln_changes(cve_id,field,old_value,new_value,changed_at) VALUES(?,?,?,?,?), (row[cve_id], field, str(old_val), str(row[field]), now)) cur.execute( INSERT OR REPLACE INTO vulns (cve_id,source,published,cvss_score,severity,description,affected,refs,raw,updated_at) VALUES(:cve_id,:source,:published,:cvss_score,:severity,:description, :affected,:refs,:raw,:updated_at) , {**row, updated_at: now}) conn.commit()逻辑说明parse_item里评分提取按 v3.1、v3.0、v2 的顺序找优先用新版本评分这是行业惯例。upsert_vuln先查旧记录逐字段比对有差异就写变更表再整体替换主表。参数上ensure_asciiFalse保证中文描述不被转义成\uXXXX方便直接阅读。INSERT OR REPLACE依赖主键去重所以cve_id不能为空解析时要过滤掉没有编号的脏数据。4. 避坑与排查那些跑一晚上才发现的问题4.1 时间格式差一个毫秒接口直接返回 404现象脚本请求 NVD 接口参数看着没问题但一直返回 404 或者空结果。原因接口要求时间格式精确到毫秒2024-01-01T00:00:00这种不带.000的写法会被判为非法。解决统一用strftime(%Y-%m-%dT%H:%M:%S.000)格式化别手拼字符串。4.2 全量拉取把内存撑爆现象想一次性把历史数据全拉下来脚本跑着跑着内存飙升然后被系统杀掉。原因把所有条目先攒在列表里再统一入库十几万条 JSON 全驻留内存。解决改成边拉边写每页解析完立刻入库并清空列表或者用生成器逐条 yield。4.3 限流触发后没有退避IP 被临时封禁现象连续请求一段时间后所有请求都返回 429等很久才恢复。原因没做主动限速重试策略里也没把 429 的退避时间拉长。解决无 Key 时强制 6 秒间隔有 Key 也别低于 0.6 秒重试的backoff_factor调大遇到 429 时额外 sleep 一段时间再继续。4.4 描述字段语言选错中文库里全是英文现象入库后发现描述全是英文检索体验很差。原因解析时没按lang过滤默认取了数组第一个元素而第一个往往是英文。解决遍历descriptions优先取lang zh没有再用en都取不到就留空并标记。4.5 增量窗口重叠导致重复告警现象同一个漏洞的变更被反复告警明明只改了一次。原因时间窗口按天切但边界处理不当同一条数据被相邻两天各抓一次第二次比对时旧值已经是新值理论上不该告警但如果入库顺序乱了就会误报。解决窗口之间留出重叠区但入库前先按cve_id加updated_at判断只有updated_at更新且字段确实不同才记变更。5. 定时调度与告警联动让这套东西自己跑起来5.1 用调度器把抓取串成流水线脚本能跑通只是第一步真正省心的是让它定时自己跑。我一般用轻量调度库把「拉取—解析—入库—告警」串成一个任务。from apscheduler.schedulers.blocking import BlockingScheduler from datetime import datetime, timedelta, timezone def daily_job(): 每天拉取前一天的漏洞数据并入库 session build_session(api_keyNone) end datetime.now(timezone.utc) start end - timedelta(days1) items fetch_nvd_window(session, start, end) conn sqlite3.connect(vulns.db) for item in items: row parse_item(item) if row[cve_id]: upsert_vuln(conn, row) conn.close() if __name__ __main__: scheduler BlockingScheduler(timezoneUTC) scheduler.add_job(daily_job, cron, hour2, minute0) scheduler.start()逻辑说明cron触发器设成每天凌晨 2 点避开白天接口高峰。任务里每次新建连接、用完关闭避免长连接失效。参数上timezoneUTC很重要漏洞库时间都是 UTC本地时区容易算错窗口。如果数据量大可以把daily_job拆成多个子任务并行但要注意总请求频率别超限。5.2 告警规则只推你真正关心的抓下来不推等于没抓。告警规则我建议分三档新增 CRITICAL/HIGH 漏洞必推已有漏洞评分上调超过 2 分必推命中资产关键词的漏洞必推。前两档靠 SQL 就能筛第三档需要维护一个资产关键词表。-- 筛选最近一天需要告警的漏洞 SELECT cve_id, severity, cvss_score, description FROM vulns WHERE published datetime(now, -1 day) AND severity IN (CRITICAL, HIGH) UNION SELECT v.cve_id, v.severity, v.cvss_score, v.description FROM vuln_changes c JOIN vulns v ON v.cve_id c.cve_id WHERE c.field cvss_score AND CAST(c.new_value AS REAL) - CAST(c.old_value AS REAL) 2.0 AND c.changed_at datetime(now, -1 day);逻辑说明第一个查询抓新增高危第二个查询抓评分跳涨。datetime(now, -1 day)是 SQLite 的相对时间写法注意它用的是 UTC。参数上评分跳涨阈值设 2.0 是经验值设太小会被噪音淹没设太大又会漏掉重要变化可以按自己资产的敏感度调。5.3 验证抓取质量三个必看的指标跑起来之后怎么知道抓得对不对我固定看三个数当日新增条数和官方公告量级对比差太多说明漏抓、字段完整率cvss_score和description非空的比例低于 90% 说明解析有问题、变更条数突然暴涨往往是源改了字段格式。这三个数写成一个巡检脚本每天跑完顺手打印比事后翻日志高效得多。def health_check(conn): 抓取质量巡检新增量、完整率、变更量 cur conn.cursor() cur.execute(SELECT COUNT(*) FROM vulns WHERE published datetime(now,-1 day)) new_count cur.fetchone()[0] cur.execute(SELECT AVG(CASE WHEN cvss_score IS NOT NULL THEN 1.0 ELSE 0 END), AVG(CASE WHEN description ! THEN 1.0 ELSE 0 END) FROM vulns WHERE published datetime(now,-7 day)) score_rate, desc_rate cur.fetchone() cur.execute(SELECT COUNT(*) FROM vuln_changes WHERE changed_at datetime(now,-1 day)) change_count cur.fetchone()[0] print(f新增{new_count} 评分完整率{score_rate:.2%} f描述完整率{desc_rate:.2%} 变更{change_count})逻辑说明完整率用AVG(CASE WHEN ... THEN 1.0 ELSE 0 END)算比例比先 count 再除更简洁。参数上完整率统计窗口取 7 天避免单日波动误判。这三个指标我一般会写进定时任务的收尾跑完自动打印哪天数字不对一眼就能看出来。这套东西我从最早的裸脚本一路改到现在最大的教训是别追求一次抓全先把增量跑稳。我早期贪心想一次把历史全拉下来结果限流、内存、字段错位全撞上折腾了两周才回到正轨。后来改成每天只拉一天的量稳定跑了几个月再回头补历史反而顺得多。如果你也打算做漏洞情报管道建议就从「一天窗口 一张主表 一个巡检」开始跑顺了再往上加源、加告警、加资产匹配。希望帮到你。本文还有配套的精品资源点击获取 
长河君 · 2026/10/14 16:21:14 · 阅读 1.2k Python爬取漏洞库实战：数据源选型、字段归一化与增量入库
简介这是一份面向网络安全初学者与Python爬虫爱好者的实战项目包围绕CNVD、CNNVD、CVE三大权威漏洞库的数据采集展开解决漏洞信息分散、难以本地统一管理的问题。项目将爬虫抓取、MySQL持久化存储与GUI可视化预览串联成完整链路适合作为课程设计、毕业设计或安全分析入门的练手素材。压缩包共72个文件约137.95MB以py爬虫与GUI脚本、toc与spec打包配置、csv与xlsx漏洞数据表、exe可执行程序及pyc缓存为主另含xml、txt、html等辅助文件结构上区分了爬取、构建与发布目录。目前已有1584人学习下载。读者可从中获得requests与BeautifulSoup解析思路、mysql-connector-python入库写法、增量与完整爬取策略以及PyQt/Tkinter界面搭建和chromedriver调用等可复用经验并理解延时、代理等反爬处理与合规边界。1. 从「python爬取漏洞库.zip」说起一份漏洞情报到底该怎么攒很多做安全运营或者漏洞管理的同行第一次看到「python爬取漏洞库.zip」这类标题脑子里冒出来的问题其实很朴素我手上有一堆资产每天要盯的漏洞公告散在好几个站点人工刷根本刷不过来能不能用 Python 把公开漏洞库的数据定时抓下来落到本地做检索、比对和告警这个方向是成立的而且落地成本比想象中低——一个稳定的抓取脚本加上一张结构合理的表就能把「今天有哪些新漏洞和我有关」这件事从人工变成自动。但这里有个反直觉的结论爬漏洞库最难的不是写 requests而是数据源选型、字段归一化和增量去重。漏洞编号有 CVE、CNVD、CNNVD 等多套体系同一漏洞在不同库里的描述、评分、影响版本写法都不一样抓下来不归一化后面根本没法用。这篇文章面向的是想自己搭一套漏洞情报管道的工程师从数据源怎么挑、脚本怎么写、字段怎么对齐一路讲到定时任务和踩坑新手能照着跑通熟手能直接拿去改。2. 数据源选型与字段设计先想清楚要抓什么2.1 三类漏洞数据源的取舍公开漏洞数据源大致分三类选型直接决定你后面脚本的复杂度。第一类是结构化 API 型典型代表是 NVD 提供的 REST 接口。它返回 JSON字段规范有分页参数适合做主力数据源。缺点是更新有延迟且对请求频率有要求需要申请 API Key 才能提高配额。第二类是网页列表型很多国家级漏洞库以 HTML 列表加详情页的形式发布。这类源信息全但需要解析 HTML页面结构一变脚本就翻车维护成本高。常见做法是用它做补充而不是主力。第三类是聚合镜像型一些开源项目会把多个源的数据合并后提供 JSON 或 SQLite 文件。好处是省事坏处是字段口径由别人定你没法控制更新节奏。我一般会这样组合NVD API 做主力网页源做补充聚合镜像做冷启动的历史数据回填。这样既有实时性又有历史深度。数据源类型代表形式优点缺点建议角色结构化 APIREST/JSON字段规范、易分页有频率限制主力网页列表HTML 页面信息全、更新快结构易变补充聚合镜像JSON/SQLite省事、含历史口径不可控冷启动2.2 字段归一化把不同库的字段对齐到一张表不管你抓几个源最后都要落到同一张表里。我建议的最小字段集是这样的cve_id统一编号没有 CVE 的用源内编号加前缀source来源标识方便回溯published发布时间统一成 ISO8601cvss_score评分统一到 0-10 浮点severity等级统一成 CRITICAL/HIGH/MEDIUM/LOWdescription描述文本affected受影响产品与版本尽量结构化references参考链接数组raw原始 JSON 备份出问题时的后悔药这里的关键是severity的映射。不同源用的等级词不一样有的用「严重/高危/中危/低危」有的用英文有的干脆只给分数。我的做法是以分数为准反推等级映射规则写死在一个字典里避免每个源各写一套判断逻辑。# severity 映射以 CVSS 分数为准统一等级口径 SEVERITY_MAP [ (9.0, CRITICAL), (7.0, HIGH), (4.0, MEDIUM), (0.1, LOW), (0.0, NONE), ] def score_to_severity(score): 把 0-10 的分数映射成统一等级None 归为 UNKNOWN if score is None: return UNKNOWN for threshold, label in SEVERITY_MAP: if score threshold: return label return UNKNOWN这段逻辑说明SEVERITY_MAP按分数从高到低排列遍历时第一个满足score threshold的就是结果所以顺序不能乱。参数上阈值取 9.0/7.0/4.0/0.1 是业界通用口径和 CVSS v3 的等级划分一致。如果某个源只给了等级没给分数就反过来查表补一个中间值保证字段完整。2.3 数据库表结构一张主表加一张变更表落库我推荐 SQLite 起步单文件、零运维几万条数据毫无压力。表结构分两张vulns存最新状态vuln_changes存每次抓取发现的字段变化。-- 主表每个漏洞一行存最新状态 CREATE TABLE IF NOT EXISTS vulns ( cve_id TEXT PRIMARY KEY, source TEXT NOT NULL, published TEXT, cvss_score REAL, severity TEXT, description TEXT, affected TEXT, refs TEXT, -- JSON 数组字符串 raw TEXT, -- 原始 JSON 备份 updated_at TEXT ); -- 变更表记录字段级变化用于增量告警 CREATE TABLE IF NOT EXISTS vuln_changes ( id INTEGER PRIMARY KEY AUTOINCREMENT, cve_id TEXT, field TEXT, old_value TEXT, new_value TEXT, changed_at TEXT ); CREATE INDEX IF NOT EXISTS idx_vulns_published ON vulns(published); CREATE INDEX IF NOT EXISTS idx_vulns_severity ON vulns(severity);逻辑说明cve_id做主键天然去重INSERT OR REPLACE就能实现「有则更新无则插入」。vuln_changes单独一张表是为了做告警——比如某个漏洞的评分从 5.0 涨到 9.8这种变化比新漏洞更值得关注。索引建在published和severity上因为日常查询基本都按时间和等级过滤。参数上refs和raw用 TEXT 存 JSON 字符串SQLite 的 JSON 函数可以直接查不用额外拆表。3. 抓取脚本落地从单次请求到增量入库3.1 带重试和限速的请求封装直接裸写requests.get是新手最容易翻车的地方——网络抖动、限流、超时全都没处理跑一晚上可能一半数据是空的。我一般会封装一个带重试和限速的会话对象。import time import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def build_session(api_keyNone): 构造带重试策略的会话统一超时和请求头 session requests.Session() retry Retry( total3, # 最多重试 3 次 backoff_factor1.5, # 退避系数间隔 1.5s、3s、6s status_forcelist[429, 500, 502, 503, 504], allowed_methods[GET], ) adapter HTTPAdapter(max_retriesretry) session.mount(https://, adapter) session.headers.update({ User-Agent: vuln-collector/1.0, Accept: application/json, }) if api_key: session.headers[apiKey] api_key return session def fetch_json(session, url, paramsNone, min_interval6.0): 请求 JSON 接口强制最小请求间隔避免触发限流 time.sleep(min_interval) resp session.get(url, paramsparams, timeout30) resp.raise_for_status() return resp.json()逻辑说明Retry的backoff_factor决定退避节奏1.5 意味着第一次重试等 1.5 秒第二次 3 秒第三次 6 秒避免短时间内反复冲击对方接口。status_forcelist里放的是值得重试的状态码429 是限流5xx 是服务端问题4xx 里的 404 不该重试所以没放。min_interval是主动限速NVD 无 Key 时建议 6 秒一次有 Key 可以降到 0.6 秒。参数上timeout30是经验值漏洞接口偶尔响应慢设太短会误判失败。3.2 分页拉取与时间窗口控制漏洞库动辄十几万条不可能每次全量拉。正确做法是按发布时间窗口增量拉取每次只取上次成功时间之后的数据。from datetime import datetime, timedelta, timezone def fetch_nvd_window(session, start, end, api_keyNone): 按时间窗口分页拉取返回原始条目列表 base https://services.nvd.nist.gov/rest/json/cves/2.0 page_size 2000 offset 0 results [] while True: params { pubStartDate: start.strftime(%Y-%m-%dT%H:%M:%S.000), pubEndDate: end.strftime(%Y-%m-%dT%H:%M:%S.000), resultsPerPage: page_size, startIndex: offset, } data fetch_json(session, base, params, min_interval0.6 if api_key else 6.0) items data.get(vulnerabilities, []) results.extend(items) total data.get(totalResults, 0) offset page_size if offset total or not items: break return results逻辑说明时间窗口用pubStartDate和pubEndDate卡住格式必须是 ISO8601 带毫秒这是接口的硬要求格式错了直接返回 404。分页靠startIndex递增resultsPerPage上限 2000。循环终止条件有两个偏移量超过总数或者本页返回空双保险防止死循环。参数上窗口建议按天切一天的数据量在几千条量级单次跑完不会太久也方便失败后重跑某一天。3.3 解析、归一化与增量入库拿到原始条目后要把它拍平成前面设计的字段再写库。增量入库的核心是「先查旧值有变化就记变更」。import json import sqlite3 from datetime import datetime, timezone def parse_item(item): 把 NVD 原始条目拍平成统一字段 cve item.get(cve, {}) cve_id cve.get(id) published cve.get(published) desc for d in cve.get(descriptions, []): if d.get(lang) en: desc d.get(value, ) break score None metrics cve.get(metrics, {}) for key in (cvssMetricV31, cvssMetricV30, cvssMetricV2): if metrics.get(key): score metrics[key][0][cvssData][baseScore] break return { cve_id: cve_id, source: nvd, published: published, cvss_score: score, severity: score_to_severity(score), description: desc, affected: json.dumps(cve.get(configurations, []), ensure_asciiFalse), refs: json.dumps([r.get(url) for r in cve.get(references, [])], ensure_asciiFalse), raw: json.dumps(item, ensure_asciiFalse), } def upsert_vuln(conn, row): 写入主表并把字段变化记入变更表 cur conn.cursor() cur.execute(SELECT cvss_score, severity, description FROM vulns WHERE cve_id?, (row[cve_id],)) old cur.fetchone() now datetime.now(timezone.utc).isoformat() if old: for field, old_val in zip((cvss_score, severity, description), old): if str(old_val) ! str(row[field]): cur.execute( INSERT INTO vuln_changes(cve_id,field,old_value,new_value,changed_at) VALUES(?,?,?,?,?), (row[cve_id], field, str(old_val), str(row[field]), now)) cur.execute( INSERT OR REPLACE INTO vulns (cve_id,source,published,cvss_score,severity,description,affected,refs,raw,updated_at) VALUES(:cve_id,:source,:published,:cvss_score,:severity,:description, :affected,:refs,:raw,:updated_at) , {**row, updated_at: now}) conn.commit()逻辑说明parse_item里评分提取按 v3.1、v3.0、v2 的顺序找优先用新版本评分这是行业惯例。upsert_vuln先查旧记录逐字段比对有差异就写变更表再整体替换主表。参数上ensure_asciiFalse保证中文描述不被转义成\uXXXX方便直接阅读。INSERT OR REPLACE依赖主键去重所以cve_id不能为空解析时要过滤掉没有编号的脏数据。4. 避坑与排查那些跑一晚上才发现的问题4.1 时间格式差一个毫秒接口直接返回 404现象脚本请求 NVD 接口参数看着没问题但一直返回 404 或者空结果。原因接口要求时间格式精确到毫秒2024-01-01T00:00:00这种不带.000的写法会被判为非法。解决统一用strftime(%Y-%m-%dT%H:%M:%S.000)格式化别手拼字符串。4.2 全量拉取把内存撑爆现象想一次性把历史数据全拉下来脚本跑着跑着内存飙升然后被系统杀掉。原因把所有条目先攒在列表里再统一入库十几万条 JSON 全驻留内存。解决改成边拉边写每页解析完立刻入库并清空列表或者用生成器逐条 yield。4.3 限流触发后没有退避IP 被临时封禁现象连续请求一段时间后所有请求都返回 429等很久才恢复。原因没做主动限速重试策略里也没把 429 的退避时间拉长。解决无 Key 时强制 6 秒间隔有 Key 也别低于 0.6 秒重试的backoff_factor调大遇到 429 时额外 sleep 一段时间再继续。4.4 描述字段语言选错中文库里全是英文现象入库后发现描述全是英文检索体验很差。原因解析时没按lang过滤默认取了数组第一个元素而第一个往往是英文。解决遍历descriptions优先取lang zh没有再用en都取不到就留空并标记。4.5 增量窗口重叠导致重复告警现象同一个漏洞的变更被反复告警明明只改了一次。原因时间窗口按天切但边界处理不当同一条数据被相邻两天各抓一次第二次比对时旧值已经是新值理论上不该告警但如果入库顺序乱了就会误报。解决窗口之间留出重叠区但入库前先按cve_id加updated_at判断只有updated_at更新且字段确实不同才记变更。5. 定时调度与告警联动让这套东西自己跑起来5.1 用调度器把抓取串成流水线脚本能跑通只是第一步真正省心的是让它定时自己跑。我一般用轻量调度库把「拉取—解析—入库—告警」串成一个任务。from apscheduler.schedulers.blocking import BlockingScheduler from datetime import datetime, timedelta, timezone def daily_job(): 每天拉取前一天的漏洞数据并入库 session build_session(api_keyNone) end datetime.now(timezone.utc) start end - timedelta(days1) items fetch_nvd_window(session, start, end) conn sqlite3.connect(vulns.db) for item in items: row parse_item(item) if row[cve_id]: upsert_vuln(conn, row) conn.close() if __name__ __main__: scheduler BlockingScheduler(timezoneUTC) scheduler.add_job(daily_job, cron, hour2, minute0) scheduler.start()逻辑说明cron触发器设成每天凌晨 2 点避开白天接口高峰。任务里每次新建连接、用完关闭避免长连接失效。参数上timezoneUTC很重要漏洞库时间都是 UTC本地时区容易算错窗口。如果数据量大可以把daily_job拆成多个子任务并行但要注意总请求频率别超限。5.2 告警规则只推你真正关心的抓下来不推等于没抓。告警规则我建议分三档新增 CRITICAL/HIGH 漏洞必推已有漏洞评分上调超过 2 分必推命中资产关键词的漏洞必推。前两档靠 SQL 就能筛第三档需要维护一个资产关键词表。-- 筛选最近一天需要告警的漏洞 SELECT cve_id, severity, cvss_score, description FROM vulns WHERE published datetime(now, -1 day) AND severity IN (CRITICAL, HIGH) UNION SELECT v.cve_id, v.severity, v.cvss_score, v.description FROM vuln_changes c JOIN vulns v ON v.cve_id c.cve_id WHERE c.field cvss_score AND CAST(c.new_value AS REAL) - CAST(c.old_value AS REAL) 2.0 AND c.changed_at datetime(now, -1 day);逻辑说明第一个查询抓新增高危第二个查询抓评分跳涨。datetime(now, -1 day)是 SQLite 的相对时间写法注意它用的是 UTC。参数上评分跳涨阈值设 2.0 是经验值设太小会被噪音淹没设太大又会漏掉重要变化可以按自己资产的敏感度调。5.3 验证抓取质量三个必看的指标跑起来之后怎么知道抓得对不对我固定看三个数当日新增条数和官方公告量级对比差太多说明漏抓、字段完整率cvss_score和description非空的比例低于 90% 说明解析有问题、变更条数突然暴涨往往是源改了字段格式。这三个数写成一个巡检脚本每天跑完顺手打印比事后翻日志高效得多。def health_check(conn): 抓取质量巡检新增量、完整率、变更量 cur conn.cursor() cur.execute(SELECT COUNT(*) FROM vulns WHERE published datetime(now,-1 day)) new_count cur.fetchone()[0] cur.execute(SELECT AVG(CASE WHEN cvss_score IS NOT NULL THEN 1.0 ELSE 0 END), AVG(CASE WHEN description ! THEN 1.0 ELSE 0 END) FROM vulns WHERE published datetime(now,-7 day)) score_rate, desc_rate cur.fetchone() cur.execute(SELECT COUNT(*) FROM vuln_changes WHERE changed_at datetime(now,-1 day)) change_count cur.fetchone()[0] print(f新增{new_count} 评分完整率{score_rate:.2%} f描述完整率{desc_rate:.2%} 变更{change_count})逻辑说明完整率用AVG(CASE WHEN ... THEN 1.0 ELSE 0 END)算比例比先 count 再除更简洁。参数上完整率统计窗口取 7 天避免单日波动误判。这三个指标我一般会写进定时任务的收尾跑完自动打印哪天数字不对一眼就能看出来。这套东西我从最早的裸脚本一路改到现在最大的教训是别追求一次抓全先把增量跑稳。我早期贪心想一次把历史全拉下来结果限流、内存、字段错位全撞上折腾了两周才回到正轨。后来改成每天只拉一天的量稳定跑了几个月再回头补历史反而顺得多。如果你也打算做漏洞情报管道建议就从「一天窗口 一张主表 一个巡检」开始跑顺了再往上加源、加告警、加资产匹配。希望帮到你。本文还有配套的精品资源点击获取