CYBERLOGS1/threat-intelligence-service
GitHub: CYBERLOGS1/threat-intelligence-service
一个轻量级个人威胁情报 pipeline，聚合多个免费 IOC 源到本地 SQLite，并提供 CLI 查询、终端/Web 仪表板和 HTTP API 供其他安全工具联动调用。
Stars: 0 | Forks: 0
# 威胁情报服务
## STREAMLIT 仪表板
## 基于 RICH 的终端仪表板
一个个人威胁情报 pipeline：拉取免费的 IOC 源，将其存储在 SQLite 中，并让你（或你的其他工具 —— `net_guard.py`、`IDS_GUARD.py`、`Ad_Blocker.py`）根据这些数据检查 IP/域名/URL/哈希值。包含一个实时的 Rich 终端仪表板和一个小型的本地 HTTP API。
## 设置
```
pip install -r requirements.txt
```
默认的源不需要 API key。
## 包含的源（免费，无需 key）
| 源 | 提供的内容 |
|---|---|
| **URLhaus** (Abuse.ch) | 最近的恶意 URL |
| **ThreatFox** (Abuse.ch) | 混合 IOC：与特定恶意软件相关的 IP、域名、URL、哈希值 |
| **Feodo Tracker** (Abuse.ch) | 活跃的 botnet C2 IP |
| **Spamhaus DROP** | 被劫持/恶意的 IP 网段（CIDR 范围） |
编辑 `config.py` 以禁用源或添加你自己的源（例如 AlienVault OTX、AbuseIPDB、VirusTotal —— 这些需要 API key，因此你需要按照现有的模式在 `core/feeds.py` 中添加一个新的 fetcher）。
## 用法
**拉取源并填充数据库：**
```
python ti_service.py update
python ti_service.py update --feed urlhaus # just one feed
```
定期运行此命令（cron / Windows 任务计划程序）—— 例如每小时一次 —— 以保持数据库最新。
**检查单个指示器：**
```
python ti_service.py check 185.220.101.5
python ti_service.py check evil-domain.example
```
IP 也会根据存储的 CIDR 范围（例如 Spamhaus DROP）进行检查，因此如果某个 IP 落在已知的恶意网段内，则无需单独列出即可被捕获。
**搜索指示器（子字符串匹配）：**
```
python ti_service.py search malware-drop
```
**一次性摘要：**
```
python ti_service.py stats
```
**实时自动刷新仪表板（终端）：**
```
python ti_service.py dashboard
python ti_service.py dashboard --interval 10
```
**Web 仪表板 (Streamlit)：**
```
streamlit run streamlit_app.py
```
打开一个浏览器标签页，包含：
- KPI 卡片（总指示器数、IP/CIDR、URL/域名、哈希值）
- 按类型和来源划分的柱状图
- 源运行健康状况表
- 一个可搜索/可过滤的指示器表（按类型、来源或文本搜索进行过滤），支持导出为 CSV
- 一个侧边栏的“立即拉取”按钮，无需触碰 CLI 即可触发源更新
- 一个侧边栏的快速 IOC 检查器
它读取与 CLI 相同的 `threat_intel.db`，因此你可以在 cron 作业上运行 `ti_service.py update`，然后只需在此处浏览结果 —— 数据每 15 秒自动刷新一次，或者通过侧边栏的“刷新视图”按钮立即刷新。
**运行查询 API**（以便其他工具/脚本可以通过 HTTP 检查 IOC）：
```
python ti_service.py serve
# 那么：
curl "http://127.0.0.1:8787/check?value=185.220.101.5"
curl "http://127.0.0.1:8787/search?q=example.com"
curl "http://127.0.0.1:8787/stats"
```
## 将其接入你的其他工具
在 `net_guard.py`、`IDS_GUARD.py` 或 `Ad_Blocker.py` 中，你可以：
1. 使用 `requests.get("http://127.0.0.1:8787/check?value=" + ip)` **查询正在运行的 API**
2. **直接导入**（如果在同一台机器上运行，则不需要 HTTP）：
```
from core.database import TIDatabase
from core.checker import verdict
import config
db = TIDatabase(config.DB_PATH)
result = verdict(some_ip_or_domain, db)
if result["malicious"]:
print(f"Blocked: {result['matches']}")
```
3. **将你自己的检测结果反馈回去** —— 当 IDS_GUARD 或 net_guard 标记某些内容时，将其作为第一方指示器插入：
```
db.upsert_indicator(
value=flagged_ip,
ioc_type="ip",
source="ids_guard", # mark it as your own detection
threat_type="port_scan",
confidence=70,
)
```
这会将你的三个工具变成一个互联的防御 stack：每一个工具的发现都会充实其他工具查询的共享情报库。
## 项目结构
```
threat_intel_service/
├── ti_service.py # CLI entry point (update/check/search/stats/dashboard/serve)
├── streamlit_app.py # Web dashboard (charts, filterable table, quick check)
├── config.py # feed list, DB path, API settings
├── core/
│ ├── database.py # SQLite storage layer
│ ├── feeds.py # feed fetchers + parsers
│ ├── checker.py # IOC classification + lookup/verdict logic
│ ├── dashboard.py # Rich terminal dashboard rendering
│ └── api.py # stdlib HTTP query API
└── requirements.txt
```
## 备注
- 所有解析器均采用软失败机制：如果 abuse.ch 调整了列布局，格式错误的行将被跳过并记录日志，而不会导致整个摄取运行崩溃。
- 数据库根据 `(value, ioc_type, source)` 进行去重，因此重新运行 `update` 只会刷新 `last_seen`/置信度，而不会创建重复项。
- `threat_intel.db` 会在首次运行时在 `ti_service.py` 旁边创建。
一个个人威胁情报 pipeline：拉取免费的 IOC 源，将其存储在 SQLite 中，并让你（或你的其他工具 —— `net_guard.py`、`IDS_GUARD.py`、`Ad_Blocker.py`）根据这些数据检查 IP/域名/URL/哈希值。包含一个实时的 Rich 终端仪表板和一个小型的本地 HTTP API。
## 设置
```
pip install -r requirements.txt
```
默认的源不需要 API key。
## 包含的源（免费，无需 key）
| 源 | 提供的内容 |
|---|---|
| **URLhaus** (Abuse.ch) | 最近的恶意 URL |
| **ThreatFox** (Abuse.ch) | 混合 IOC：与特定恶意软件相关的 IP、域名、URL、哈希值 |
| **Feodo Tracker** (Abuse.ch) | 活跃的 botnet C2 IP |
| **Spamhaus DROP** | 被劫持/恶意的 IP 网段（CIDR 范围） |
编辑 `config.py` 以禁用源或添加你自己的源（例如 AlienVault OTX、AbuseIPDB、VirusTotal —— 这些需要 API key，因此你需要按照现有的模式在 `core/feeds.py` 中添加一个新的 fetcher）。
## 用法
**拉取源并填充数据库：**
```
python ti_service.py update
python ti_service.py update --feed urlhaus # just one feed
```
定期运行此命令（cron / Windows 任务计划程序）—— 例如每小时一次 —— 以保持数据库最新。
**检查单个指示器：**
```
python ti_service.py check 185.220.101.5
python ti_service.py check evil-domain.example
```
IP 也会根据存储的 CIDR 范围（例如 Spamhaus DROP）进行检查，因此如果某个 IP 落在已知的恶意网段内，则无需单独列出即可被捕获。
**搜索指示器（子字符串匹配）：**
```
python ti_service.py search malware-drop
```
**一次性摘要：**
```
python ti_service.py stats
```
**实时自动刷新仪表板（终端）：**
```
python ti_service.py dashboard
python ti_service.py dashboard --interval 10
```
**Web 仪表板 (Streamlit)：**
```
streamlit run streamlit_app.py
```
打开一个浏览器标签页，包含：
- KPI 卡片（总指示器数、IP/CIDR、URL/域名、哈希值）
- 按类型和来源划分的柱状图
- 源运行健康状况表
- 一个可搜索/可过滤的指示器表（按类型、来源或文本搜索进行过滤），支持导出为 CSV
- 一个侧边栏的“立即拉取”按钮，无需触碰 CLI 即可触发源更新
- 一个侧边栏的快速 IOC 检查器
它读取与 CLI 相同的 `threat_intel.db`，因此你可以在 cron 作业上运行 `ti_service.py update`，然后只需在此处浏览结果 —— 数据每 15 秒自动刷新一次，或者通过侧边栏的“刷新视图”按钮立即刷新。
**运行查询 API**（以便其他工具/脚本可以通过 HTTP 检查 IOC）：
```
python ti_service.py serve
# 那么：
curl "http://127.0.0.1:8787/check?value=185.220.101.5"
curl "http://127.0.0.1:8787/search?q=example.com"
curl "http://127.0.0.1:8787/stats"
```
## 将其接入你的其他工具
在 `net_guard.py`、`IDS_GUARD.py` 或 `Ad_Blocker.py` 中，你可以：
1. 使用 `requests.get("http://127.0.0.1:8787/check?value=" + ip)` **查询正在运行的 API**
2. **直接导入**（如果在同一台机器上运行，则不需要 HTTP）：
```
from core.database import TIDatabase
from core.checker import verdict
import config
db = TIDatabase(config.DB_PATH)
result = verdict(some_ip_or_domain, db)
if result["malicious"]:
print(f"Blocked: {result['matches']}")
```
3. **将你自己的检测结果反馈回去** —— 当 IDS_GUARD 或 net_guard 标记某些内容时，将其作为第一方指示器插入：
```
db.upsert_indicator(
value=flagged_ip,
ioc_type="ip",
source="ids_guard", # mark it as your own detection
threat_type="port_scan",
confidence=70,
)
```
这会将你的三个工具变成一个互联的防御 stack：每一个工具的发现都会充实其他工具查询的共享情报库。
## 项目结构
```
threat_intel_service/
├── ti_service.py # CLI entry point (update/check/search/stats/dashboard/serve)
├── streamlit_app.py # Web dashboard (charts, filterable table, quick check)
├── config.py # feed list, DB path, API settings
├── core/
│ ├── database.py # SQLite storage layer
│ ├── feeds.py # feed fetchers + parsers
│ ├── checker.py # IOC classification + lookup/verdict logic
│ ├── dashboard.py # Rich terminal dashboard rendering
│ └── api.py # stdlib HTTP query API
└── requirements.txt
```
## 备注
- 所有解析器均采用软失败机制：如果 abuse.ch 调整了列布局，格式错误的行将被跳过并记录日志，而不会导致整个摄取运行崩溃。
- 数据库根据 `(value, ioc_type, source)` 进行去重，因此重新运行 `update` 只会刷新 `last_seen`/置信度，而不会创建重复项。
- `threat_intel.db` 会在首次运行时在 `ti_service.py` 旁边创建。标签：IP 地址批量处理, Kubernetes, Python, Python脚本, SQLite, 威胁情报, 开发者工具, 无后门, 本地API, 逆向工具, 配置审计