Ayyadi266/threat-intel-platform

# mini-misp — 一个小巧、诚实的威胁情报平台 [](https://github.com/Ayyadi266/threat-intel-platform/actions/workflows/ci.yml)    收集免费的威胁情报 feeds，将其标准化为统一的 indicator schema，跨来源进行去重，以可论证的方式对置信度进行评分，并导出为 SOC 实际消耗的格式 —— Sigma 规则、STIX 2.1、CSV、blocklists 以及 SIEM watchlist。 ## 为什么会有这个项目 检测项目会消耗威胁情报。但几乎没有什么东西教你如何*生产*情报。我已经有了检测端 —— Sigma 规则、Python 日志分析引擎、EDR、Wazuh 实验室 —— 它们全都始于同一个未经检验的假设：一份“坏” indicator 列表会不知从哪里冒出来。这个工具就是那个“ somewhere（来源）”，并且它认真对待了其中令人不适的部分： - 免费的 feeds 之间存在分歧、会过期、有重叠，并且由少数几家相同的组织运营。 - 一个 indicator 是**某个人在某个时间点的一份报告**，而不是一个事实。 - 一个没人能质疑的置信度分数比没有分数更糟糕，因为无论如何它都会被盲目信任。 ## 诚实政策 这是我在面试中最会极力捍卫的部分，因为它是通过代码强制执行的，而不是在 docstring 中承诺的。 | 规则 | 如何强制执行 | | --- | --- | | 永远不对 indicator 标注“恶意” | `IOC` **没有** `malicious` / `verdict` / `threat_level` 字段，并且 [`test_schema_has_no_malicious_verdict_field`](tests/test_ioc.py) 会检查 `IOC.__slots__`，以防未来的某个提交出于好意而添加此类字段 | | 每条记录都有其来源描述 | `IOC.describe()` → `ipv4 192.0.2.55 \| reported by 3 source(s) (blocklist_de, feodo, otx) \| confidence 69/100 (medium) \| last seen 2026-07-29` | | 分数永远不会脱离其原因独立存在 | `ConfidenceAssessment` 携带 `reasons`；`query --explain` 会打印出完整的计算过程 | | 未评分 ≠ 零分 | 从未评分过的记录在每种输出格式中都会报告为 `confidence not computed` | | 过期的分数永远无法导出 | `IOC.merge()` 会**清除**缓存的 confidence，因此源自单一来源的分数在结构上无法在引入第二个来源后依然存续 | | 导出不能断言恶意性 | 每个导出器都会在 [`test_never_asserts_maliciousness_as_fact`](tests/test_exporters.py) 中接受扫描，检查是否存在绝对的断言 | | 每个 feed 必须记录其弱点 | 一致性测试会令任何 connector 的 `notes` 字段未能描述该 feed 缺陷的测试失败 | | 佐证必须是真实的 —— 由同一运营商共享的 feeds 会被降权 | 三个 abuse.ch 列表达成一致只能算作一家组织的观点（见 [置信度](#the-confidence-model)） | ## 快速开始 ``` git clone https://github.com/Ayyadi266/threat-intel-platform.git cd threat-intel-platform python -m pip install -e ".[dev,web]" # or just `pip install -e .` for the CLI python main.py feeds --verbose # what is configured, and its caveats python main.py update # collect everything available python main.py query --min-confidence 55 # what is worth acting on python main.py query 192.0.2.55 --explain # why that score python main.py export sigma --min-confidence 55 --out rules/intel.yml python main.py serve # dashboard on http://127.0.0.1:5000 ``` 无需 API key，无需网络，无需配置： ``` python main.py update --offline # exits 0; every feed reports "skipped" ``` 可选的免费 key 可以解锁另外三个 feed。没有它们也不会有任何破坏： ``` export ABUSECH_AUTH_KEY=... # one key for URLhaus + Feodo Tracker + ThreatFox export ABUSEIPDB_API_KEY=... export OTX_API_KEY=... ``` ## 架构 ``` flowchart TB subgraph SRC["Free feeds (network — mocked in every test)"] F1["URLhaus
malware URLs"] F2["Feodo Tracker
botnet C2 IPs"] F3["ThreatFox
mixed IOCs"] F4["OpenPhish
phishing URLs"] F5["Blocklist.de
attacker IPs"] F6["AbuseIPDB · OTX
(key-gated, optional)"] end subgraph FEEDS["feeds/ — one class per feed, zero core changes to add one"] HTTP["http.py
HttpClient Protocol
UrllibClient · OfflineClient"] BASE["base.py
FeedConnector ABC
fetch() ⟂ parse()"] REG["registry.py
auto-discovers feeds/*.py"] end subgraph CORE["core/ — normalization & intelligence"] IOC["ioc.py
IOC · SourceRef · TLP
per-type validation"] DEDUP["dedup.py
N feeds → 1 record
+ overlap matrix"] SCORE["scoring.py
reliability × fidelity × decay
− shared-operator discount"] CORR["correlation.py
URL→domain→IP
agreement tagging"] STORE["storage.py
SQLite upsert + merge"] end subgraph OUT["exporters/ — read-only views"] E1["Sigma"] E2["STIX 2.1-lite"] E3["CSV"] E4["blocklist"] E5["watchlist (JSONL)"] end CLI["main.py
update · query · export · stats"] WEB["web/ Flask dashboard + /api
(read-only, no intel logic)"] SRC --> HTTP --> BASE REG --> BASE BASE -->|"ParseResult: IOC[] + issues[]"| DEDUP IOC -.->|schema| BASE DEDUP --> SCORE --> CORR --> STORE STORE --> OUT CLI --> BASE CLI --> STORE CLI --> OUT STORE --> WEB classDef net fill:#3b1f1f,stroke:#a33,color:#eee classDef pure fill:#1f2b3b,stroke:#48c,color:#eee class SRC,HTTP net class IOC,DEDUP,SCORE,CORR,STORE,OUT pure ``` 红色部分接触网络。蓝色部分是纯粹且确定性的。依赖箭头仅单向运行：`core/` 绝不从 `feeds/` 导入任何内容，这就是为什么评分机制可以使用虚构的 feeds 进行单元测试。有关更新时序图，请参阅 [ARCHITECTURE.md](ARCHITECTURE.md)。 **一切所依赖的唯一决策：** `fetch()`（涉及网络、简单机械）与 `parse()`（纯粹的 `bytes → ParseResult`）被分离开来。每个解析器测试都会读取一个已提交的样本文件，因此测试套件中**完全没有使用任何 mocking 库** —— 格式错误的 feed 只会产生已记录的解析 *issues*，而绝不会导致更新运行崩溃。 ## Feeds 与可靠性先验 每个 connector 都在 0.0–1.0 之间声明一个 `reliability` 先验，作为其在评分中的权重。在这里填一个没有根据的数字比不填更糟，因此以下是每个数字的理由。 | Feed | 先验 | 运营商 | 类型 | 时间戳 | Key | 为什么是这个数字 | | --- | --- | --- | --- | --- | --- | --- | | **Feodo Tracker** | **0.90** | abuse.ch | ipv4, ipv6 | first + last | 可选 | 针对少数命名家族确认的 C2。条目是*专用* C2 主机，而不是共享的 Web 托管，因此误报在结构上是罕见的。设计上很窄 —— 未检出并不证明什么。 | | **URLhaus** | **0.85** | abuse.ch | url | first + last | 可选 | 社区精心维护的提供恶意软件的 URL 报告；准确率高。包含最近*已离线*的 URL（保留为 `status:` 标签），这对回溯狩猎非常有价值，但对拦截毫无用处。 | | **OpenPhish** | **0.75** | openphish | url | 无 | 否 | 自动化钓鱼检测，准确率不错，是对 URLhaus 的补充而非重叠。无时间戳，无家族归属，且免费 feed 只是一个很小的子集。 | | **ThreatFox** | **0.70** | abuse.ch | 全部 8 种 | first + last | **是** | 覆盖面广，且对恶意软件家族的归属做得很好，但由社区提交，维护程度远不及 Feodo。会发布其自身的 confidence，但这只会调节其自身的权重。 | | **AbuseIPDB** | **0.65** | abuseipdb | ipv4, ipv6 | 仅 last | **是** | 众包的*报告*，而非经过验证的发现。经常命中 VPN 出口、云出口和 NAT 网关。作为佐证很有用，作为唯一来源则很危险。 | | **AlienVault OTX** | **0.60** | alienvault-otx | 全部 8 种 | 仅 last | **是** | 质量完全取决于 pulse 的作者，从厂商分析文章到转发的 blocklists 不等。indicators 会在 pulses 间循环，因此所有 pulses 都被归属于同一个来源。结果取决于你的订阅，因此不同用户之间的运行结果不可复现。 | | **Blocklist.de** | **0.55** | blocklist.de | ipv4, ipv6 | 无 | 否 | 志愿者运行的 fail2ban 传感器：意味着“观察到失败的认证尝试”，而不是 C2。没有时间戳，并且动态 IP 变动频繁，因此条目很可能已经过期。 | 这张表的存在是为了揭示两个结构性事实： 1. **URLhaus、Feodo Tracker 和 ThreatFox 都是 abuse.ch 的。** 七个 feed 中有三个属于同一家组织。把它们的认同视为三次独立确认将是这个平台所能撒下的最大谎言，因此 `FeedMetadata.operator` 是一个必须具备的概念，且评分机制会对“兄弟姐妹”来源进行降权。 2. **abuse.ch 现在将批量访问限制在需要免费账户 key。** URLhaus 和 Feodo 会尝试进行无认证访问，如果收到 `403`，系统会报告 *"endpoint requires ABUSECH_AUTH_KEY"* —— 这是一个配置事实，而不是崩溃。 ### 添加一个 feed 只需要一个文件 ``` # feeds/my_feed.py — 无需注册，无需修改 core，无需修改 CLI class MyFeed(FeedConnector): meta = FeedMetadata(name="my_feed", reliability=0.6, operator="someone", ...) def fetch(self, http: HttpClient) -> bytes: return http.get(self.meta.endpoint) def parse(self, payload: bytes, *, fetched_at: datetime) -> ParseResult: ... # pure; returns ParseResult(iocs=[...], issues=[...]) ``` 将文件放入 `feeds/` 中，在 `tests/fixtures/feeds/my_feed__sample.json` 添加一个样本响应，它就会继承整个一致性测试套件：metadata 验证、离线降级、403 处理、垃圾 payload 处理、确定性、仅限声明类型，以及“从不分配自己的 confidence”。 ## 置信度模型 在 [`core/scoring.py`](core/scoring.py) 中实现，模型版本为 `v1-saturating-support`。 每个报告的 feed 都会贡献 **support**；总 support 会通过一个饱和曲线映射到 0–100 的区间： ``` support_i = reliability_i × fidelity_i × self_assessment_i × recency_i × sibling_i support = Σ support_i score = round(100 × support / (support + K)), K = 0.8 ``` ### 五个单来源因子 | 因子 | 范围 | 编码内容 | | --- | --- | --- | | `reliability` | 0.0–1.0 | 上表中该 feed 的先验值。未知 feed 获得的分数为 **0.40** —— 陌生人会被视为平庸水平，不被信任，并且 reason 行会直接声明这一点。 | | `fidelity` | 0.85 / 0.95 / 1.00 | feed 对数据打时间戳的程度（`none` / `last_seen_only` / `first_and_last`）。没有发布时间戳的 feed 会用我们获取时的时间代替，因此其新鲜度无法验证。 | | `self_assessment` | 0.75–1.25 | 如果 feed 发布了自身的 confidence（ThreatFox, AbuseIPDB），它最多会将**该 feed 自身的权重**调节 ±25%。feed 永远无法越过我们的先验进行自我评分。 | | `recency` | (0, 1] | `0.5 ** (age_days / half_life)`，基于**该来源的** `last_seen` 计算 —— 因此不断重新报告的 feed 会保持其贡献的新鲜度。 | | `sibling` | 1.0 或 0.5 | 在同一个运营商内，最强的来源完全计入；每增加一个额外来源，权重减半。 | ### 衰减半衰期因 indicator 类型而异 | 类型 | 半衰期 | 推理 | | --- | --- | --- | | `url` | **7 天** | 停用和轮换会在几天内使分发 URL 失效 | | `domain` | **21 天** | 寿命比其上托管的 URL 长，但注册商会采取行动 | | `ipv4`, `ipv6` | **14 天** | C2 频繁变动，并且动态地址空间会被重新分配给无辜的订阅者 | | `email` | **30 天** | 发件人地址在一段时间内会跨活动重复使用 | | `md5`, `sha1`, `sha256` | **180 天** | 哈希值能永久标识一个不可变文件；只有它的*相关性*会逐渐消失 | 一个统一的全局半衰期要么会让哈希值过期太快，要么会让死 URL 存活数月 —— 这就是为什么这是一个表格而不是一个常量。 ### 为什么使用饱和曲线而不是概率 OR Noisy-OR (`1 − Π(1 − rᵢ)`) 是教科书般的选择，但它远太自信、远太快速了：两个 0.85 的 feeds 会产生 **0.978**，即“98/100 确定”，没有哪个诚实的分析师会在只有两个 blocklists 达成一致时这么说。 `support / (support + K)` 在来源数量和可靠性上都是单调的，具有边际递减效应，并且渐近线*低于* 100 —— 该模型从字面上看就无法声称确定性。当 `K = 0.8` 时，一个新鲜且高质量的来源得分约为 53，四个来源得分约为 81，这就产生了诚实政策所需的属性： 分级：`high` ≥ 80，`medium` ≥ 55，`low` ≥ 30，否则为 `informational`。`K` 是控制整体严格程度的唯一旋钮 —— 提高它以要求更多的佐证。 ### 它会打印出自己的计算过程 ``` $ python main.py query 192.0.2.55 --explain ipv4 192.0.2.55 | reported by 3 source(s) (blocklist_de, feodo, otx) | confidence 69/100 (medium) | last seen 2026-07-29 confidence 69/100 (medium, model v1-saturating-support) - feodo: reliability 0.90 x timestamps 1.00 (first_and_last) x recency 0.91 (1.9d old, 14d half-life) = 0.817 support - otx: reliability 0.60 x timestamps 0.95 (last_seen_only) x recency 0.91 (1.9d old, 14d half-life) = 0.518 support - blocklist_de: reliability 0.55 x timestamps 0.85 (none) x recency 1.00 (0.0d old, 14d half-life) = 0.468 support - combined support 1.802 from 3 feed(s) across 3 independent operator(s) - score = 100 x 1.802 / (1.802 + 0.80) = 69.3 -> 69 tags: activity:brute-force, activity:c2, agreement:corroborated, agreement:independent-operators, ... source blocklist_de: seen 2026-07-29 .. 2026-07-29 source feodo: seen 2026-06-01 .. 2026-07-28 https://feodotracker.abuse.ch/browse/host/192.0.2.55/ source otx: seen 2026-07-28 .. 2026-07-28 https://otx.alienvault.com/pulse/66a1f0c0deadbeef00000002 ``` 如果那三个是 abuse.ch 的 feeds，相同的输出将会包含 `agreement:single-operator` 以及这行字：*"note: all sources are operated by 'abuse.ch', so this is not independent corroboration"*。 ### 每次更新时都会应用衰减 `update` 会对**整个存储库**重新评分，而不仅仅是新收集的内容，因为衰减是关于*现在*的函数。如果没有这样做，没人重新报告的 indicators 将永远保持昨天的分数，存储库将慢慢塞满过期的、高置信度的垃圾数据。（`--no-reprice` 可以选择跳过此步骤。） ## 关联分析 [`core/correlation.py`](core/correlation.py) 仅添加可以从已持有数据中推导出的内容 —— 它永远不会向网络请求任何东西。 - **结构关系。** 每当某个 URL 的主机 domain 或 IP *同时*存在于集合中时，该 URL 就会链接到其主机。链接是双向的，并存在于 `IOC.related` 中。 - **一致性标记。** `agreement:corroborated` / `agreement:single-source`，以及 `agreement:independent-operators` 对比 `agreement:single-operator` —— 后者标记那些*看起来*有佐证但实际上只是一家组织观点的。 - **基础设施枢纽。** 一个托管了 ≥2 个独立报告的 URL 的 host，将获得 `infra:hosts-multiple-flagged-urls` 标签和计数标签。 - **选择性推导。** `update --derive` 会将每个 URL 的主机生成为独立的 indicator，标记为 `derived:from-url`，供那些在 DNS 或防火墙级别进行拦截且无法匹配完整 URL 的消费者使用。 **永远不进行 DNS 解析。** 将 domain 转换为 IP 要么需要主动查找 —— 这会触及攻击者基础设施并暴露我们正在监控的事实 —— 要么需要付费的被动 DNS feed。因此 domain→IP 的边缘关系只存在于 feed 已经报告了两者的情况下，并且 dashboard 会在每个详情页面上如实陈述这一点。 ## 导出器 | 格式 | 文件 | 消费者 | 备注 | | --- | --- | --- | --- | | `sigma` | `.yml` | 通过 `sigma convert` 的 SIEM | 每个 **(type, confidence band)** 一条规则，因此每条规则的 `level` 对其内容都是真实的。确定性的 `uuid5` 规则 ID 源自规则*身份*，因此每晚重新导出会更新规则而不是生成重复规则。每条规则都包含诚实的 `falsepositives`。 | | `stix` | `.json` | MISP, OpenCTI, TIPs | STIX 2.1 bundle：`identity`，带有真实 patterns 的 `indicator` SDOs，`confidence`，每个 feed 的 `external_references`，TLP 2.0 `object_marking_refs`，以及 `related-to` `relationship` 对象。 | | `csv` | `.csv` | 电子表格，临时分析 | 包含一个 `confidence_reasons` 列，携带完整的推导过程。固定使用 `\n` 换行符以确保 diff 可复现。 | | `blocklist` | `.txt` | 防火墙/代理/DNS ACLs | 将裸值分组到带有计数的、清晰标记的按类型划分的区块中。 | | `blocklist-annotated` | `.txt` | 相同，适用于允许注释的场景 | `value # reported by N sources, confidence X, last seen D` | | `watchlist` | `.jsonl` | 我的日志分析引擎，Splunk lookups，Sentinel watchlists | 见下文。 | 添加一个格式也只需一个文件 —— 导出器使用相同的自动发现机制。 ### `sigma` 输出是真正的 Sigma ``` title: 'Threat Intel Match: ipv4 (medium confidence)' id: 13506b89-5963-5b36-996a-b005045bf031 status: experimental description: 'Matches 2 ipv4 indicator(s) ... Confidence 60-69/100 (band ''medium''). These are REPORTS, not verdicts: a match means an indicator was reported by a feed, not that the activity is confirmed malicious. Verify before acting.' tags: - attack.command-and-control - malware.emotet logsource: category: network_connection detection: selection: DestinationIp: - 192.0.2.55 - 192.0.2.77 condition: selection falsepositives: - 'Shared hosting, CDNs and URL shorteners that also serve legitimate content' - 'NAT gateways, VPN exit nodes and cloud egress addresses reported by sensor feeds' level: medium ``` 注意*缺少了什么*：没有捏造的 `attack.tXXXX` 技术 ID。一条 blocklist 记录并不能识别一项技术，因此只会输出由 feed 自身的活动标签所暗示的战术。YAML 由约 40 行的发射器编写，以保持零依赖的承诺；其引用规则经过了单元测试。 ### `watchlist` schema (JSON Lines, v1) 每行一个 JSON 对象。第 1 行是 `_meta` 记录（schema 版本、选择内容、免责声明）；随后的每一行都是一个 indicator： ``` {"indicator":"192.0.2.55","type":"ipv4","match_field":"dest_ip","confidence":69, "band":"medium","source_count":3,"sources":["blocklist_de","feodo","otx"], "operators_known":true,"first_seen":"2026-06-01T09:14:22+00:00", "last_seen":"2026-07-29T...","age_days":0.0,"tlp":"TLP:CLEAR", "tags":["activity:c2","malware:emotet"],"related":[], "note":"ipv4 192.0.2.55 | reported by 3 source(s) ... | last seen 2026-07-29", "reasons":["feodo: reliability 0.90 x ..."]} ``` 选择 JSONL 而不是 JSON 或 CSV 的三个原因：检测器可以在文件读取完毕之前就开始流式匹配；一个*写了一部分*的 JSONL 文件仍然是有效的（而截断的 JSON 数组则不是，当 cron 覆盖检测器正在读取的文件时这很重要）；并且类型原生保留，因此 `confidence` 是数字而 `tags` 是列表。 这里故意**没有** `action`、`severity` 或 `malicious` 键。决定是否拦截是消费者的事，它需要利用 `confidence`、`age_days` 和 `source_count` 来做出决定 —— 这正是该 schema 所移交的内容。 ## CLI ``` $ python main.py --help feeds list connectors, reliability priors and documented caveats check validate/normalize indicator values (no network, no database) update collect every feed, dedup, correlate, score, persist query search the store, optionally printing the confidence derivation export render the store in any registered format stats what is in the store and when each feed last worked serve read-only Flask dashboard ``` `update` 报告实际发生的情况，包括跨 feed 的重叠情况： ``` $ python main.py update feed status indicators records issues seconds detail ------------ ------- ---------- ------- ------ ------- --------------------------------- abuseipdb skipped 0 0 0 0.00 no API key: set ABUSEIPDB_API_KEY blocklist_de ok 8 11 3 0.41 feodo ok 3 5 2 0.33 openphish ok 5 6 1 0.28 otx ok 6 8 2 0.51 threatfox ok 3 5 2 0.44 urlhaus ok 4 7 3 0.36 dedup: 29 records -> 23 indicators (6 merged, 20.7% duplication); 2 corroborated by 2+ feeds overlap: blocklist_de & feodo: 1 shared indicators overlap: blocklist_de & otx: 1 shared indicators overlap: blocklist_de & threatfox: 1 shared indicators overlap: feodo & otx: 1 shared indicators correlation: 2 relationships, 2 corroborated, 0 single-operator-only, 0 hosts with 2+ flagged URLs stored: 23 new, 0 updated, 26 new source links, 0 rescored repriced: 23 stored indicator(s) rescored for decay ``` *（真实的输出，通过在 pipeline 中回放已提交的测试 fixtures 并故意取消设置 AbuseIPDB 的 key 产生。各 feed 的耗时来自实时运行；fixture 回放会更快。）* 那个 `overlap` 区块是运行中最有趣的产物：它回答了“收集七个 feed 给我带来了七个 feed 价值的情报了吗？”在这里，收集的记录中有 20.7% 是重复的。 注意 `issues` 列：未解析的记录会被**计数并支持打印**（`--show-issues`），绝不悄悄丢弃。一个 connector 如果悄悄失效，正是这一列专门用来捕获的故障模式。 `check` 会标准化你从报告中粘贴的任何内容，无论是否经过防御性处理： ``` $ python main.py check "hxxps://Evil[.]example[.]com:443/Payload.EXE" "192[.]0[.]2[.]10" hxxps://Evil[.]example[.]com:443/Payload.EXE -> url https://evil.example.com/Payload.EXE 192[.]0[.]2[.]10 -> ipv4 192.0.2.10 ``` **退出代码：** `0` 成功 · `1` 无可报告内容（无匹配、空导出或 feed 错误） · `2` 使用错误。跳过*不是*错误 —— cron 作业不应该仅仅因为某个可选的 API key 未设置就呼叫任何人。 ## Dashboard `python main.py serve` —— 一个只读视图。它**不包含任何**情报逻辑：没有评分、没有去重、没有关联、没有收集。这一点在结构上由 [`test_web_layer_imports_no_intelligence_modules`](tests/test_web.py) 验证，该测试对 `web/app.py` 进行 AST 解析，如果它导入了 `core.scoring`、`core.dedup`、`core.correlation` 或 `feeds/` 中的任何内容，就会失败；配套测试断言从未调用任何写入方法。每个路由都仅限 GET。 两个视图： - **Index** —— 一个可过滤的 indicator 表（类型、feed、标签、最低置信度、时效性），带有显示分级分布和确认计数的统计卡片。 - **Detail** —— 单个 indicator 的完整评分推导，显示每个 feed 观测窗口及其自身声称的置信度的单来源溯源表，以及任何结构性关系。