简介：这份开源威胁情报采集系统资源包面向网络安全初学者与安全运维人员，帮助读者理解威胁情报的获取、分析与利用流程，并搭建可运行的情报采集与响应机制。压缩包共30个文件，约45KB，以16个Python源码与10个pyc编译文件为核心，辅以txt依赖说明、cfg配置、md说明文档等，整体结构清晰，便于按模块阅读与调试。教程部分覆盖威胁情报来源、爬虫抓取与反爬处理、数据清洗与Pandas分析、威胁建模、实时监控与警报、可视化报告以及法律合规等关键知识点；数据部分则提供历史威胁事件、恶意IP列表、漏洞信息与样本文件等素材，可用于威胁模拟、情报验证与共享实践。目前已有268人学习，适合希望系统掌握情报采集方法、提升个人技能或企业安全防护能力的读者参考。
1. 开源威胁情报采集系统：从一堆散落 IOC 到可查询情报库的最小闭环
手里拿到一个叫「开源威胁情报采集系统内含教程以及数据.zip」的包，多数人的第一反应是解压、找 README、跑起来看看。但真正卡住人的从来不是安装，而是跑通之后：采集源怎么配、IOC 怎么去重、数据存哪、怎么查。开源威胁情报采集系统要解决的核心问题，是把散落在公开渠道的恶意 IP、域名、URL、文件哈希这些 IOC（Indicator of Compromise，失陷指标），自动抓回来、清洗归一、落库，最后变成能被人和机器查询的情报资产。它适合安全运营、威胁狩猎、SOC 告警富化这几类场景的从业者，也适合想自己搭一套情报管道的工程师。这篇不聊概念，直接按「采集 → 解析 → 归一 → 存储 → 查询」这条链路，把每一步的参数、代码和翻车点讲清楚，让你照着能复现一个最小可用版本。
2. 采集层怎么搭：源选型、调度与去重的三个关键决策
威胁情报采集系统的第一层是「把数据拿回来」。这一步看着简单，实际决定了后面所有环节的质量。源选错，后面全是噪声；调度写崩，IP 被封；去重没做，库里全是重复 IOC。下面把这三个决策拆开讲。
2.1 情报源选型：免费源、社区源和自建源怎么配比
常见做法是把源分成三类。第一类是结构化程度高的公开情报源，通常提供 JSON 或纯文本格式的 IOC 列表，字段规整，适合直接入库。第二类是社区分享型源，格式不统一，需要写针对性解析器。第三类是自己从日志、蜜罐、沙箱里产出的内部源，质量最高但量小。
选型时我一般按三个维度打分：更新频率、字段完整度、误报率。更新频率低于每天一次的源，做实时告警富化基本没用；字段只有 IP 没有时间戳和置信度的源，入库后没法做时效衰减；误报率高的源要么降权，要么只做旁路参考。
| 源类型 | 典型格式 | 更新频率 | 适合用途 | 注意事项 |
|---|---|---|---|---|
| 结构化公开源 | JSON / CSV | 小时级到天级 | 批量入库、富化 | 注意字段映射和时区 |
| 社区分享源 | 文本 / 混合 | 不定 | 补充覆盖 | 需写解析器，误报偏高 |
| 内部自建源 | 自定义 | 实时 | 高置信告警 | 量小，需与外部源关联 |
配比上，我一般让外部源占 IOC 总量的七成左右，内部源占三成但权重最高。查询时按来源打置信度分，而不是一视同仁。
2.2 用 Python 写一个带重试和限速的采集器
采集器最容易翻车的地方是没做限速和重试，跑几次就被源站封了。下面是一个最小可用的采集骨架，带指数退避重试和请求间隔控制。
import time import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry # 配置重试策略：最多重试3次，退避因子0.5秒 session = requests.Session() retry = Retry( total=3, backoff_factor=0.5, status_forcelist=[429, 500, 502, 503, 504], ) session.mount("https://", HTTPAdapter(max_retries=retry)) def fetch_ioc(url, interval=2.0, timeout=15): """采集单个源，interval 控制两次请求最小间隔""" time.sleep(interval) # 简单限速，避免触发源站风控 resp = session.get(url, timeout=timeout) resp.raise_for_status() return resp.text if __name__ == "__main__": sources = [ "https://example.com/ioc/ip.txt", "https://example.com/ioc/domain.json", ] for u in sources: try: data = fetch_ioc(u) print(f"{u} 采集成功，长度 {len(data)}") except Exception as e: print(f"{u} 采集失败：{e}")逻辑说明：Retry对象负责网络层重试，status_forcelist里列的状态码才会触发重试，429 是限流、5xx 是服务端错误，这两类重试有意义；4xx 里的 403、404 重试没意义，所以不列。interval是请求间隔，公开源一般 1 到 3 秒比较安全，具体看源站 robots 和实际反馈。
参数说明：total=3是总重试次数，别设太大，否则一个坏源会拖住整个调度；backoff_factor=0.5意味着第一次重试等 0.5 秒，第二次 1 秒，第三次 2 秒，指数增长；timeout=15是单次请求超时，采集源响应慢时别设太小，否则大量误判失败。
2.3 采集去重：布隆过滤器还是数据库唯一索引
去重放在采集层还是入库层，是个常见分歧。我的经验是两层都做，但职责不同。采集层用布隆过滤器做快速预判，避免把明显重复的数据往解析流程里送；入库层用数据库唯一索引做最终兜底，保证不重。
布隆过滤器的坑在于有假阳性，它说「可能存在」时不一定真存在，但说「不存在」时一定不存在。所以采集层用它过滤「确定不存在」的，剩下的交给入库层判断。如果 IOC 量在百万级以内，直接用数据库唯一索引也扛得住，不必上布隆过滤器增加复杂度。
from bloom_filter2 import BloomFilter # 预计存100万条，误判率0.1% bloom = BloomFilter(max_elements=1_000_000, error_rate=0.001) def is_new_ioc(ioc): """返回 True 表示可能是新 IOC，False 表示一定重复""" if ioc in bloom: return False bloom.add(ioc) return True这段代码里max_elements和error_rate要按实际量估。估小了误判率飙升，估大了内存浪费。100 万条、0.1% 误判率大概占几 MB 内存，可以接受。
3. 解析与归一：把五花八门的 IOC 变成统一结构
采集回来的数据格式千奇百怪，有的用逗号分隔，有的嵌套 JSON，有的还带 HTML 标签。这一层的目标是把它们统一成一张表能存的结构：IOC 值、类型、来源、首次发现时间、最后更新时间、置信度。归一没做好，后面查询就是灾难。
3.1 IOC 类型识别：正则匹配的边界与误判
IOC 主要分四类：IP、域名、URL、文件哈希。识别靠正则，但正则写太松会误判，写太紧会漏。下面是我常用的识别函数。
import re PATTERNS = { "ipv4": re.compile(r"^(?:(?:25[0-5]|2[0-4]\d|1?\d?\d)\.){3}(?:25[0-5]|2[0-4]\d|1?\d?\d)$"), "domain": re.compile(r"^(?:[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?\.)+[a-zA-Z]{2,}$"), "md5": re.compile(r"^[a-fA-F0-9]{32}$"), "sha256": re.compile(r"^[a-fA-F0-9]{64}$"), "url": re.compile(r"^https?://[^\s]+$"), } def classify_ioc(value): """按优先级匹配，URL 优先于域名，哈希优先于域名""" value = value.strip() for ioc_type in ["url", "sha256", "md5", "ipv4", "domain"]: if PATTERNS[ioc_type].match(value): return ioc_type return None逻辑说明：匹配顺序很关键。URL 里包含域名，如果先匹配域名，http://evil.com/a会被误判成域名。哈希是纯十六进制，某些短域名也可能全是十六进制字符，所以哈希要排在域名前面。ipv4的正则限制了每段 0 到 255，避免999.1.1.1这种误判。
参数说明：域名正则里{0,61}是单段标签长度限制，符合 DNS 规范；{2,}是顶级域至少两位。实际会遇到国际化域名和特殊顶级域，需要时再补。
3.2 字段归一：时间戳、置信度和来源标记怎么统一
不同源的时间格式不一样，有 Unix 时间戳、ISO 8601、还有2024/01/01 12:00这种。统一转成 UTC 的 ISO 格式最省事。置信度如果源没给，我一般按源的历史准确率给个默认值，比如结构化公开源给 60，社区源给 40，内部源给 90。
from datetime import datetime, timezone def normalize_time(raw): """把多种时间格式统一成 UTC ISO 字符串""" if isinstance(raw, (int, float)): return datetime.fromtimestamp(raw, tz=timezone.utc).isoformat() for fmt in ("%Y-%m-%dT%H:%M:%S", "%Y/%m/%d %H:%M", "%Y-%m-%d %H:%M:%S"): try: dt = datetime.strptime(raw, fmt).replace(tzinfo=timezone.utc) return dt.isoformat() except (ValueError, TypeError): continue return None # 解析不了就返回 None，入库时标记为未知这段的关键是「解析不了返回 None」而不是抛异常。采集源里混进脏数据是常态，一条脏数据不该让整批入库失败。入库时把 None 的时间标记为未知，后续可以人工补。
3.3 归一后的数据结构设计
归一后的每条记录我一般保留这些字段：ioc_value、ioc_type、source、first_seen、last_seen、confidence、tags。tags用来放家族名、攻击类型这类标签，方便后续按标签查询。表结构用下面这个建表语句就够起步。
CREATE TABLE ioc_records ( id BIGINT AUTO_INCREMENT PRIMARY KEY, ioc_value VARCHAR(512) NOT NULL, ioc_type VARCHAR(16) NOT NULL, source VARCHAR(128) NOT NULL, first_seen DATETIME, last_seen DATETIME, confidence TINYINT DEFAULT 50, tags VARCHAR(512), UNIQUE KEY uk_value_type (ioc_value(255), ioc_type), INDEX idx_type_seen (ioc_type, last_seen) );UNIQUE KEY用ioc_value前 255 字符加ioc_type做唯一约束，防止同一 IOC 重复入库。idx_type_seen索引服务于「查某类型最近更新的 IOC」这类高频查询。注意ioc_value用VARCHAR(512)是因为 URL 可能很长，但唯一索引只取前 255 字符，超长 URL 截断后可能碰撞，实际用的时候要么对 URL 做哈希存，要么单独处理。
4. 存储与查询：让情报库真正能被用起来
数据存进去只是第一步，能不能快速查出来才决定这套系统有没有价值。查询场景主要有三类：按 IOC 精确查、按类型和时间范围批量拉、按标签关联查。这三类对索引的要求不一样。
4.1 精确查询与批量查询的索引策略
精确查询走uk_value_type唯一索引，毫秒级返回。批量拉取走idx_type_seen，按类型加时间范围过滤。如果还要按来源过滤，可以再加一个idx_source_type联合索引。索引不是越多越好，每个索引都会拖慢写入，采集量大时写入性能会明显下降。
-- 精确查询单个 IOC SELECT ioc_value, ioc_type, source, confidence, last_seen FROM ioc_records WHERE ioc_value = '198.51.100.1' AND ioc_type = 'ipv4'; -- 批量拉取最近7天更新的域名类 IOC SELECT ioc_value, source, confidence FROM ioc_records WHERE ioc_type = 'domain' AND last_seen >= DATE_SUB(UTC_TIMESTAMP(), INTERVAL 7 DAY) ORDER BY last_seen DESC LIMIT 1000;第一条走唯一索引，第二条走idx_type_seen。注意时间函数用UTC_TIMESTAMP()而不是NOW()，因为入库时统一用了 UTC，查询也要对齐，否则时区差会漏数据。
4.2 时效衰减：过期情报怎么处理
威胁情报有时效性，一个半年前的恶意 IP 现在可能已经换主了。我一般给每条 IOC 算一个「有效分」，随时间衰减，查询时按有效分排序。
from datetime import datetime, timezone def freshness_score(last_seen, half_life_days=30): """按半衰期计算新鲜度，返回 0 到 1 之间的分数""" if not last_seen: return 0.0 now = datetime.now(timezone.utc) delta_days = (now - last_seen).total_seconds() / 86400 return 0.5 ** (delta_days / half_life_days)half_life_days=30意味着 30 天后分数降到 0.5，60 天后 0.25。这个参数按 IOC 类型调，IP 变化快可以设 15 天，文件哈希相对稳定可以设 90 天。查询时把freshness_score和confidence相乘作为综合排序依据，比单纯按时间排更合理。
4.3 用 Docker 把存储和查询服务跑起来
本地验证阶段，用 Docker 起一个 MySQL 最省事。下面是最小 compose 配置。
version: "3.8" services: ioc-db: image: mysql:8.0 environment: MYSQL_ROOT_PASSWORD: change_me MYSQL_DATABASE: threat_intel ports: - "3306:3306" volumes: - ./data:/var/lib/mysql command: --character-set-server=utf8mb4 --collation-server=utf8mb4_unicode_ciutf8mb4是为了支持 IOC 里可能出现的特殊字符。volumes把数据挂到本地，容器删了数据还在。生产环境别用 root 账号，建独立用户并限制权限。启动后把第 3 章的建表语句执行一遍就能用。
5. 避坑与排查：采集系统上线后最容易翻车的五个点
这一章全是血泪经验，每条都按「现象 → 原因 → 解决」写，遇到对应情况直接对号入座。
现象一：采集任务跑几天后突然全部失败，日志显示连接超时。原因：源站把你的 IP 限流或封了，通常是请求频率太高或没带合理的 User-Agent。 解决：把请求间隔调大到 3 到 5 秒，加上真实的 User-Agent 头，必要时分散到多个出口。别用默认的 python-requests UA，很多源站直接拦。
现象二：入库后发现同一个 IOC 存了好几条，唯一索引没生效。原因：ioc_value大小写不一致，或者 URL 带了不同的查询参数被当成不同值。 解决：入库前统一转小写（域名和哈希不区分大小写），URL 做规范化处理，去掉无意义的跟踪参数。唯一索引建之前先清洗存量数据，否则建索引会失败。
现象三：查询变慢，明明建了索引却走全表扫描。原因：查询条件里对索引列用了函数，比如WHERE DATE(last_seen) = '2024-01-01'，函数会让索引失效。 解决：改成范围查询WHERE last_seen >= '2024-01-01' AND last_seen < '2024-01-02'，让索引能用上。
现象四：置信度高的源和置信度低的源混在一起，告警噪声大。原因：查询时没按来源权重过滤，所有 IOC 一视同仁。 解决：查询加confidence >= 60这类条件，或者把来源权重做成配置表，查询时关联计算综合分。内部源单独走高优先级通道。
现象五：时间字段对不上，明明刚采集的数据显示是几小时前。原因：源给的是本地时间但没标时区，你按 UTC 解析了，或者反过来。 解决：入库前统一转 UTC，解析时如果源没标时区，按源站所在时区假设再转。查询展示时再按用户时区转回来。这个坑最隐蔽，建议入库时同时存原始时间字符串备查。
6. 进阶技巧：用标签关联把孤立 IOC 串成攻击画像
单条 IOC 的价值有限，真正有用的是把同一家族、同一攻击活动的 IOC 关联起来。做法是在归一阶段给 IOC 打标签，标签来源可以是源本身带的家族名，也可以是从上下文里提取的关键词。有了标签，查询就能从「查一个 IP」升级成「查这个家族最近用了哪些 IP 和域名」。
标签关联的一个实用技巧是做共现分析：如果两个标签经常出现在同一批 IOC 上，它们很可能属于同一攻击活动。下面这段代码演示怎么从标签共现里找出强关联对。
from collections import defaultdict from itertools import combinations def tag_cooccurrence(records, min_count=3): """records 是 (ioc_value, tags) 列表，tags 是逗号分隔字符串""" pair_count = defaultdict(int) for _, tags in records: tag_list = [t.strip() for t in tags.split(",") if t.strip()] for a, b in combinations(sorted(set(tag_list)), 2): pair_count[(a, b)] += 1 # 只保留共现次数达到阈值的对 return {pair: cnt for pair, cnt in pair_count.items() if cnt >= min_count} # 示例：找出共现3次以上的标签对 records = [ ("1.1.1.1", "apt-x,loader"), ("2.2.2.2", "apt-x,loader"), ("3.3.3.3", "apt-x,c2"), ("4.4.4.4", "apt-x,loader"), ] print(tag_cooccurrence(records, min_count=3))min_count是共现阈值，设太低会引入噪声，设太高会漏掉弱关联。我一般从 3 开始试，看结果再调。combinations前先set去重，避免同一标签重复配对。这个分析跑出来的强关联对，可以反过来指导标签体系的合并和拆分。
验证这套系统有没有真正跑通，我习惯做一件事：拿一个已知的恶意 IP，从采集到入库到查询走一遍全流程，看每一步的耗时和数据是否一致。如果查询结果里first_seen和last_seen合理、置信度符合来源预期、标签能关联出其他 IOC，这套最小闭环就算立住了。我自己踩过最深的坑是早期没做时间归一，导致时效衰减算出来全是负数，排查了大半天才发现是时区问题。所以每次上线新源，先拿几条数据手工核对时间字段，这个习惯帮我省了很多后悔药。希望帮到你。
本文还有配套的精品资源，点击获取
