Python构建漏洞情报采集流水线：从数据抓取到资产匹配的实战指南

1. 为什么我要自己动手搭一套漏洞情报采集流水线

做安全运营的同行大概都有类似的体会：每天早上打开邮箱，各种漏洞通告、安全公告、厂商补丁说明铺天盖地，光是筛选出跟自己资产相关的那几条，就得花掉一两个小时。更麻烦的是，很多漏洞信息分散在不同页面、不同格式里，有的用表格，有的用列表，有的干脆就是一大段纯文本，想做个统计或者跟内部资产库做匹配，手工整理几乎不可能。

我所在的团队负责维护一批线上业务系统，资产规模不算特别大，但涉及的组件种类很杂，从常见的Web中间件到各种开源依赖库都有。之前一直靠人工订阅邮件列表加手工记录的方式做漏洞跟踪，效率低不说，还经常漏掉一些关键信息。后来我下定决心，花了两周时间搭了一套自动化的漏洞情报采集与处理流水线，核心目标就一个：把公开渠道发布的漏洞数据自动抓下来、清洗干净、结构化存储，然后跟内部资产清单做匹配，输出一份每天可以直接看的风险清单。

这套东西的技术栈其实不复杂，Python加上几个常用的解析库就能跑起来。但真正做下来才发现，坑比想象中多得多。数据源的页面结构会变、反爬策略会升级、字段格式不统一、去重逻辑要考虑版本号比较……每一个环节都有细节需要打磨。这篇文章就把我整个搭建过程中积累的经验、踩过的坑、以及最终跑通的方案完整分享出来，适合有一定Python基础、想做安全数据自动化采集的朋友参考。哪怕你之前没接触过爬虫，跟着思路走也能理解整套逻辑。

2. 整体架构设计与技术选型思路

2.1 为什么选择分层架构而不是一把梭

最开始我想得特别简单：写个脚本，请求页面、解析HTML、存数据库，完事。但实际跑起来发现，如果把请求、解析、存储全揉在一个函数里，后面任何一个环节出问题都得从头调试，而且数据源一多，代码会变得极其臃肿。

所以我最终采用的是分层架构，把整个流程拆成四个独立的层：

• 采集层：负责发送HTTP请求、处理重试和限速，只关心“把页面内容拿回来”
• 解析层：负责从HTML或JSON中提取结构化字段，只关心“把内容变成字典”
• 处理层：负责数据清洗、格式统一、去重和版本比较，只关心“把字典变成规范记录”
• 存储与输出层：负责写入数据库、生成报表、推送通知，只关心“把记录送到该去的地方”

这样拆的好处非常明显。比如某天某个数据源的页面改版了，我只需要改解析层对应的那个解析器，采集层和处理层完全不用动。再比如后面想加一个新的输出渠道，也只需要在输出层扩展，不影响前面的逻辑。

提示：分层不是目的，解耦才是。如果你的数据源只有一个且结构极其稳定，硬套分层反而增加复杂度。但只要数据源超过两个，或者页面结构可能变化，分层就是值得的。

2.2 技术栈选择与理由

在具体工具的选择上，我对比过几种方案，最终确定的组合是这样的：

这里重点说一下为什么用BeautifulSoup而不是Scrapy。Scrapy确实强大，自带调度、去重、中间件，但它更适合大规模、多站点的爬取任务。我的场景是数据源数量有限（大概十几个），每个源的页面结构差异大，需要针对性地写解析逻辑，用Scrapy反而显得笨重。requests加BeautifulSoup的组合更灵活，调试也方便，出问题了直接print就能定位。

至于存储，一开始我用的是SQLite，单文件、零配置，本地跑完全没问题。后来团队其他人也要用这份数据，就迁移到了PostgreSQL，改的只是连接字符串和少量SQL语法，整体迁移成本很低。

2.3 数据源分类与采集策略

公开的漏洞信息来源大致可以分成几类，每类的采集策略不太一样：

第一类是结构化的漏洞数据库，这类源通常有固定的字段和分页机制，页面结构相对稳定，解析起来最省心。采集频率可以设为每天一次，因为更新不会特别频繁。

第二类是厂商发布的安全公告页面，这类页面的特点是格式不统一，有的用表格，有的用嵌套的div，而且不同厂商的字段命名差异很大。采集频率建议每天两次，因为厂商发布补丁往往比较紧急。

第三类是社区讨论和邮件列表归档，这类源信息量大但噪音也大，需要做更多的过滤和关键词匹配。采集频率可以放宽到每两天一次，重点是把标题和摘要抓下来做初步筛选。

针对不同类型的源，我在采集层设置了不同的请求头、超时时间和重试策略。比如结构化数据库的源响应快，超时设10秒就够；厂商公告页面有时候加载慢，超时设到30秒，重试次数也多加一次。

3. 核心模块拆解与关键实现细节

3.1 采集层：请求封装与反爬应对

采集层看起来简单，但细节最多。我封装了一个通用的请求函数，核心逻辑包括：设置合理的请求头、控制请求间隔、处理超时和重试、记录请求日志。

请求头这块，最关键的是User-Agent和Accept-Language。很多站点会检查User-Agent来判断是不是正常浏览器访问，用一个常见的浏览器UA字符串能避免大部分基础拦截。Accept-Language设成中文环境，有些站点会根据这个返回不同的内容版本。

请求间隔的控制非常重要。我一开始没注意，连续快速请求同一个站点，结果IP被临时封了半小时。后来加了一个简单的令牌桶限流，对同一个域名的请求间隔至少1.5秒，不同域名之间可以并行。实测下来，这个间隔既能保证采集效率，又不会触发大部分站点的频率限制。

重试策略我用的是指数退避：第一次失败等2秒重试，第二次等4秒，第三次等8秒，最多重试三次。这样设计的原因是，很多临时性失败（比如网络抖动、服务端短暂过载）在几秒后就能恢复，立即重试反而容易连续失败。

import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry import time def create_session(): session = requests.Session() retry = Retry( total=3, backoff_factor=2, status_forcelist=[500, 502, 503, 504] ) adapter = HTTPAdapter(max_retries=retry) session.mount('http://', adapter) session.mount('https://', adapter) session.headers.update({ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36', 'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8' }) return session _last_request_time = {} def rate_limited_get(session, url, min_interval=1.5): from urllib.parse import urlparse domain = urlparse(url).netloc now = time.time() if domain in _last_request_time: elapsed = now - _last_request_time[domain] if elapsed < min_interval: time.sleep(min_interval - elapsed) resp = session.get(url, timeout=15) _last_request_time[domain] = time.time() return resp

这段代码里，Retry的backoff_factor=2意味着重试间隔会按2的幂次增长，配合status_forcelist只对服务端错误重试，客户端错误（如404）不重试，避免无意义的等待。

注意：限流的时间间隔不要设得太死。有些站点在夜间维护时段响应特别慢，如果间隔设得太短，容易触发超时。我后来把间隔改成了可配置的，不同源用不同的值，灵活很多。

3.2 解析层：多格式内容提取与字段映射

解析层是整个流水线里最需要耐心的地方，因为每个数据源的页面结构都不一样。我的做法是给每个数据源写一个独立的解析器类，统一继承一个基类，基类定义接口，子类实现具体逻辑。

基类定义了两个核心方法：parse_list负责从列表页提取出详情页的链接，parse_detail负责从详情页提取出结构化字段。这样设计的好处是，列表页和详情页的解析逻辑分开，互不干扰。

字段映射是另一个关键点。不同源对同一个概念的叫法完全不同，比如漏洞编号，有的叫“CVE ID”，有的叫“漏洞编号”，有的直接就是“ID”。我在解析层统一映射成标准字段名：cve_id、title、severity、publish_date、description、affected_products、reference_links。

class BaseParser: def parse_list(self, html): raise NotImplementedError def parse_detail(self, html): raise NotImplementedError def normalize_severity(self, raw_text): mapping = { 'critical': '严重', '高危': '高危', 'high': '高危', 'medium': '中危', '中危': '中危', 'low': '低危', '低危': '低危' } raw_lower = raw_text.strip().lower() for key, val in mapping.items(): if key in raw_lower: return val return '未知'

严重程度的归一化特别重要。有的源用英文（Critical/High/Medium/Low），有的用中文（严重/高危/中危/低危），还有的用数字评分（9.8、7.5）。我统一映射成四个等级，方便后续做优先级排序。

日期格式也是个大坑。有的源用“2024-01-15”，有的用“Jan 15, 2024”，还有的用“2024年1月15日”。我写了一个通用的日期解析函数，用dateutil库做兜底，能覆盖绝大多数格式。

3.3 处理层：数据清洗、去重与版本比较

处理层要做的事情比想象中多。原始数据抓下来之后，至少需要经过这几道处理：

第一道是空值处理。有些字段在某些源里可能不存在，比如有的公告不写受影响产品版本。对于这种情况，我统一填“未提供”，而不是留空。留空在后续做统计和匹配时容易出问题。

第二道是文本清洗。HTML标签残留、多余的空格和换行、特殊字符编码问题，这些都要处理。我写了一个clean_text函数，用正则表达式去掉HTML标签，把连续空白字符合并成一个空格，再统一编码格式。

第三道是去重。去重不能简单地按CVE编号来，因为同一个CVE可能在不同源里出现多次，而且描述信息可能略有差异。我的策略是：优先按CVE编号去重，如果CVE编号缺失，则按“标题+受影响产品”的组合去重。去重时保留信息最完整的那条记录。

第四道是版本号比较。这个是最容易出错的环节。漏洞影响范围通常用版本号区间表示，比如“影响 2.0.0 到 2.3.5 之间的版本”。要判断某个具体版本是否在影响范围内，就需要做版本号比较。版本号比较不能直接按字符串比，因为“2.10.0”在字符串比较里小于“2.9.0”，但实际版本号是大于的。

def compare_version(v1, v2): def normalize(v): parts = v.split('.') result = [] for p in parts: num = ''.join(filter(str.isdigit, p)) result.append(int(num) if num else 0) return result n1, n2 = normalize(v1), normalize(v2) max_len = max(len(n1), len(n2)) n1 += [0] * (max_len - len(n1)) n2 += [0] * (max_len - len(n2)) for a, b in zip(n1, n2): if a > b: return 1 if a < b: return -1 return 0

这个函数把版本号拆成数字列表，逐位比较，能正确处理“2.10.0 > 2.9.0”这种情况。对于带字母的版本号（如“1.0.0-beta”），我做了简化处理，只取数字部分，因为大多数漏洞影响范围描述不会精确到beta版本。

3.4 存储层：表结构设计与索引优化

存储层的表结构设计直接影响后续查询效率。我最终确定的表结构包含以下字段：

索引方面，cve_id建唯一索引，publish_date和severity建普通索引。这样设计的原因是，日常查询最常用的条件就是“最近一周的高危漏洞”，publish_date和severity的组合索引能大幅提升这类查询的速度。

受影响产品和参考链接用JSON格式存储，是因为这两个字段的内容长度不固定，而且后续可能需要做更复杂的查询。如果用的是PostgreSQL，可以直接用JSONB类型，查询效率更高。SQLite的话就用TEXT存JSON字符串，在应用层做解析。

提示：如果数据量不大（几万条以内），SQLite完全够用，而且备份和迁移特别方便，直接复制文件就行。数据量上到十万级以上，或者需要多人同时写入，再考虑迁移到PostgreSQL。

4. 完整实操流程与关键环节实现

4.1 环境准备与依赖安装

整个项目用Python 3.9以上版本就能跑，依赖库不多，用pip一次性装好：

pip install requests beautifulsoup4 lxml pandas python-dateutil schedule

如果要用PostgreSQL，再加一个psycopg2-binary。我建议用虚拟环境来管理依赖，避免跟系统里的其他Python项目冲突：

python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows pip install -r requirements.txt

项目目录结构我习惯这样组织：

vuln_collector/ ├── config/ │ └── sources.yaml # 数据源配置 ├── collectors/ │ ├── base.py # 采集基类 │ └── http_client.py # 请求封装 ├── parsers/ │ ├── base.py # 解析基类 │ ├── source_a.py # 各源解析器 │ └── source_b.py ├── processors/ │ ├── cleaner.py # 清洗逻辑 │ └── dedup.py # 去重逻辑 ├── storage/ │ └── db.py # 数据库操作 ├── output/ │ └── reporter.py # 报表生成 ├── main.py # 入口 └── requirements.txt

这个结构的好处是每个模块职责清晰，新人接手时能快速定位到相关代码。配置文件用YAML格式，把数据源的URL、解析器类名、采集频率都放在配置里，加新源的时候只需要改配置和加一个解析器文件，不用动主流程。

4.2 数据源配置与解析器编写

配置文件的结构大概长这样：

sources: - name: source_a base_url: "https://example-a.com/vuln/list" parser: "SourceAParser" interval_hours: 24 max_pages: 10 - name: source_b base_url: "https://example-b.com/security" parser: "SourceBParser" interval_hours: 12 max_pages: 5

每个源配置了名称、基础URL、对应的解析器类名、采集间隔和最大翻页数。最大翻页数是为了防止某个源突然出现大量历史数据导致采集时间过长，设一个上限，只抓最新的几页。

解析器的编写我以其中一个源为例说明。假设这个源的列表页是一个表格，每行包含漏洞标题、严重等级和详情链接：

from bs4 import BeautifulSoup from parsers.base import BaseParser class SourceAParser(BaseParser): def parse_list(self, html): soup = BeautifulSoup(html, 'lxml') items = [] table = soup.find('table', class_='vuln-table') if not table: return items for row in table.find_all('tr')[1:]: cols = row.find_all('td') if len(cols) < 3: continue link = cols[0].find('a') if not link: continue items.append({ 'title': link.get_text(strip=True), 'severity': self.normalize_severity(cols[1].get_text(strip=True)), 'detail_url': link.get('href') }) return items def parse_detail(self, html): soup = BeautifulSoup(html, 'lxml') result = {} result['cve_id'] = self._extract_text(soup, 'span', 'cve-id') result['description'] = self._extract_text(soup, 'div', 'vuln-desc') result['publish_date'] = self._parse_date( self._extract_text(soup, 'span', 'publish-date') ) result['affected_products'] = self._extract_products(soup) result['reference_links'] = self._extract_links(soup) return result

这里有几个细节值得注意。parse_list里对表格行的处理加了长度检查，防止某些异常行导致索引错误。parse_detail里每个字段的提取都做了容错，提取不到就返回空字符串，而不是抛异常。这样即使页面部分改版，也不会导致整个采集任务失败。

4.3 去重与版本匹配的实操细节

去重逻辑我单独放在一个模块里，核心思路是：先按CVE编号查重，如果已存在则比较更新时间，保留更新的那条；如果CVE编号为空，则按标题的哈希值查重。

import hashlib def generate_fingerprint(record): if record.get('cve_id'): return f"cve:{record['cve_id']}" title = record.get('title', '').strip().lower() products = record.get('affected_products', '') raw = f"{title}|{products}" return f"hash:{hashlib.md5(raw.encode()).hexdigest()}"

版本匹配是另一个核心功能。内部资产清单里记录了每个组件当前使用的版本号，我需要判断这个版本是否在漏洞影响范围内。影响范围通常是一个区间，我用两个字段表示：affected_min和affected_max，分别是最小受影响版本和最大受影响版本。

def is_version_affected(current_version, min_version, max_version): if min_version and compare_version(current_version, min_version) < 0: return False if max_version and compare_version(current_version, max_version) > 0: return False return True

实际使用中，有些漏洞的影响范围不是连续的，比如“影响 1.x 所有版本和 2.0 到 2.3”。这种情况我会把影响范围拆成多条记录，每条记录对应一个连续区间，匹配时只要命中任意一条就算受影响。

注意：版本号比较一定要处理“没有数字”的情况。我遇到过某个组件的版本号是“latest”，这种没法比较，只能标记为“需人工确认”。不要试图用字符串比较来蒙混过关，否则会漏报或误报。

4.4 定时调度与增量采集实现

调度我用的是schedule库，配合一个while循环，每60秒检查一次是否有任务需要执行。这种方式的优点是配置简单，所有调度逻辑都在Python代码里，不需要额外维护crontab。

import schedule import time def job_source_a(): collector = Collector('source_a') collector.run() schedule.every(24).hours.do(job_source_a) schedule.every(12).hours.do(job_source_b) while True: schedule.run_pending() time.sleep(60)

增量采集的关键是记录每个源上次采集到的最新时间戳或最新ID。我在数据库里建了一张collector_state表，记录每个源的最后采集时间和最后一条记录的标识。下次采集时，只抓取比这个标识更新的内容。

对于支持按时间筛选的源，直接在请求参数里带上起始时间。对于不支持时间筛选的源，就只能翻页抓取，然后在处理层根据发布时间过滤掉已经处理过的记录。虽然后者效率低一些，但兼容性更好。

4.5 输出报表与通知推送

采集和处理完成之后，最终要输出一份人能看懂的报表。我生成的报表包含三部分：新增漏洞列表、高危漏洞汇总、与内部资产匹配的风险清单。

报表格式我用的是Markdown，因为可以直接贴到内部文档系统里，也方便通过Webhook推送到团队沟通工具。推送内容做了精简，只包含最关键的信息：漏洞编号、标题、严重等级、受影响组件、修复建议链接。

def build_report(new_vulns, matched_risks): lines = ["## 每日漏洞情报汇总\n"] lines.append(f"新增漏洞：{len(new_vulns)} 条\n") lines.append(f"资产匹配风险：{len(matched_risks)} 条\n") if matched_risks: lines.append("\n### 需优先处理\n") for risk in matched_risks: lines.append(f"- [{risk['severity']}] {risk['cve_id']} " f"{risk['title']}（影响组件：{risk['component']}）") return "\n".join(lines)

推送频率我设的是每天一次，早上九点推送到团队群。如果遇到严重等级为“严重”的漏洞，会额外触发一次即时推送，确保团队能第一时间响应。

5. 常见问题排查与避坑经验实录

5.1 请求被拦截或返回异常内容

这是最常见的问题。表现是请求返回403、429状态码，或者返回的页面内容明显不对（比如返回了一个验证页面）。

排查思路按这个顺序来：先检查请求头是否完整，特别是User-Agent和Referer；再检查请求频率是否过高，适当加大间隔；然后检查是否有Cookie要求，有些站点需要先访问首页获取Cookie再访问详情页。

我遇到过一个比较隐蔽的情况：某个站点对请求头里的字段顺序有要求，虽然理论上HTTP头是无序的，但那个站点的防护逻辑就是按字符串匹配来做的。后来我把请求头的顺序调整成跟浏览器一致，问题就解决了。这种问题没有通用解法，只能靠抓包对比来定位。

提示：如果某个源持续返回异常，不要死磕。先把它标记为“暂停采集”，把其他源跑通，回头再单独处理。我一开始在一个源上卡了三天，后来发现那个源本身就不稳定，隔几天再试就正常了。

5.2 解析结果字段缺失或错位

页面改版是解析层最大的敌人。表现是某些字段突然提取不到了，或者提取到的内容跟预期不符。

我的应对策略是：在解析器里加字段校验，如果关键字段（如CVE编号、标题）提取失败，记录一条警告日志，并把原始HTML保存下来供后续分析。同时，解析结果里加一个parse_quality字段，标记这条记录的解析完整度。完整度低于阈值的记录会被单独挑出来人工检查。

另一个常见问题是编码问题。有些页面声明是UTF-8，实际内容是GBK，导致中文乱码。我的处理方式是在请求返回后，先用resp.apparent_encoding检测实际编码，再用检测到的编码来解码内容，而不是盲目相信响应头里的声明。

5.3 去重逻辑误判或漏判

去重误判的典型表现是：同一条漏洞被当成两条不同记录存了进去，或者两条不同的漏洞被误判为同一条。

前者的原因通常是CVE编号格式不一致，比如有的源写“CVE-2024-1234”，有的写“cve-2024-1234”，有的写“CVE-2024-1234 ”（末尾有空格）。解决方法是统一转成大写并去掉首尾空格再做比较。

后者的原因通常是标题哈希碰撞或者标题过于相似。比如两个不同的漏洞标题都是“安全更新公告”，只是发布日期不同。这种情况我加了一个辅助判断：如果标题相同但发布日期相差超过7天，就认为是两条不同记录。

5.4 版本匹配的边界情况处理

版本匹配最容易出问题的地方是边界值。比如影响范围写的是“小于 2.3.5”，那2.3.5本身是否受影响？不同源的表述习惯不一样，有的包含边界，有的不包含。

我的处理方式是：在解析影响范围时，同时提取边界包含关系。如果原文写的是“小于等于 2.3.5”，那2.3.5就受影响；如果写的是“小于 2.3.5”，那2.3.5就不受影响。对于表述模糊的（比如只写“2.3.5及以下”），我统一按包含边界处理，宁可多报不可漏报。

下面这张表整理了我遇到过的典型问题及对应的解决思路：

5.5 采集任务中断与恢复

长时间运行的采集任务可能会因为网络中断、进程被杀等原因意外停止。我的做法是给每个源维护一个采集状态，记录当前处理到第几页、最后一条记录的时间戳。任务重启时，从上次中断的位置继续，而不是从头开始。

状态信息存在数据库里，每次成功处理完一页就更新一次。这样即使中断，最多只会丢失当前页的数据，不会影响整体进度。

另外，我在主流程外面包了一层异常捕获，任何未处理的异常都会被记录到日志文件，同时触发一次告警通知。这样即使半夜任务挂了，第二天早上也能第一时间知道。

6. 我在实际运行中积累的几条硬核经验

这套流水线跑了大半年，中间经历过几次大的调整，有些经验是踩了坑才总结出来的，这里分享几条我觉得最有价值的。

第一条：不要追求一次抓全，先跑通再优化。我一开始想把所有能想到的字段都抓下来，结果解析器写得极其复杂，调试了两周还没跑通。后来砍掉了一半非核心字段，先把主流程跑起来，后面再逐步补充。这个思路帮我节省了大量时间。

第二条：日志要详细，但不要泛滥。每个请求都打日志会导致日志文件迅速膨胀，但关键节点（如解析失败、去重命中、版本匹配异常）的日志一定要详细。我的做法是分级记录：INFO级别只记录每轮采集的汇总信息，WARNING和ERROR级别记录具体的问题详情。

第三条：数据质量比数据量重要。与其抓一万条格式混乱的数据，不如抓一千条干净规范的数据。我在处理层花的时间比采集层和解析层加起来还多，但这是值得的。干净的数据后续做任何分析都方便，脏数据只会带来无尽的麻烦。

第四条：定期回顾和清理。有些数据源可能过一段时间就失效了，或者内容质量下降。我每个月会检查一次各源的采集成功率和数据质量，把长期表现不好的源下线，把精力集中在高质量源上。

第五条：留好手动干预的入口。自动化不是万能的，总有一些情况需要人工判断。我在流程里保留了手动导入和手动修正的功能，遇到解析不了的页面或者需要人工确认的版本匹配，可以快速介入处理，不用等代码修复。

这套东西说到底就是一个思路：把重复性的信息收集工作交给程序，把人的精力留给真正需要判断和分析的事情。工具本身不复杂，难的是持续维护和迭代。希望这些经验对正在做类似事情的朋友有所帮助。