简介一份基于Scrapy框架的威胁情报抓取与处理系统的毕业设计文档面向网络安全、Python爬虫及威胁情报方向的学生与研究者。内容围绕如何利用Scrapy爬虫高效采集开源威胁网站与博客数据并完成解析、入库与展示涉及知识图谱、APT知识图谱、Flask_adminnginx、pyecharts等关键技术覆盖系统分析与设计、爬虫模块、数据解析模块、数据展示模块等章节适合作为相关课题设计、论文写作或项目开发的参考资料。资源为单个docx文件大小2.58MB共1份文档结构完整包含中英文摘要、目录及正文内容。已有194人学习下载适用性较好。1. 基于Scrapy框架的威胁情报抓取处理系统从情报源到可查指标威胁情报的价值本质上是时间的函数。攻击者更换一次 C2 域名、上线一个新的钓鱼页面留给防御方反应的时间往往按小时计算。靠人工巡看安全公众号、抓包群和厂商公告的传统采集链路字段格式不统一落地到检测规则时还要先手工清洗一遍这段周转本身就消耗了宝贵的处置窗口。基于Scrapy框架的威胁情报抓取以及处理系统要解决的正是从情报源到可查询数据资产的工程化压缩让 Scrapy 调度对安全博客、漏洞库、代码托管平台和厂商通告的分批抓取用 Item Pipeline 完成字段抽取、格式标准化和置信度去重最终写入可检索存储。这套系统适合安全分析团队和蓝队工程师他们要的不是一个简单爬虫 demo而是能把采集、清洗、入库固化为日常作业的完整链路。下文从工程搭建切入逐步展开反爬应对、字段处理和验证方法。2. Scrapy框架下威胁情报抓取系统的架构设计与工程搭建2.1 威胁情报源的类型划分与Scrapy适配性评估先说选型边界。不是所有威胁情报源都适合用 Scrapy 去抓有些直接提供 JSON API 或历史下载包用 requests 脚本反而更省事。决定用 Scrapy 的标准是信息源本身是 HTML 页面流、有翻页或站内搜索语义、需要做频控和断点续爬。符合这个条件的典型情报源有下面几类情报源类型典型表现形态抓取方式Scrapy适配度RSS/Atom 订阅厂商通告、安全博客更新XML 字段直接抽取高漏洞库NVD、CNVD 的列表详情页分页列表 详情页高代码托管仓库IoC 列表更新、README 变更文件接口或仓库提交 API中商务情报平台混合渲染页面通用爬虫或动态渲染中论坛/社区板块分页帖子通用爬虫 登录态中低选用 Scrapy 的价值在这些场景下才成立列表页与详情页结构稳定、增量以天或小时为单位变化、请求量不大但条目众多时手工维护会失控。尤其是一类常见需求比如把公众号相关文章的自动搜索监控抓取也纳入情报来源虽然公众号网页是动态渲染但通过 Scrapy 配合浏览器驱动一样能落到同一套 Pipeline 里。2.2 爬虫工程目录与Spider骨架实现创建项目属于固定动作但目录职责最好从第一天就分清楚scrapy startproject threat_intel cd threat_intel scrapy genspider -t crawl rss_feed feed.example.comstartproject生成项目骨架genspider -t crawl用 CrawlSpider 模板创建一个名为 rss_feed 的爬虫。CrawlSpider 自带链接提取规则适合列表页稳定的序列型情报源省掉手写 parse 分发。threat_intel/ ├── items.py # IoC 字段容器 ├── middlewares.py # 请求头切换、cookie 刷新 ├── pipelines.py # 清洗、去重、入库 ├── settings.py # 并发、限速、下载延迟配置 ├── spiders/ │ ├── rss_feed.py # RSS 源适配 │ ├── blog_crawler.py # 安全博客页面抓取 │ └── vuln_feed.py # 漏洞库增量抓取 └── requirements.txt如果把多个来源塞进同一个 Spider后续改动一处字段就会牵连其他源反而不利于维护。常见做法是每一类结构相近的情报源独立成 Spider在 settings 里用SPIDER_LOADER_WARN_ONLY控制加载告警避免无关源互相影响。提示Spider 文件命名建议直接使用情报源的域名或业务名爬虫日志里按爬虫名过滤时能一眼定位是哪条链路出了问题。下面这个样例是漏洞库爬虫跑的是列表页到详情页的两段跳转# spiders/vuln_feed.py from urllib.parse import urljoin from datetime import date import scrapy from scrapy.loader import ItemLoader from threat_intel.items import IoCItem class VulnFeedSpider(scrapy.Spider): name vuln_feed allowed_domains [nvd.nist.gov] start_urls [https://nvd.nist.gov/vuln/data-feeds] def parse(self, response): # 只挑列表页里 CVE 开头的条目过滤掉导航和其他噪声 for row in response.css(table.news-table tr): title row.css(a::text).get() href row.css(a::attr(href)).get() if href and title and CVE in title: yield scrapy.Request( urljoin(response.url, href), callbackself.parse_detail, meta{title: title.strip()}, ) def parse_detail(self, response): # 详情页里摘 CVE 描述不需要精确到全表 loader ItemLoader(itemIoCItem(), responseresponse) loader.add_value(ioc_type, cve) loader.add_value(ioc_value, response.url.split(/)[-1]) loader.add_css(description, #vulnDetailTableView::text) loader.add_value(source, response.url) loader.add_value(last_seen, date.today().isoformat()) yield loader.load_item()这段代码刻意的点有两个一是在 parse 里先用标题过滤再决定是否发详情页请求能少一半无意义的页面抓取二是用 ItemLoader 而不是手写字典为后续 Pipeline 统一处理留了接缝。meta 传入的 title 能在详情页拿不到字段时用作兜底不至于因为一个字段丢失丢掉整条记录。2.3 请求调度、指纹去重与增量采集Scrapy 内置的 RFPDupeFilter 默认按请求 URL 生成指纹对纯 GET 的情报源够用但同一个漏洞可能从两个情报源各抓一次需要在入库前去重。更可靠的做法是建立业务指纹# pipelines.py 中自定义业务指纹示例 import hashlib def build_fingerprint(payload): raw |.join([ payload.get(ioc_type, ), payload.get(ioc_value, ), payload.get(source_url, ), ]) return hashlib.sha256(raw.encode(utf-8)).hexdigest()指纹把类型、值和来源拼在一起做哈希。简洁设计里可以去掉source_url让同一指标只保留一条记录但威胁情报场景中来源信息是重要审计线索一般会保留。调度侧做增量采集时可以在 settings 里开启 JOBDIR# settings.py DUPEFILTER_CLASS scrapy.dupefilters.RFPDupeFilter JOBDIR scheduler_state/vuln_feed SCHEDULER_DEBUG FalseJOBDIR 的作用是把未完成的请求队列持久化到磁盘中断重启后能接着抓不会重复请求已完成页面。SCHEDULER_DEBUG 平时建议置 False不然每次调度都会输出大量调试日志反而干扰对实际采集状态的判断。3. 威胁情报抓取的中间件定制与反爬应对策略3.1 请求头伪装与Cookie生命周期刷新威胁情报源大多有基础的访问频率限制少数还会校验 User-Agent 特征。抓取端需要做的是让采集机的请求看起来像普通浏览器行为而不是一个固定不变的 curl 头。常见做法是维护一个 UA 池在中间件中随机挑选并注入请求头# middlewares.py import random USER_AGENT_POOL [ Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0 Safari/537.36, Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0 Safari/537.36, ] class RandomUAProcessMiddleware: 在下载前为每个请求注入随机的 UA def process_request(self, request, spider): request.headers[User-Agent] random.choice(USER_AGENT_POOL) return None中间件在两个方向上有用process_request在请求发出前改头process_response在拿到响应后检查状态码。如果发现 403 或者页面要求登录可以在process_response里直接丢弃响应并返回一个新请求避免污染后续解析。Cookie 过期是情报源抓取里最常见的隐性失效原因。首次登录后把 cookie 写入文件Spider 开始时加载发现过期时重新触发一次登录请求再继续原有队列。需要注意把登录和正文抓取放在同一个 Spider 里会让重试状态复杂建议单独用一个小脚本维护登录态正文 Spider 只负责消费。3.2 动态渲染情报源的 Playwright 集成部分情报平台是前后端分离列表数据在 iframe 里异步加载直接抓 HTML 只能拿到空壳。这类场景用 scrapy-playwright 扩展接管下载器是一个可靠方案。配置如下# settings.py DOWNLOAD_HANDLERS { http: scrapy_playwright.handler.ScrapyPlaywrightDownloadHandler, https: scrapy_playwright.handler.ScrapyPlaywrightDownloadHandler, } TWISTED_REACTOR twisted.internet.asyncioreactor.AsyncioSelectorReactor PLAYWRIGHT_BROWSER_TYPE chromium PLAYWRIGHT_LAUNCH_OPTIONS {headless: True, args: [--disable-gpu]}然后在请求里通过 meta 告诉下载器用 Playwright 渲染# spiders/dynamic_source.py async def parse_iframe(self, response): page response.meta[playwright_page] await page.wait_for_selector(table.ioc-list) html await page.content() # 拿到渲染完成的 HTML 后再走解析 ...meta 里的 playwright 开关是精确到单请求的不会让整个 Spider 所有请求都挂浏览器进程对性能影响可控。playwright_include_page负责把浏览器页面句柄传给回调让异步逻辑能在页面上下文里执行。注意混用异步和同步解析时用 Twisted 的 asyncio 反应器是必须的否则事件循环无法驱动浏览器。3.3 限速与重试参数场景化调优Scrapy 默认参数是通用爬虫取向对威胁情报采集机偏激进。情报源大多请求量不大但采集连续性要求高建议优先放宽延时、降低并发配置项默认值威胁情报采集建议改动原因CONCURRENT_REQUESTS164~8降低瞬时并发避免触发频率封禁DOWNLOAD_DELAY02~5单请求间隔固定时间AUTOTHROTTLE_ENABLEDFalseTrue让采集机按源站响应速度自适应RETRY_TIMES23~5情报源偶尔发生 5xx重试即可恢复DOWNLOAD_TIMEOUT18060缩短单请求挂起时间一次性把这些参数设置死并不合理更好的做法是按源分组设置。Scrapy 支持在 Spider 内直接覆盖custom_settings对响应慢、偶尔返回 504 的历史漏洞库把 AUTOTHROTTLE_ENABLED 开启并把 RETRY_HTTP_CODES 扩到 502、503、504对厂商通告这类更新周期长的源把 DOWNLOAD_DELAY 调大减少空转请求。需要留意的坑是 AUTOTHROTTLE 与 DOWNLOAD_DELAY 同时开启时的行为差异。开了 AUTOTHROTTLE 后Scrapy 会动态计算延迟DOWNLOAD_DELAY 会被覆盖两者同时设并不冲突但直觉上会误以为加了固定延时。若想要确定性行为可以把 AUTOTHROTTLE_ENABLED 关掉只保留固定 DOWNLOAD_DELAY。4. 威胁情报处理管线字段模型、数据清洗与置信度评估4.1 情报指标字段模型与 Item 定义抓取只解决了数据到达的问题处理阶段决定数据是否可用。威胁情报的指标字段应该至少覆盖四类实体IP、域名、哈希值、漏洞编号在此基础上再附加来源、时间戳和描述上下文。Item 定义如下# items.py import scrapy class IoCItem(scrapy.Item): ioc_type scrapy.Field() # ip / domain / hash / cve / url ioc_value scrapy.Field() # 指标原文 confidence scrapy.Field() # 0.0 - 1.0表示这条指标可信程度 first_seen scrapy.Field() # 首次发现时间 last_seen scrapy.Field() # 最近一次活跃时间 source scrapy.Field() # 来源页面 URL tags scrapy.Field() # 标签如 c2/phishing/malware description scrapy.Field() # 自由文本上下文 fingerprint scrapy.Field() # 去重指纹字段定义本身不复杂真正的复杂度在于如何把不同情报源的文本映射到这套统一的字段模型里。一个漏洞详情页给出的可能是一大段描述文字而不是干净的 CVE 编号。这里需要转角在提取阶段做正则收敛。4.2 自由文本到标准 IoC 的清洗规则从 HTML 里抓到的字段往往是这样的几种形态带超链接的域名、混合了端口号的 IP、带前缀的哈希、夹在句式里的 CVE 编号。清洗函数的目标是把这些形态统一成标准值并用 None 表示不可用# pipelines.py import ipaddress import re CVE_RE re.compile(rCVE-\d{4}-\d{4,7}, re.IGNORECASE) DOMAIN_RE re.compile(r^[a-z0-9]([a-z0-9\-]{0,61}[a-z0-9])?(\.[a-z0-9]([a-z0-9\-]{0,61}[a-z0-9])?)$) def normalize_ioc(ioc_type: str, value: str): value (value or ).strip() if ioc_type ip: try: return str(ipaddress.ip_address(value.split(:)[0])) # 去掉端口 except ValueError: return None if ioc_type cve: m CVE_RE.search(value) return m.group(0).upper() if m else None if ioc_type domain: m DOMAIN_RE.match(value) return m.group(0) if m else None return value函数里用正则和 ipaddress 库做了三层拦截IP 先把端口剥掉再校验合法性CVE 从任意文本里找回编号域名只接受完整的多级域名格式。返回 None 的字段会走异常分支丢弃或标记为待人工核验不能直接进表否则下游安全设备解析时会产生误报。4.3 指纹去重与置信度加权威胁情报的去重不是简单判断字段相等。同一域名今天被博客提到明天又在漏洞库出现如果不做交叉归一30 天就会堆出几万条相似记录。前面的业务指纹方案再加一层指纹后处理管线会把同一条 IoC 的不同来源记录合并更新 last_seen 并提升置信度def merge_and_update(item, existing_record): item[confidence] max(item[confidence], existing_record[confidence]) item[first_seen] min(item[first_seen], existing_record[first_seen]) item[last_seen] max(item[last_seen], existing_record[last_seen]) return item置信度的初始值建议按来源类型给权重而不是全部默认 1.0。常见做法是给来源打一个基线分实操中比较实用的权重表长这样来源类型基线置信度说明厂商官方通告0.90发布前经过内部确认结构化漏洞库0.85有编号体系和收录流程中大型安全博客0.60分析质量参差但有引用链论坛/社交平台单帖0.30需要其他源交叉验证这个权重表的作用是给下游查询排序而不是直接丢弃低置信度记录。安全运营中低置信度情报也常包含有效线索建议保留但标注等有其他源提到时再自动升级。5. 威胁情报的结构化存储与增量更新设计5.1 存储选型关系型与文档型数据库取舍威胁情报存储选型要看查询场景。如果核心需求是按 IoC 字段精确匹配、按时间窗口扫最近活跃条目用 MySQL 或 PostgreSQL 完全够而且引入成本最低。如果还希望支持全文检索和聚合分析Elasticsearch 更合适但运维成本会明显上升。常见取舍如下存储方案优势劣势适合场景SQLite零部署、单文件并发写入弱单机原型MySQL/PostgreSQL事务可靠、索引成熟需要实例维护团队级情报库Elasticsearch全文检索、聚合快集群维护成本高大规模查询与分析ClickHouse写入吞吐高、压缩好事务能力弱历史流水存档见过不少团队一开始就上 ClickHouse结果查询接口还没写先被集群运维拖垮。威胁情报的日增量在千到万条级别MySQL 绰绰有余只有做长期统计报表时才值得把历史数据归档到分析型引擎。5.2 表结构与 UPSERT 更新策略一张单表即可支撑情报存取。核心表设计如下CREATE TABLE threat_ioc ( id BIGINT AUTO_INCREMENT PRIMARY KEY, ioc_type VARCHAR(16) NOT NULL, ioc_value VARCHAR(512) NOT NULL, confidence DECIMAL(3,2) DEFAULT 0.00, first_seen DATETIME NOT NULL, last_seen DATETIME NOT NULL, source VARCHAR(1024), tags JSON, description TEXT, fingerprint CHAR(64) NOT NULL, UNIQUE KEY uk_fingerprint (fingerprint) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;fingerprint 是唯一键这就是上一章提的指纹落地的位置。入库时用 INSERT ... ON DUPLICATE KEY UPDATE 实现幂等更新INSERT INTO threat_ioc (ioc_type, ioc_value, confidence, first_seen, last_seen, source, tags, description, fingerprint) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE last_seen VALUES(last_seen), confidence GREATEST(confidence, VALUES(confidence));这段 SQL 的逻辑是如果指纹已存在只更新 last_seen 到最新时间并把置信度取较大值不存在则整体插入。这样同一 IoC 多条来源不用先查一次再决定插入还是更新减少一次往返同时保证 first_seen 不会被新数据覆盖。5.3 Pipeline 与存储层的衔接将清洗结果写入 MySQL 的 Pipeline核心是开连接、逐条入库、关闭连接# pipelines.py import json import pymysql class MysqlPipeline: def open_spider(self, spider): self.conn pymysql.connect( host127.0.0.1, userthreat, password***, databasethreat_intel, charsetutf8mb4, ) self.cursor self.conn.cursor() def process_item(self, item, spider): self.cursor.execute( INSERT_SQL, (item[ioc_type], item[ioc_value], item[confidence], item[first_seen], item[last_seen], item[source], json.dumps(item[tags]), item[description], item[fingerprint]), ) self.conn.commit() return item def close_spider(self, spider): self.cursor.close() self.conn.close()这个写法有个性能隐患每行都 commit。规模上涨后应改为在 open_spider 关闭自动提交攒够 50 行或每隔 5 秒提交一次。批量提交能显著降低磁盘同步压力也不影响数据完整性顶多出现最后一批未提交的场景。另外Pipeline 在 settings.py 里ITEM_PIPELINES中的声明顺序就是执行顺序。清洗与入库如果分开写不要把入库放在清洗之前否则脏数据先进库再处理就晚了。6. 威胁情报抓取处理系统的验证手段与调优技巧6.1 采集质量的一次性核验上线后第一件事不是看日志有没有报错而是验证抓下来的字段有没有悬空值。一条来自论坛的情报如果 description 字段全空、confidence 全是 0.3说明清洗流程把太多内容挡掉了。用一条 SQL 就能看出来SELECT ioc_type, COUNT(*) AS cnt, SUM(confidence 0.5) AS low_conf_cnt, SUM(description IS NULL OR description ) AS empty_desc_cnt FROM threat_ioc GROUP BY ioc_type;low_conf_cnt 占比长期超过 60%说明来源权重表该调整了empty_desc_cnt 高说明 Spider 的解析选择器没对上真实页面结构。顺带可以用 Scrapy 运行日志核对 item_scraped_count 是否符合预期比如一次调度期望抓 1200 条日志却只统计到 400 条多半是哪个列表页被反爬拦截了。6.2 定时调度与断点续爬的配合威胁情报是周期性的增量采集用 cron 比常驻进程更稳妥。常驻进程一旦脚本异常整个采集服务就停了cron 则每次独立启动挂了下次还会再拉起来15 */2 * * * cd /opt/threat_intel /usr/local/bin/scrapy crawl vuln_feed -s LOG_LEVELINFO -s JOBDIRscheduler_state/vuln_feed /var/log/threat/vuln_feed.log 21这个 cron 每两小时的 15 分启动一次。JOBDIR 让上次中断的请求队列直接继续不用全量重抓LOG_LEVELINFO 只记录关键事件日志文件能控制在一个可以接受的大小。线上跑一段时间后用grep -c item_scraped_count /var/log/threat/vuln_feed.log检查每天的调度是否都在产出新记录一旦为零就说明某个源的结构变了。6.3 快速验证解析规则时的补盲技巧调整选择器时反复重启爬虫很浪费时间。我会先在 Scrapy shell 里用单页面验证选择器命中情况确认无误再整体跑。定位动态渲染页时shell 里加上scrapy shell --nolog配合 Playwright 的 response 对象能很快确认 iframe 里到底有没有目标表格。抓到新情报后用 Python 直接对 fingerprint 字段做一轮随机抽样检查入库样本和前几次调度的指纹重复比例重复率若高于 30%大多不是情报重复而是去重指纹拼的字段在不同源之间格式不一致导致需要对 normalize_ioc 涉及的边界条件做一次回归。本文还有配套的精品资源点击获取 基于Scrapy的威胁情报抓取处理系统：从情报源到可查指标
简介一份基于Scrapy框架的威胁情报抓取与处理系统的毕业设计文档面向网络安全、Python爬虫及威胁情报方向的学生与研究者。内容围绕如何利用Scrapy爬虫高效采集开源威胁网站与博客数据并完成解析、入库与展示涉及知识图谱、APT知识图谱、Flask_adminnginx、pyecharts等关键技术覆盖系统分析与设计、爬虫模块、数据解析模块、数据展示模块等章节适合作为相关课题设计、论文写作或项目开发的参考资料。资源为单个docx文件大小2.58MB共1份文档结构完整包含中英文摘要、目录及正文内容。已有194人学习下载适用性较好。1. 基于Scrapy框架的威胁情报抓取处理系统从情报源到可查指标威胁情报的价值本质上是时间的函数。攻击者更换一次 C2 域名、上线一个新的钓鱼页面留给防御方反应的时间往往按小时计算。靠人工巡看安全公众号、抓包群和厂商公告的传统采集链路字段格式不统一落地到检测规则时还要先手工清洗一遍这段周转本身就消耗了宝贵的处置窗口。基于Scrapy框架的威胁情报抓取以及处理系统要解决的正是从情报源到可查询数据资产的工程化压缩让 Scrapy 调度对安全博客、漏洞库、代码托管平台和厂商通告的分批抓取用 Item Pipeline 完成字段抽取、格式标准化和置信度去重最终写入可检索存储。这套系统适合安全分析团队和蓝队工程师他们要的不是一个简单爬虫 demo而是能把采集、清洗、入库固化为日常作业的完整链路。下文从工程搭建切入逐步展开反爬应对、字段处理和验证方法。2. Scrapy框架下威胁情报抓取系统的架构设计与工程搭建2.1 威胁情报源的类型划分与Scrapy适配性评估先说选型边界。不是所有威胁情报源都适合用 Scrapy 去抓有些直接提供 JSON API 或历史下载包用 requests 脚本反而更省事。决定用 Scrapy 的标准是信息源本身是 HTML 页面流、有翻页或站内搜索语义、需要做频控和断点续爬。符合这个条件的典型情报源有下面几类情报源类型典型表现形态抓取方式Scrapy适配度RSS/Atom 订阅厂商通告、安全博客更新XML 字段直接抽取高漏洞库NVD、CNVD 的列表详情页分页列表 详情页高代码托管仓库IoC 列表更新、README 变更文件接口或仓库提交 API中商务情报平台混合渲染页面通用爬虫或动态渲染中论坛/社区板块分页帖子通用爬虫 登录态中低选用 Scrapy 的价值在这些场景下才成立列表页与详情页结构稳定、增量以天或小时为单位变化、请求量不大但条目众多时手工维护会失控。尤其是一类常见需求比如把公众号相关文章的自动搜索监控抓取也纳入情报来源虽然公众号网页是动态渲染但通过 Scrapy 配合浏览器驱动一样能落到同一套 Pipeline 里。2.2 爬虫工程目录与Spider骨架实现创建项目属于固定动作但目录职责最好从第一天就分清楚scrapy startproject threat_intel cd threat_intel scrapy genspider -t crawl rss_feed feed.example.comstartproject生成项目骨架genspider -t crawl用 CrawlSpider 模板创建一个名为 rss_feed 的爬虫。CrawlSpider 自带链接提取规则适合列表页稳定的序列型情报源省掉手写 parse 分发。threat_intel/ ├── items.py # IoC 字段容器 ├── middlewares.py # 请求头切换、cookie 刷新 ├── pipelines.py # 清洗、去重、入库 ├── settings.py # 并发、限速、下载延迟配置 ├── spiders/ │ ├── rss_feed.py # RSS 源适配 │ ├── blog_crawler.py # 安全博客页面抓取 │ └── vuln_feed.py # 漏洞库增量抓取 └── requirements.txt如果把多个来源塞进同一个 Spider后续改动一处字段就会牵连其他源反而不利于维护。常见做法是每一类结构相近的情报源独立成 Spider在 settings 里用SPIDER_LOADER_WARN_ONLY控制加载告警避免无关源互相影响。提示Spider 文件命名建议直接使用情报源的域名或业务名爬虫日志里按爬虫名过滤时能一眼定位是哪条链路出了问题。下面这个样例是漏洞库爬虫跑的是列表页到详情页的两段跳转# spiders/vuln_feed.py from urllib.parse import urljoin from datetime import date import scrapy from scrapy.loader import ItemLoader from threat_intel.items import IoCItem class VulnFeedSpider(scrapy.Spider): name vuln_feed allowed_domains [nvd.nist.gov] start_urls [https://nvd.nist.gov/vuln/data-feeds] def parse(self, response): # 只挑列表页里 CVE 开头的条目过滤掉导航和其他噪声 for row in response.css(table.news-table tr): title row.css(a::text).get() href row.css(a::attr(href)).get() if href and title and CVE in title: yield scrapy.Request( urljoin(response.url, href), callbackself.parse_detail, meta{title: title.strip()}, ) def parse_detail(self, response): # 详情页里摘 CVE 描述不需要精确到全表 loader ItemLoader(itemIoCItem(), responseresponse) loader.add_value(ioc_type, cve) loader.add_value(ioc_value, response.url.split(/)[-1]) loader.add_css(description, #vulnDetailTableView::text) loader.add_value(source, response.url) loader.add_value(last_seen, date.today().isoformat()) yield loader.load_item()这段代码刻意的点有两个一是在 parse 里先用标题过滤再决定是否发详情页请求能少一半无意义的页面抓取二是用 ItemLoader 而不是手写字典为后续 Pipeline 统一处理留了接缝。meta 传入的 title 能在详情页拿不到字段时用作兜底不至于因为一个字段丢失丢掉整条记录。2.3 请求调度、指纹去重与增量采集Scrapy 内置的 RFPDupeFilter 默认按请求 URL 生成指纹对纯 GET 的情报源够用但同一个漏洞可能从两个情报源各抓一次需要在入库前去重。更可靠的做法是建立业务指纹# pipelines.py 中自定义业务指纹示例 import hashlib def build_fingerprint(payload): raw |.join([ payload.get(ioc_type, ), payload.get(ioc_value, ), payload.get(source_url, ), ]) return hashlib.sha256(raw.encode(utf-8)).hexdigest()指纹把类型、值和来源拼在一起做哈希。简洁设计里可以去掉source_url让同一指标只保留一条记录但威胁情报场景中来源信息是重要审计线索一般会保留。调度侧做增量采集时可以在 settings 里开启 JOBDIR# settings.py DUPEFILTER_CLASS scrapy.dupefilters.RFPDupeFilter JOBDIR scheduler_state/vuln_feed SCHEDULER_DEBUG FalseJOBDIR 的作用是把未完成的请求队列持久化到磁盘中断重启后能接着抓不会重复请求已完成页面。SCHEDULER_DEBUG 平时建议置 False不然每次调度都会输出大量调试日志反而干扰对实际采集状态的判断。3. 威胁情报抓取的中间件定制与反爬应对策略3.1 请求头伪装与Cookie生命周期刷新威胁情报源大多有基础的访问频率限制少数还会校验 User-Agent 特征。抓取端需要做的是让采集机的请求看起来像普通浏览器行为而不是一个固定不变的 curl 头。常见做法是维护一个 UA 池在中间件中随机挑选并注入请求头# middlewares.py import random USER_AGENT_POOL [ Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0 Safari/537.36, Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0 Safari/537.36, ] class RandomUAProcessMiddleware: 在下载前为每个请求注入随机的 UA def process_request(self, request, spider): request.headers[User-Agent] random.choice(USER_AGENT_POOL) return None中间件在两个方向上有用process_request在请求发出前改头process_response在拿到响应后检查状态码。如果发现 403 或者页面要求登录可以在process_response里直接丢弃响应并返回一个新请求避免污染后续解析。Cookie 过期是情报源抓取里最常见的隐性失效原因。首次登录后把 cookie 写入文件Spider 开始时加载发现过期时重新触发一次登录请求再继续原有队列。需要注意把登录和正文抓取放在同一个 Spider 里会让重试状态复杂建议单独用一个小脚本维护登录态正文 Spider 只负责消费。3.2 动态渲染情报源的 Playwright 集成部分情报平台是前后端分离列表数据在 iframe 里异步加载直接抓 HTML 只能拿到空壳。这类场景用 scrapy-playwright 扩展接管下载器是一个可靠方案。配置如下# settings.py DOWNLOAD_HANDLERS { http: scrapy_playwright.handler.ScrapyPlaywrightDownloadHandler, https: scrapy_playwright.handler.ScrapyPlaywrightDownloadHandler, } TWISTED_REACTOR twisted.internet.asyncioreactor.AsyncioSelectorReactor PLAYWRIGHT_BROWSER_TYPE chromium PLAYWRIGHT_LAUNCH_OPTIONS {headless: True, args: [--disable-gpu]}然后在请求里通过 meta 告诉下载器用 Playwright 渲染# spiders/dynamic_source.py async def parse_iframe(self, response): page response.meta[playwright_page] await page.wait_for_selector(table.ioc-list) html await page.content() # 拿到渲染完成的 HTML 后再走解析 ...meta 里的 playwright 开关是精确到单请求的不会让整个 Spider 所有请求都挂浏览器进程对性能影响可控。playwright_include_page负责把浏览器页面句柄传给回调让异步逻辑能在页面上下文里执行。注意混用异步和同步解析时用 Twisted 的 asyncio 反应器是必须的否则事件循环无法驱动浏览器。3.3 限速与重试参数场景化调优Scrapy 默认参数是通用爬虫取向对威胁情报采集机偏激进。情报源大多请求量不大但采集连续性要求高建议优先放宽延时、降低并发配置项默认值威胁情报采集建议改动原因CONCURRENT_REQUESTS164~8降低瞬时并发避免触发频率封禁DOWNLOAD_DELAY02~5单请求间隔固定时间AUTOTHROTTLE_ENABLEDFalseTrue让采集机按源站响应速度自适应RETRY_TIMES23~5情报源偶尔发生 5xx重试即可恢复DOWNLOAD_TIMEOUT18060缩短单请求挂起时间一次性把这些参数设置死并不合理更好的做法是按源分组设置。Scrapy 支持在 Spider 内直接覆盖custom_settings对响应慢、偶尔返回 504 的历史漏洞库把 AUTOTHROTTLE_ENABLED 开启并把 RETRY_HTTP_CODES 扩到 502、503、504对厂商通告这类更新周期长的源把 DOWNLOAD_DELAY 调大减少空转请求。需要留意的坑是 AUTOTHROTTLE 与 DOWNLOAD_DELAY 同时开启时的行为差异。开了 AUTOTHROTTLE 后Scrapy 会动态计算延迟DOWNLOAD_DELAY 会被覆盖两者同时设并不冲突但直觉上会误以为加了固定延时。若想要确定性行为可以把 AUTOTHROTTLE_ENABLED 关掉只保留固定 DOWNLOAD_DELAY。4. 威胁情报处理管线字段模型、数据清洗与置信度评估4.1 情报指标字段模型与 Item 定义抓取只解决了数据到达的问题处理阶段决定数据是否可用。威胁情报的指标字段应该至少覆盖四类实体IP、域名、哈希值、漏洞编号在此基础上再附加来源、时间戳和描述上下文。Item 定义如下# items.py import scrapy class IoCItem(scrapy.Item): ioc_type scrapy.Field() # ip / domain / hash / cve / url ioc_value scrapy.Field() # 指标原文 confidence scrapy.Field() # 0.0 - 1.0表示这条指标可信程度 first_seen scrapy.Field() # 首次发现时间 last_seen scrapy.Field() # 最近一次活跃时间 source scrapy.Field() # 来源页面 URL tags scrapy.Field() # 标签如 c2/phishing/malware description scrapy.Field() # 自由文本上下文 fingerprint scrapy.Field() # 去重指纹字段定义本身不复杂真正的复杂度在于如何把不同情报源的文本映射到这套统一的字段模型里。一个漏洞详情页给出的可能是一大段描述文字而不是干净的 CVE 编号。这里需要转角在提取阶段做正则收敛。4.2 自由文本到标准 IoC 的清洗规则从 HTML 里抓到的字段往往是这样的几种形态带超链接的域名、混合了端口号的 IP、带前缀的哈希、夹在句式里的 CVE 编号。清洗函数的目标是把这些形态统一成标准值并用 None 表示不可用# pipelines.py import ipaddress import re CVE_RE re.compile(rCVE-\d{4}-\d{4,7}, re.IGNORECASE) DOMAIN_RE re.compile(r^[a-z0-9]([a-z0-9\-]{0,61}[a-z0-9])?(\.[a-z0-9]([a-z0-9\-]{0,61}[a-z0-9])?)$) def normalize_ioc(ioc_type: str, value: str): value (value or ).strip() if ioc_type ip: try: return str(ipaddress.ip_address(value.split(:)[0])) # 去掉端口 except ValueError: return None if ioc_type cve: m CVE_RE.search(value) return m.group(0).upper() if m else None if ioc_type domain: m DOMAIN_RE.match(value) return m.group(0) if m else None return value函数里用正则和 ipaddress 库做了三层拦截IP 先把端口剥掉再校验合法性CVE 从任意文本里找回编号域名只接受完整的多级域名格式。返回 None 的字段会走异常分支丢弃或标记为待人工核验不能直接进表否则下游安全设备解析时会产生误报。4.3 指纹去重与置信度加权威胁情报的去重不是简单判断字段相等。同一域名今天被博客提到明天又在漏洞库出现如果不做交叉归一30 天就会堆出几万条相似记录。前面的业务指纹方案再加一层指纹后处理管线会把同一条 IoC 的不同来源记录合并更新 last_seen 并提升置信度def merge_and_update(item, existing_record): item[confidence] max(item[confidence], existing_record[confidence]) item[first_seen] min(item[first_seen], existing_record[first_seen]) item[last_seen] max(item[last_seen], existing_record[last_seen]) return item置信度的初始值建议按来源类型给权重而不是全部默认 1.0。常见做法是给来源打一个基线分实操中比较实用的权重表长这样来源类型基线置信度说明厂商官方通告0.90发布前经过内部确认结构化漏洞库0.85有编号体系和收录流程中大型安全博客0.60分析质量参差但有引用链论坛/社交平台单帖0.30需要其他源交叉验证这个权重表的作用是给下游查询排序而不是直接丢弃低置信度记录。安全运营中低置信度情报也常包含有效线索建议保留但标注等有其他源提到时再自动升级。5. 威胁情报的结构化存储与增量更新设计5.1 存储选型关系型与文档型数据库取舍威胁情报存储选型要看查询场景。如果核心需求是按 IoC 字段精确匹配、按时间窗口扫最近活跃条目用 MySQL 或 PostgreSQL 完全够而且引入成本最低。如果还希望支持全文检索和聚合分析Elasticsearch 更合适但运维成本会明显上升。常见取舍如下存储方案优势劣势适合场景SQLite零部署、单文件并发写入弱单机原型MySQL/PostgreSQL事务可靠、索引成熟需要实例维护团队级情报库Elasticsearch全文检索、聚合快集群维护成本高大规模查询与分析ClickHouse写入吞吐高、压缩好事务能力弱历史流水存档见过不少团队一开始就上 ClickHouse结果查询接口还没写先被集群运维拖垮。威胁情报的日增量在千到万条级别MySQL 绰绰有余只有做长期统计报表时才值得把历史数据归档到分析型引擎。5.2 表结构与 UPSERT 更新策略一张单表即可支撑情报存取。核心表设计如下CREATE TABLE threat_ioc ( id BIGINT AUTO_INCREMENT PRIMARY KEY, ioc_type VARCHAR(16) NOT NULL, ioc_value VARCHAR(512) NOT NULL, confidence DECIMAL(3,2) DEFAULT 0.00, first_seen DATETIME NOT NULL, last_seen DATETIME NOT NULL, source VARCHAR(1024), tags JSON, description TEXT, fingerprint CHAR(64) NOT NULL, UNIQUE KEY uk_fingerprint (fingerprint) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;fingerprint 是唯一键这就是上一章提的指纹落地的位置。入库时用 INSERT ... ON DUPLICATE KEY UPDATE 实现幂等更新INSERT INTO threat_ioc (ioc_type, ioc_value, confidence, first_seen, last_seen, source, tags, description, fingerprint) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE last_seen VALUES(last_seen), confidence GREATEST(confidence, VALUES(confidence));这段 SQL 的逻辑是如果指纹已存在只更新 last_seen 到最新时间并把置信度取较大值不存在则整体插入。这样同一 IoC 多条来源不用先查一次再决定插入还是更新减少一次往返同时保证 first_seen 不会被新数据覆盖。5.3 Pipeline 与存储层的衔接将清洗结果写入 MySQL 的 Pipeline核心是开连接、逐条入库、关闭连接# pipelines.py import json import pymysql class MysqlPipeline: def open_spider(self, spider): self.conn pymysql.connect( host127.0.0.1, userthreat, password***, databasethreat_intel, charsetutf8mb4, ) self.cursor self.conn.cursor() def process_item(self, item, spider): self.cursor.execute( INSERT_SQL, (item[ioc_type], item[ioc_value], item[confidence], item[first_seen], item[last_seen], item[source], json.dumps(item[tags]), item[description], item[fingerprint]), ) self.conn.commit() return item def close_spider(self, spider): self.cursor.close() self.conn.close()这个写法有个性能隐患每行都 commit。规模上涨后应改为在 open_spider 关闭自动提交攒够 50 行或每隔 5 秒提交一次。批量提交能显著降低磁盘同步压力也不影响数据完整性顶多出现最后一批未提交的场景。另外Pipeline 在 settings.py 里ITEM_PIPELINES中的声明顺序就是执行顺序。清洗与入库如果分开写不要把入库放在清洗之前否则脏数据先进库再处理就晚了。6. 威胁情报抓取处理系统的验证手段与调优技巧6.1 采集质量的一次性核验上线后第一件事不是看日志有没有报错而是验证抓下来的字段有没有悬空值。一条来自论坛的情报如果 description 字段全空、confidence 全是 0.3说明清洗流程把太多内容挡掉了。用一条 SQL 就能看出来SELECT ioc_type, COUNT(*) AS cnt, SUM(confidence 0.5) AS low_conf_cnt, SUM(description IS NULL OR description ) AS empty_desc_cnt FROM threat_ioc GROUP BY ioc_type;low_conf_cnt 占比长期超过 60%说明来源权重表该调整了empty_desc_cnt 高说明 Spider 的解析选择器没对上真实页面结构。顺带可以用 Scrapy 运行日志核对 item_scraped_count 是否符合预期比如一次调度期望抓 1200 条日志却只统计到 400 条多半是哪个列表页被反爬拦截了。6.2 定时调度与断点续爬的配合威胁情报是周期性的增量采集用 cron 比常驻进程更稳妥。常驻进程一旦脚本异常整个采集服务就停了cron 则每次独立启动挂了下次还会再拉起来15 */2 * * * cd /opt/threat_intel /usr/local/bin/scrapy crawl vuln_feed -s LOG_LEVELINFO -s JOBDIRscheduler_state/vuln_feed /var/log/threat/vuln_feed.log 21这个 cron 每两小时的 15 分启动一次。JOBDIR 让上次中断的请求队列直接继续不用全量重抓LOG_LEVELINFO 只记录关键事件日志文件能控制在一个可以接受的大小。线上跑一段时间后用grep -c item_scraped_count /var/log/threat/vuln_feed.log检查每天的调度是否都在产出新记录一旦为零就说明某个源的结构变了。6.3 快速验证解析规则时的补盲技巧调整选择器时反复重启爬虫很浪费时间。我会先在 Scrapy shell 里用单页面验证选择器命中情况确认无误再整体跑。定位动态渲染页时shell 里加上scrapy shell --nolog配合 Playwright 的 response 对象能很快确认 iframe 里到底有没有目标表格。抓到新情报后用 Python 直接对 fingerprint 字段做一轮随机抽样检查入库样本和前几次调度的指纹重复比例重复率若高于 30%大多不是情报重复而是去重指纹拼的字段在不同源之间格式不一致导致需要对 normalize_ioc 涉及的边界条件做一次回归。本文还有配套的精品资源点击获取