尧图网站建设 
简介这是一套基于Python的威胁情报自动播报与聚合工具面向安全运维人员、威胁情报分析者及对漏洞动态有持续跟踪需求的开发者。项目整合360、奇安信、红后、绿盟、安全客、斗象等公开情报源自动爬取CVE等威胁信息并通过邮件推送、页面TOP10展示与本地归档三种方式输出情报。资源共70个文件以Python脚本25个、XML配置、DAT缓存数据、HTML页面模板为主压缩包仅738KB代码结构清晰包含爬虫、数据处理、消息通知、SQL建库脚本及GitHub Actions工作流配置便于二次开发和自定义部署。已有191人学习下载。借助本项目可快速搭建一套免服务器运维的情报订阅系统理解多源情报采集、去重归档与多渠道播报的完整实现思路适合个人或小团队持续跟踪外部威胁动态。1. 威胁情报不是爬虫项目先想清楚你要汇总什么做威胁情报threat-intelligence踩过最大的坑是把它当成爬虫项目来启动。真正跑起来才发现80% 的工作量不在抓取而在清洗、去重、验证和关联。公开渠道的安全通告、漏洞库、恶意 IP 列表每天产生上千条数据但其中大量是重复的还有不少是误报和过时信息。这套资源的核心不是给你一堆采集脚本而是把「采集 → 解析 → 存储 → 查询 → 验证」的完整链路串起来用 Python 实现一个能落地的 IOC入侵指标汇总分析流程。适合安全工程师、运维人员以及需要做内部威胁情报平台选型但又不想一开始就上重型商业方案的人。你跟着做出来的东西可以对接已有的告警系统也可以当成一个独立的情报查询工具用。2. 整体架构与数据模型SQLite 加事件表不上一堆重组件2.1 为什么不用 Elasticsearch 或关系型大库很多人一上来就考虑 Elasticsearch理由是情报数据要全文检索。但实际场景里一个中小团队每天处理的情报量在几千到几万条级别SQLite 完全扛得住而且部署成本几乎为零。Elasticsearch 的运维负担、内存占用、数据导入导出的麻烦在这个量级上不划算。我一般建议先跑通 SQLite 版本等真的出现性能瓶颈再迁移到时候表结构和查询逻辑大体不用改。数据模型的设计上我见过不少反例有人把 IOC 塞进一个超大宽表结果查一次关联要 JOIN 五六个字段效率极低。核心思路应该是一张事件主表incidents加一张指标明细表iocs两表之间通过事件 ID 关联。主表存事件元信息明细表存 IP、域名、哈希、URL 等具体指标。这样设计的好处是一个事件可能包含多个 IOC 类型而你查询某个 IP 时只需要在明细表里做索引扫描速度快逻辑也清晰。2.2 表结构设计字段别贪多够用就行给出一份可以直接照抄的建表 SQL这是整套资源里所有脚本的基础。CREATE TABLE incidents ( id INTEGER PRIMARY KEY AUTOINCREMENT, source TEXT NOT NULL, title TEXT, description TEXT, published_at TEXT, fetched_at TEXT DEFAULT CURRENT_TIMESTAMP, severity INTEGER DEFAULT 2, status TEXT DEFAULT new ); CREATE TABLE iocs ( id INTEGER PRIMARY KEY AUTOINCREMENT, incident_id INTEGER NOT NULL, ioc_type TEXT NOT NULL, value TEXT NOT NULL, confidence REAL DEFAULT 0.5, first_seen TEXT, last_seen TEXT, FOREIGN KEY (incident_id) REFERENCES incidents(id) ); CREATE INDEX idx_iocs_type_value ON iocs(ioc_type, value); CREATE INDEX idx_iocs_value ON iocs(value);这里几个字段需要解释一下。severity用整数不用字符串0 到 3 分别对应低危到危急排序和过滤都方便confidence是浮点数0 到 1 之间表示这条 IOC 的可信度来自厂商标注或者你自己的验证逻辑status字段标记处理状态new、investigating、confirmed、false_positive四种。索引方面iocs表必须建value的单列索引因为这是查询频率最高的路径。2.3 模块划分采集、解析、入库三个脚本各自独立这套资源的代码包按功能拆成了四个模块collector.py负责抓取parser.py负责解析storage.py负责入库query.py负责查询导出。分开写比一个大文件实用得多——采集源出问题只需要改 collector解析规则变了只动 parser互不干扰。3. 采集层实现多源抓取与频率控制3.1 公开安全通告源的通用采集模板常见的威胁情报源大致分两类一类是结构化 API返回 JSON 或 CSV另一类是安全公告页面返回 HTML 或 RSS。第一类写起来简单第二类才是真正麻烦的地方。给一个处理 RSS 通告源的模板适配大多数安全厂商的公告系统。import requests import xml.etree.ElementTree as ET from datetime import datetime, timezone FEED_URLS { example_cert: https://example-cert.org/feeds/advisories, example_vendor: https://example-vendor.com/security/bulletin.rss, } def fetch_feed(name, url, timeout15): headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) threat-intel-collector/1.0 } try: resp requests.get(url, headersheaders, timeouttimeout) resp.raise_for_status() return resp.content except requests.exceptions.Timeout: print(f[{name}] timeout after {timeout}s, skip this round) return None except requests.exceptions.RequestException as e: print(f[{name}] request failed: {e}) return None def parse_rss(content): root ET.fromstring(content) items [] for item in root.iter(item): entry { title: item.findtext(title, ).strip(), link: item.findtext(link, ).strip(), description: item.findtext(description, ).strip(), pub_date: item.findtext(pubDate, ).strip(), } items.append(entry) return items两个参数值得注意。timeout15是请求超时阈值设短了网络抖动容易失败设长了采集任务会被卡住15 秒是我试下来比较平衡的值。User-Agent必须设置成真实浏览器样式很多情报源会拒绝裸的 Python 请求。RSS 解析用xml.etree.ElementTree而不是正则因为 XML 实体的转义规则用正则容易漏尤其 description 字段里经常带 HTML 标签。3.2 增量采集与去重记住上次的游标位置全量抓取每次都从头开始既浪费带宽又会产生大量重复数据。正确做法是记录每个源上次成功获取到的最新条目时间或 ID下次增量拉取。下面这段代码展示了如何利用fetched_at做增量判断import sqlite3 from datetime import datetime, timedelta def get_last_fetch_time(conn, source): row conn.execute( SELECT MAX(fetched_at) FROM incidents WHERE source ?, (source,) ).fetchone() if row and row[0]: return datetime.fromisoformat(row[0]) return datetime.now(timezone.utc) - timedelta(days7) def is_duplicate(conn, source, title): row conn.execute( SELECT id FROM incidents WHERE source ? AND title ? LIMIT 1, (source, title), ).fetchone() return row is not Noneget_last_fetch_time的作用是给采集器一个时间窗口只处理这个时间之后发布的条目。is_duplicate用source title作为自然键判断是否已经入库。这里有个细节不要用link字段做唯一性判断因为有些源的链接会带跟踪参数同一个通告可能生成多个不同 URL导致重复入库。3.3 抓取失败的补偿策略采集任务跑起来之后你会发现情报源不稳定是常态尤其是境外源经常超时或者返回 502。常见的做法是连续失败三次才告警单次失败只记录日志下一轮继续重试。这个逻辑必须写进主循环否则任何一次抓取异常都会白屏崩溃。import time FAIL_THRESHOLD 3 def run_collector(conn, source, url, fail_count0): content fetch_feed(source, url) if content is None: fail_count 1 if fail_count FAIL_THRESHOLD: print(f[{source}] failed {FAIL_THRESHOLD} times, manual check required) fail_count 0 return fail_count items parse_rss(content) for item in items: if not is_duplicate(conn, source, item[title]): insert_incident(conn, source, item) return 0这段代码的核心是fail_count的累积与重置机制。连续失败三次说明源可能挂了或者改版了需要人工介入单次失败不处理等下次调度。采集频率控制在每 30 到 60 分钟一轮比较合适低于 30 分钟容易被封 IP高于 60 分钟情报时效性会打折。4. 情报解析与 IOC 提取正则之外还要有启发式规则4.1 从通告正文里抠出四类 IOC通告的正文通常是一段自然语言描述夹杂着 IP 地址、域名、哈希值、URL。第一版我直接用正则提取效果很差URL 里包含的域名会被误识别为独立域名文件哈希和软件版本号混在一起也分不清。后来改成「正则提取候选值 启发式规则分类」的两步走思路。import re IP_RE re.compile(r(?!\d)(?:\d{1,3}\.){3}\d{1,3}(?!\d)) HASH_RE re.compile(r\b(?:[a-fA-F0-9]{32}|[a-fA-F0-9]{40}|[a-fA-F0-9]{64})\b) DOMAIN_RE re.compile(r\b(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)[a-z]{2,}\b, re.IGNORECASE) URL_RE re.compile(rhttps?://[^\s\]) def extract_iocs(text): iocs [] for m in URL_RE.finditer(text): iocs.append((url, m.group(0))) for m in HASH_RE.finditer(text): h m.group(0).lower() if len(h) 32: iocs.append((md5, h)) elif len(h) 40: iocs.append((sha1, h)) else: iocs.append((sha256, h)) for m in DOMAIN_RE.finditer(text): domain m.group(0).lower() if not any(domain in u for u, _ in iocs if _ url): iocs.append((domain, domain)) for m in IP_RE.finditer(text): ip m.group(0) if is_valid_ip(ip): iocs.append((ip, ip)) return iocs这里有个顺序问题必须先提取 URL再从结果里排除域名否则 URL 里的域名会被当成独立域名重复计入。HASH_RE里三个长度分别对应 MD5、SHA1、SHA256提取后统一转成小写因为后面做关联分析时大小写不一致会导致 JOIN 失败。is_valid_ip是一个校验函数过滤掉类似999.1.1.1这种正则能匹配但不合法的值。4.2 启发式过滤去掉明显误报正则提取只是第一步误报依然很多。比如正文里提到的示例 IP192.0.2.1RFC 文档保留地址段、内部 IP10.0.0.1、以及作者举例用的假域名example.com。这些必须在解析阶段干掉否则存进去之后每次查询都会被干扰。RESERVED_IP_RANGES [10., 192.168., 127., 0., 172.16., 172.17., 172.18., 172.19., 172.2, 172.3] RESERVED_DOMAINS [example.com, example.org, example.net, test.com, localhost] def filter_false_positives(iocs): filtered [] for ioc_type, value in iocs: if ioc_type ip: if any(value.startswith(prefix) for prefix in RESERVED_IP_RANGES): continue elif ioc_type domain: if any(value.endswith(domain) for domain in RESERVED_DOMAINS): continue filtered.append((ioc_type, value)) return filtered这段过滤逻辑看着简单但实际很有用。地址段前缀的判断用了startswith而不是精确匹配是因为172.16.0.1到172.31.255.255这些私网地址段范围较宽用前缀匹配是业内默认做法。注意172.2和172.3是为了覆盖172.20.x.x到172.30.x.x的部分地址段不算精确但够用。4.3 置信度打分不是所有情报可信度都一样厂商通告里的 IOC 可信度高第三方汇总源的可信度低。这套资源里把置信度拆成两个维度来源权重和指标类型权重。厂商官方源给 0.9第三方聚合源给 0.6IP 类指标因为动态变化频繁打 0.1 的降权系数域名类降权 0.2。SOURCE_WEIGHT {vendor_official: 0.9, aggregator: 0.6, community: 0.4} TYPE_WEIGHT {sha256: 1.0, sha1: 0.9, md5: 0.8, domain: 0.7, url: 0.6, ip: 0.4} def compute_confidence(source_type, ioc_type): return round(SOURCE_WEIGHT.get(source_type, 0.5) * TYPE_WEIGHT.get(ioc_type, 0.5), 2)哈希值的权重比 IP 高是有原因的攻击者换 IP 成本极低换一个文件哈希成本高得多所以哈希的稳定性更强。置信度最终存到iocs表的confidence字段里查询的时候可以按这个字段排序或过滤。5. 踩坑与排查存活性验证、编码脏数据、增量逻辑失效5.1 存活性验证情报库里 70% 的 IP 可能已经失效现象从情报库里随机抽出 100 条 IP 数据做检测能连上的不到三成大量情报来源已经是死数据。原因IP 和域名的存活性周期很短攻击基础设施经常变国内 IP 更是说封就封。采集入库时不验证结果只能当历史档案用。解决入库后 24 小时内必须做一轮存活性探测。IP 用 TCP 连接测试常见端口域名做 DNS 解析确认是否仍然指向有效地址。注意事项是探测频率不能太高否则情报源会把你封掉。5.2 编码与脏数据公告页面编码不一致导致解析挂掉现象某个情报源昨天还能正常解析今天突然提取不到任何 IOC检查日志发现解析器报了一堆乱码错误。原因不同源用的编码不一样有的 GBK有的 UTF-8还有的页面声称是 UTF-8 实际是 Latin-1。requests 的resp.text默认使用响应头声称的编码遇到错误的声明就会解出乱码。解决抓取后强制用resp.content配合chardet重新检测编码不要直接用resp.text。代码库里已经内置了这个逻辑你只需要在改源的时候注意别把这个步骤跳过去。5.3 增量逻辑失效漏抓两周数据现象某天发现incidents表里有个源的数据停留在两周前但采集任务一直显示成功执行。原因get_last_fetch_time用的是MAX(fetched_at)但fetched_at是入库时间不是信息发布的时间。如果某次批量导入历史数据fetched_at会被整体刷新导致增量窗口错乱。解决改用published_at字段作为增量游标这个字段保存的是情报源原始发布时间不会因为入库操作而变化。如果某个源不提供published_at就用link指纹去重两套逻辑互相兜底。5.4 业务误报情报平台命中≠安全事件现象情报平台频繁告警说内网 IP 访问了恶意域名但核查后发现是某个业务系统的合法调用。原因很多云服务商的 IP 段被误报进威胁情报库比如共享 IP 上有人跑过恶意程序同 IP 的其他用户全部受害。这种情况只靠情报源本身无法区分。解决在查询层加白名单机制把业务已知的合法域名和 IP 段维护成一张bypass_rules表命中告警时先查白名单再决定是否升级。这不是让你忽略威胁而是降低无效告警对研判精力的消耗。6. 关联分析与内部情报平台对接让数据真正用起来6.1 交叉关联同一攻击者留下的线索串起来情报数据的价值在关联而不在单条。一个事件里提取出的恶意 IP 可能在其他事件的域名解析记录里出现过把这个关联关系找出来就能画出攻击基础设施的轮廓。def correlate_by_ip(conn, target_ip): rows conn.execute( SELECT DISTINCT i2.incident_id, i2.value, i2.ioc_type FROM iocs i1 JOIN iocs i2 ON i1.incident_id i2.incident_id WHERE i1.value ? AND i1.ioc_type ip AND i2.value ! ? , (target_ip, target_ip), ).fetchall() return rows这段 SQL 做的事很直接找到包含目标 IP 的所有事件再把这些事件涉及的其他 IOC 全部拉出来。SELECT DISTINCT是为了去重因为同一个事件里同一个 IOC 可能被重复提取多次。实际用的时候还可以加i1.confidence 0.6这类条件把低置信度的关联过滤掉避免噪声干扰判断。6.2 MRF 排序怎么知道哪条情报最值得先处理事件批量入库之后你面临的就是「先看哪条」的问题。按时间倒序是最初级的方式但容易漏掉重要性高的事件。我在这套资源里实现了 MRFMost Relevant First排序综合事件严重级别、IOC 数量、来源可信度三个维度计算一个分数def compute_mrf_score(conn, incident_id): inc conn.execute( SELECT severity, source FROM incidents WHERE id ?, (incident_id,) ).fetchone() if not inc: return 0 severity, source inc ioc_count conn.execute( SELECT COUNT(*) FROM iocs WHERE incident_id ?, (incident_id,) ).fetchone()[0] source_conf SOURCE_WEIGHT.get(source, 0.5) score severity * 10 min(ioc_count, 10) * 2 source_conf * 5 return score这个打分公式是可调的严重级别的权重最大因为漏洞利用的紧急程度是首要维度IOC 数量起到辅助作用一个事件带 10 条指标比只带 1 条的研判价值更高来源权重最小只是兜底。你可以根据自己的业务调整系数比如对挖矿事件更敏感就把severity系数加大。6.3 导出对接生成给防火墙和 EDR 用的 IOC 列表情报平台的最终输出是给设备消费的。代码包里包含一个导出模块支持两种格式分别是纯文本 IP 列表和 CSV 格式。python query.py --type ip --min-confidence 0.7 --format txt python query.py --type sha256 --since 2024-01-01 --format csv导出结果可以直接导入防火墙的地址组、SIEM 的威胁情报插件或者分发给内部威胁研判群。建议每天定时导出一次而不是实时查询——实时查询会让数据库频繁被读影响采集入库的性能。我第一次就踩了这个坑导出任务和采集任务抢数据库锁导致入库延迟。从那以后我强制把导出改成每天凌晨跑一次数据写入和读取彻底分开再没出过问题。情报系统是一步步长大的先把采集和入库跑通再叠加过滤和关联最后再对接设备。希望这套资源帮你少走点弯路。本文还有配套的精品资源点击获取 Python构建威胁情报IOC汇总分析系统实战指南
简介这是一套基于Python的威胁情报自动播报与聚合工具面向安全运维人员、威胁情报分析者及对漏洞动态有持续跟踪需求的开发者。项目整合360、奇安信、红后、绿盟、安全客、斗象等公开情报源自动爬取CVE等威胁信息并通过邮件推送、页面TOP10展示与本地归档三种方式输出情报。资源共70个文件以Python脚本25个、XML配置、DAT缓存数据、HTML页面模板为主压缩包仅738KB代码结构清晰包含爬虫、数据处理、消息通知、SQL建库脚本及GitHub Actions工作流配置便于二次开发和自定义部署。已有191人学习下载。借助本项目可快速搭建一套免服务器运维的情报订阅系统理解多源情报采集、去重归档与多渠道播报的完整实现思路适合个人或小团队持续跟踪外部威胁动态。1. 威胁情报不是爬虫项目先想清楚你要汇总什么做威胁情报threat-intelligence踩过最大的坑是把它当成爬虫项目来启动。真正跑起来才发现80% 的工作量不在抓取而在清洗、去重、验证和关联。公开渠道的安全通告、漏洞库、恶意 IP 列表每天产生上千条数据但其中大量是重复的还有不少是误报和过时信息。这套资源的核心不是给你一堆采集脚本而是把「采集 → 解析 → 存储 → 查询 → 验证」的完整链路串起来用 Python 实现一个能落地的 IOC入侵指标汇总分析流程。适合安全工程师、运维人员以及需要做内部威胁情报平台选型但又不想一开始就上重型商业方案的人。你跟着做出来的东西可以对接已有的告警系统也可以当成一个独立的情报查询工具用。2. 整体架构与数据模型SQLite 加事件表不上一堆重组件2.1 为什么不用 Elasticsearch 或关系型大库很多人一上来就考虑 Elasticsearch理由是情报数据要全文检索。但实际场景里一个中小团队每天处理的情报量在几千到几万条级别SQLite 完全扛得住而且部署成本几乎为零。Elasticsearch 的运维负担、内存占用、数据导入导出的麻烦在这个量级上不划算。我一般建议先跑通 SQLite 版本等真的出现性能瓶颈再迁移到时候表结构和查询逻辑大体不用改。数据模型的设计上我见过不少反例有人把 IOC 塞进一个超大宽表结果查一次关联要 JOIN 五六个字段效率极低。核心思路应该是一张事件主表incidents加一张指标明细表iocs两表之间通过事件 ID 关联。主表存事件元信息明细表存 IP、域名、哈希、URL 等具体指标。这样设计的好处是一个事件可能包含多个 IOC 类型而你查询某个 IP 时只需要在明细表里做索引扫描速度快逻辑也清晰。2.2 表结构设计字段别贪多够用就行给出一份可以直接照抄的建表 SQL这是整套资源里所有脚本的基础。CREATE TABLE incidents ( id INTEGER PRIMARY KEY AUTOINCREMENT, source TEXT NOT NULL, title TEXT, description TEXT, published_at TEXT, fetched_at TEXT DEFAULT CURRENT_TIMESTAMP, severity INTEGER DEFAULT 2, status TEXT DEFAULT new ); CREATE TABLE iocs ( id INTEGER PRIMARY KEY AUTOINCREMENT, incident_id INTEGER NOT NULL, ioc_type TEXT NOT NULL, value TEXT NOT NULL, confidence REAL DEFAULT 0.5, first_seen TEXT, last_seen TEXT, FOREIGN KEY (incident_id) REFERENCES incidents(id) ); CREATE INDEX idx_iocs_type_value ON iocs(ioc_type, value); CREATE INDEX idx_iocs_value ON iocs(value);这里几个字段需要解释一下。severity用整数不用字符串0 到 3 分别对应低危到危急排序和过滤都方便confidence是浮点数0 到 1 之间表示这条 IOC 的可信度来自厂商标注或者你自己的验证逻辑status字段标记处理状态new、investigating、confirmed、false_positive四种。索引方面iocs表必须建value的单列索引因为这是查询频率最高的路径。2.3 模块划分采集、解析、入库三个脚本各自独立这套资源的代码包按功能拆成了四个模块collector.py负责抓取parser.py负责解析storage.py负责入库query.py负责查询导出。分开写比一个大文件实用得多——采集源出问题只需要改 collector解析规则变了只动 parser互不干扰。3. 采集层实现多源抓取与频率控制3.1 公开安全通告源的通用采集模板常见的威胁情报源大致分两类一类是结构化 API返回 JSON 或 CSV另一类是安全公告页面返回 HTML 或 RSS。第一类写起来简单第二类才是真正麻烦的地方。给一个处理 RSS 通告源的模板适配大多数安全厂商的公告系统。import requests import xml.etree.ElementTree as ET from datetime import datetime, timezone FEED_URLS { example_cert: https://example-cert.org/feeds/advisories, example_vendor: https://example-vendor.com/security/bulletin.rss, } def fetch_feed(name, url, timeout15): headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) threat-intel-collector/1.0 } try: resp requests.get(url, headersheaders, timeouttimeout) resp.raise_for_status() return resp.content except requests.exceptions.Timeout: print(f[{name}] timeout after {timeout}s, skip this round) return None except requests.exceptions.RequestException as e: print(f[{name}] request failed: {e}) return None def parse_rss(content): root ET.fromstring(content) items [] for item in root.iter(item): entry { title: item.findtext(title, ).strip(), link: item.findtext(link, ).strip(), description: item.findtext(description, ).strip(), pub_date: item.findtext(pubDate, ).strip(), } items.append(entry) return items两个参数值得注意。timeout15是请求超时阈值设短了网络抖动容易失败设长了采集任务会被卡住15 秒是我试下来比较平衡的值。User-Agent必须设置成真实浏览器样式很多情报源会拒绝裸的 Python 请求。RSS 解析用xml.etree.ElementTree而不是正则因为 XML 实体的转义规则用正则容易漏尤其 description 字段里经常带 HTML 标签。3.2 增量采集与去重记住上次的游标位置全量抓取每次都从头开始既浪费带宽又会产生大量重复数据。正确做法是记录每个源上次成功获取到的最新条目时间或 ID下次增量拉取。下面这段代码展示了如何利用fetched_at做增量判断import sqlite3 from datetime import datetime, timedelta def get_last_fetch_time(conn, source): row conn.execute( SELECT MAX(fetched_at) FROM incidents WHERE source ?, (source,) ).fetchone() if row and row[0]: return datetime.fromisoformat(row[0]) return datetime.now(timezone.utc) - timedelta(days7) def is_duplicate(conn, source, title): row conn.execute( SELECT id FROM incidents WHERE source ? AND title ? LIMIT 1, (source, title), ).fetchone() return row is not Noneget_last_fetch_time的作用是给采集器一个时间窗口只处理这个时间之后发布的条目。is_duplicate用source title作为自然键判断是否已经入库。这里有个细节不要用link字段做唯一性判断因为有些源的链接会带跟踪参数同一个通告可能生成多个不同 URL导致重复入库。3.3 抓取失败的补偿策略采集任务跑起来之后你会发现情报源不稳定是常态尤其是境外源经常超时或者返回 502。常见的做法是连续失败三次才告警单次失败只记录日志下一轮继续重试。这个逻辑必须写进主循环否则任何一次抓取异常都会白屏崩溃。import time FAIL_THRESHOLD 3 def run_collector(conn, source, url, fail_count0): content fetch_feed(source, url) if content is None: fail_count 1 if fail_count FAIL_THRESHOLD: print(f[{source}] failed {FAIL_THRESHOLD} times, manual check required) fail_count 0 return fail_count items parse_rss(content) for item in items: if not is_duplicate(conn, source, item[title]): insert_incident(conn, source, item) return 0这段代码的核心是fail_count的累积与重置机制。连续失败三次说明源可能挂了或者改版了需要人工介入单次失败不处理等下次调度。采集频率控制在每 30 到 60 分钟一轮比较合适低于 30 分钟容易被封 IP高于 60 分钟情报时效性会打折。4. 情报解析与 IOC 提取正则之外还要有启发式规则4.1 从通告正文里抠出四类 IOC通告的正文通常是一段自然语言描述夹杂着 IP 地址、域名、哈希值、URL。第一版我直接用正则提取效果很差URL 里包含的域名会被误识别为独立域名文件哈希和软件版本号混在一起也分不清。后来改成「正则提取候选值 启发式规则分类」的两步走思路。import re IP_RE re.compile(r(?!\d)(?:\d{1,3}\.){3}\d{1,3}(?!\d)) HASH_RE re.compile(r\b(?:[a-fA-F0-9]{32}|[a-fA-F0-9]{40}|[a-fA-F0-9]{64})\b) DOMAIN_RE re.compile(r\b(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)[a-z]{2,}\b, re.IGNORECASE) URL_RE re.compile(rhttps?://[^\s\]) def extract_iocs(text): iocs [] for m in URL_RE.finditer(text): iocs.append((url, m.group(0))) for m in HASH_RE.finditer(text): h m.group(0).lower() if len(h) 32: iocs.append((md5, h)) elif len(h) 40: iocs.append((sha1, h)) else: iocs.append((sha256, h)) for m in DOMAIN_RE.finditer(text): domain m.group(0).lower() if not any(domain in u for u, _ in iocs if _ url): iocs.append((domain, domain)) for m in IP_RE.finditer(text): ip m.group(0) if is_valid_ip(ip): iocs.append((ip, ip)) return iocs这里有个顺序问题必须先提取 URL再从结果里排除域名否则 URL 里的域名会被当成独立域名重复计入。HASH_RE里三个长度分别对应 MD5、SHA1、SHA256提取后统一转成小写因为后面做关联分析时大小写不一致会导致 JOIN 失败。is_valid_ip是一个校验函数过滤掉类似999.1.1.1这种正则能匹配但不合法的值。4.2 启发式过滤去掉明显误报正则提取只是第一步误报依然很多。比如正文里提到的示例 IP192.0.2.1RFC 文档保留地址段、内部 IP10.0.0.1、以及作者举例用的假域名example.com。这些必须在解析阶段干掉否则存进去之后每次查询都会被干扰。RESERVED_IP_RANGES [10., 192.168., 127., 0., 172.16., 172.17., 172.18., 172.19., 172.2, 172.3] RESERVED_DOMAINS [example.com, example.org, example.net, test.com, localhost] def filter_false_positives(iocs): filtered [] for ioc_type, value in iocs: if ioc_type ip: if any(value.startswith(prefix) for prefix in RESERVED_IP_RANGES): continue elif ioc_type domain: if any(value.endswith(domain) for domain in RESERVED_DOMAINS): continue filtered.append((ioc_type, value)) return filtered这段过滤逻辑看着简单但实际很有用。地址段前缀的判断用了startswith而不是精确匹配是因为172.16.0.1到172.31.255.255这些私网地址段范围较宽用前缀匹配是业内默认做法。注意172.2和172.3是为了覆盖172.20.x.x到172.30.x.x的部分地址段不算精确但够用。4.3 置信度打分不是所有情报可信度都一样厂商通告里的 IOC 可信度高第三方汇总源的可信度低。这套资源里把置信度拆成两个维度来源权重和指标类型权重。厂商官方源给 0.9第三方聚合源给 0.6IP 类指标因为动态变化频繁打 0.1 的降权系数域名类降权 0.2。SOURCE_WEIGHT {vendor_official: 0.9, aggregator: 0.6, community: 0.4} TYPE_WEIGHT {sha256: 1.0, sha1: 0.9, md5: 0.8, domain: 0.7, url: 0.6, ip: 0.4} def compute_confidence(source_type, ioc_type): return round(SOURCE_WEIGHT.get(source_type, 0.5) * TYPE_WEIGHT.get(ioc_type, 0.5), 2)哈希值的权重比 IP 高是有原因的攻击者换 IP 成本极低换一个文件哈希成本高得多所以哈希的稳定性更强。置信度最终存到iocs表的confidence字段里查询的时候可以按这个字段排序或过滤。5. 踩坑与排查存活性验证、编码脏数据、增量逻辑失效5.1 存活性验证情报库里 70% 的 IP 可能已经失效现象从情报库里随机抽出 100 条 IP 数据做检测能连上的不到三成大量情报来源已经是死数据。原因IP 和域名的存活性周期很短攻击基础设施经常变国内 IP 更是说封就封。采集入库时不验证结果只能当历史档案用。解决入库后 24 小时内必须做一轮存活性探测。IP 用 TCP 连接测试常见端口域名做 DNS 解析确认是否仍然指向有效地址。注意事项是探测频率不能太高否则情报源会把你封掉。5.2 编码与脏数据公告页面编码不一致导致解析挂掉现象某个情报源昨天还能正常解析今天突然提取不到任何 IOC检查日志发现解析器报了一堆乱码错误。原因不同源用的编码不一样有的 GBK有的 UTF-8还有的页面声称是 UTF-8 实际是 Latin-1。requests 的resp.text默认使用响应头声称的编码遇到错误的声明就会解出乱码。解决抓取后强制用resp.content配合chardet重新检测编码不要直接用resp.text。代码库里已经内置了这个逻辑你只需要在改源的时候注意别把这个步骤跳过去。5.3 增量逻辑失效漏抓两周数据现象某天发现incidents表里有个源的数据停留在两周前但采集任务一直显示成功执行。原因get_last_fetch_time用的是MAX(fetched_at)但fetched_at是入库时间不是信息发布的时间。如果某次批量导入历史数据fetched_at会被整体刷新导致增量窗口错乱。解决改用published_at字段作为增量游标这个字段保存的是情报源原始发布时间不会因为入库操作而变化。如果某个源不提供published_at就用link指纹去重两套逻辑互相兜底。5.4 业务误报情报平台命中≠安全事件现象情报平台频繁告警说内网 IP 访问了恶意域名但核查后发现是某个业务系统的合法调用。原因很多云服务商的 IP 段被误报进威胁情报库比如共享 IP 上有人跑过恶意程序同 IP 的其他用户全部受害。这种情况只靠情报源本身无法区分。解决在查询层加白名单机制把业务已知的合法域名和 IP 段维护成一张bypass_rules表命中告警时先查白名单再决定是否升级。这不是让你忽略威胁而是降低无效告警对研判精力的消耗。6. 关联分析与内部情报平台对接让数据真正用起来6.1 交叉关联同一攻击者留下的线索串起来情报数据的价值在关联而不在单条。一个事件里提取出的恶意 IP 可能在其他事件的域名解析记录里出现过把这个关联关系找出来就能画出攻击基础设施的轮廓。def correlate_by_ip(conn, target_ip): rows conn.execute( SELECT DISTINCT i2.incident_id, i2.value, i2.ioc_type FROM iocs i1 JOIN iocs i2 ON i1.incident_id i2.incident_id WHERE i1.value ? AND i1.ioc_type ip AND i2.value ! ? , (target_ip, target_ip), ).fetchall() return rows这段 SQL 做的事很直接找到包含目标 IP 的所有事件再把这些事件涉及的其他 IOC 全部拉出来。SELECT DISTINCT是为了去重因为同一个事件里同一个 IOC 可能被重复提取多次。实际用的时候还可以加i1.confidence 0.6这类条件把低置信度的关联过滤掉避免噪声干扰判断。6.2 MRF 排序怎么知道哪条情报最值得先处理事件批量入库之后你面临的就是「先看哪条」的问题。按时间倒序是最初级的方式但容易漏掉重要性高的事件。我在这套资源里实现了 MRFMost Relevant First排序综合事件严重级别、IOC 数量、来源可信度三个维度计算一个分数def compute_mrf_score(conn, incident_id): inc conn.execute( SELECT severity, source FROM incidents WHERE id ?, (incident_id,) ).fetchone() if not inc: return 0 severity, source inc ioc_count conn.execute( SELECT COUNT(*) FROM iocs WHERE incident_id ?, (incident_id,) ).fetchone()[0] source_conf SOURCE_WEIGHT.get(source, 0.5) score severity * 10 min(ioc_count, 10) * 2 source_conf * 5 return score这个打分公式是可调的严重级别的权重最大因为漏洞利用的紧急程度是首要维度IOC 数量起到辅助作用一个事件带 10 条指标比只带 1 条的研判价值更高来源权重最小只是兜底。你可以根据自己的业务调整系数比如对挖矿事件更敏感就把severity系数加大。6.3 导出对接生成给防火墙和 EDR 用的 IOC 列表情报平台的最终输出是给设备消费的。代码包里包含一个导出模块支持两种格式分别是纯文本 IP 列表和 CSV 格式。python query.py --type ip --min-confidence 0.7 --format txt python query.py --type sha256 --since 2024-01-01 --format csv导出结果可以直接导入防火墙的地址组、SIEM 的威胁情报插件或者分发给内部威胁研判群。建议每天定时导出一次而不是实时查询——实时查询会让数据库频繁被读影响采集入库的性能。我第一次就踩了这个坑导出任务和采集任务抢数据库锁导致入库延迟。从那以后我强制把导出改成每天凌晨跑一次数据写入和读取彻底分开再没出过问题。情报系统是一步步长大的先把采集和入库跑通再叠加过滤和关联最后再对接设备。希望这套资源帮你少走点弯路。本文还有配套的精品资源点击获取