avikengineer007/Honeypot-Threat-Intel_harvester

# Honeypot + Threat Intel Harvester 一个轻量级的 SSH/HTTP honeypot，能够捕获真实的攻击者交互，自动提取结构化的 IOC，对攻击者的 TTP 进行分类，并将其公开为可查询的 threat-intel feed —— 并提供了一个可选的集成 hook，可接入 [Sentrywatch](../sentrywatch/) 的 IP reputation 系统。 ## 架构 ``` ┌────────────────────┐ ┌─────────────────────┐ │ SSH Honeypot │ │ HTTP Honeypot │ │ (simulated shell, │ │ (bait endpoints, │ │ no real exec) │ │ common CVE paths) │ └────────┬───────────┘ └──────────┬────────────┘ │ │ └───────────┬───────────────┘ ▼ ┌──────────────────────┐ │ Session Store │ │ (SQLite, raw logs) │ └──────────┬───────────┘ ▼ ┌──────────────────────┐ │ IOC Extractor │ │ (deterministic │ │ regex/parsers) │ └──────────┬───────────┘ ▼ ┌──────────────────────┐ │ TTP Classifier │◄── Claude (advisory only, │ (rule-first) │ unclassified residue) └──────────┬───────────┘ ▼ ┌──────────────────────┐ │ Threat-Intel Store │ │ (IOCs + TTP labels) │ └────────┬─────────────┘ ┌─────────┴──────────┐ ▼ ▼ ┌─────────────────┐ ┌──────────────────┐ │ Export │ │ Sentrywatch Hook │ │ (JSON/CSV/STIX) │ │ (off by default) │ └─────────────────┘ └──────────────────┘ │ ▼ ┌─────────────────┐ │ Terminal UI │ │ (rich-based) │ └─────────────────┘ ``` ## 遏制优先设计 每一个设计决策都始于这样一个问题：*如果攻击者向我们发送了意外内容会怎样？* ### 零真实执行 SSH honeypot 提供了一个令人信服的伪造 shell —— 攻击者可以输入 `ls`、`cat /etc/passwd`、`wget http://...` —— 但**没有任何命令会被传递给宿主机操作系统**。`_execute_fake_cmd()` 调度器是一个纯 Python dict 查找，具有硬编码的字符串响应。没有 `os.system()`，没有 `subprocess`，也没有 `eval()`。`SandboxGuard.assert_zero_host_exec()` 方法的存在专门用于在测试中记录并验证这一不变性。 ### 边界处的输入清理 所有攻击者输入在存储之前都会通过 `SandboxGuard.sanitize_input()` 进行处理： - ANSI 转义序列被剥离（防止在 TUI 中发生终端注入） - 过滤不可打印字节（仅存储标准空格 + ASCII 32–126） - 输入在持久化之前被限制在安全长度内 ### 速率限制 `SandboxGuard.is_rate_limited()` 对每个 IP 实施滑动窗口连接限制（默认：50 个连接 / 60 秒）。攻击者无法以无限制的速率耗尽线程或填满数据库。 ### 审计完整性 每个会话、IOC 提取、分类和导出事件都会被写入一个 **SHA-256 哈希链审计日志**。可以使用 `python main.py audit-verify` 验证该链条。篡改任何存储的事件都会破坏链条。 ### 遏制测试 `tests/test_isolation.py` 显式地向 `_execute_fake_cmd()` 发射常见的命令注入和容器逃逸 payload，并断言它们返回安全的字符串 —— 永远不会触发真实的宿主机执行。 ## 确定性规则与 LLM 咨询 分类器遵循与 Sentrywatch 的 Decision Engine 相同的理念：**确定性规则始终优先运行；LLM 只处理规则无法解释的部分。** | TTP 类别 | 检测逻辑 | |---|---| | `credential_stuffing` | ≥3 个不同的 AUTH_ATTEMPT 事件 | | `known_cve_probe` | 请求路径/payload 匹配 CVE 特征库 | | `web_shell_attempt` | POST/PUT + 可执行 payload 特征（`eval(`、`base64_decode` 等） | | `recon_scan` | 已知的扫描器 User-Agents（nmap、masscan、nikto、zgrab 等） | | **`unclassified`** | 不匹配任何规则 → Claude 咨询二次处理 | `unclassified` 类别是一个**刻意的设计选择**。承认一个会话存在歧义，总比强行将其归入某个类别更正确。Claude 对这些会话的输出结果为： - 明确标记为 `method: llm_advisory`（在 TUI 中以斜体/暗色显示 —— 绝不看起来像规则结果） - 根据 LLM 自身的评估，以 `confidence: low/medium/high` 存储 - 在 Sentrywatch 推送阈值计算中，绝不视为权威数据 ## Sentrywatch 集成 可选的 Sentrywatch hook 会将高严重性的 IP 作为**预置的负面 reputation 条目**写入 Sentrywatch 的 `ip_reputation` 表中。关键设计属性如下： 1. **默认关闭** —— 配置中的 `sentrywatch_hook_enabled = False`。未经操作员意图，任何数据都不会跨越边界。 2. **阈值控制** —— 仅推送高于 `sentrywatch_severity_threshold`（默认：70/100）的 IP。 3. **仅限数据，不涉及决策** —— 该 hook 写入 reputation 分数。它从不调用 Sentrywatch 的 Decision Engine 或直接触发拦截。Sentrywatch 自身的确定性逻辑仍然控制着是否对 IP 进行拦截。 4. **经过审计** —— 每次推送都会连同 IP、分数和目标数据库路径一起记录在哈希链审计日志中。 这保留了一个原则，即**没有任何自动化系统会信任另一个自动化系统来触发强制执行** —— 数据在它们之间流动，但执行权限保持单一和确定性。 ## 快速开始 ### 安装 ``` pip install -r requirements.txt ``` ### 运行（带 Terminal UI） ``` python main.py run ``` 在 `:2222` 上启动 SSH honeypot，并在 `:8080` 上启动 HTTP honeypot。按 **1–6** 切换视图，按 **q** 退出。 ### 无界面运行（无 TUI） ``` python main.py run --no-tui ``` ### 查询捕获的数据 ``` # 列出 Top 攻击者 IP python main.py query # 查找特定 IP python main.py query --ip 192.168.1.100 ``` ### 导出 threat-intel feed ``` # JSON feed（默认） python main.py export --format json --output feed.json # CSV python main.py export --format csv --output feed.csv # STIX 2.1-lite bundle python main.py export --format stix --output feed.stix.json ``` ### 验证审计日志完整性 ``` python main.py audit-verify # → [OK] 审计日志 hash-chain 已验证（N 个事件验证完整）。 ``` ### 将高严重性 IP 推送到 Sentrywatch ``` # 必须先在 config 中启用 hook python main.py sentrywatch-push ``` ### 显示活动配置 ``` python main.py config ``` ## 配置 所有配置都位于 [`config.py`](config.py) 中，作为一个 Pydantic model，支持环境变量覆盖。 | 设置 | 默认值 | 描述 | |---|---|---| | `ssh_host` / `ssh_port` | `0.0.0.0:2222` | SSH honeypot 绑定地址 | | `http_host` / `http_port` | `0.0.0.0:8080` | HTTP honeypot 绑定地址 | | `ssh_banner` | OpenSSH 8.9p1 | 向扫描器展示的 SSH 版本字符串 | | `ssh_hostname` | `prod-db-master-01` | 伪造 shell 提示符中显示的主机名 | | `http_server_header` | nginx/1.18.0 | HTTP 响应中的 Server header | | `bait_endpoints` | 12 个常见路径 | 吸引探测的 CVE/CMS 路径 | | `cve_signatures` | 8 个 CVE 模式 | 确定性 CVE 检测规则 | | `sentrywatch_hook_enabled` | `False` | 必须显式启用 | | `sentrywatch_severity_threshold` | `70` | Sentrywatch 推送的分数阈值 | | `anthropic_api_key` | `$ANTHROPIC_API_KEY` | Claude API key（可选） | ## Terminal UI 视图 | 按键 | 视图 | 描述 | |---|---|---| | `1` | **实时仪表板** | 实时会话 feed、热门 IP、服务状态栏 | | `2` | **会话详情** | 最近会话的完整记录、IOC、TTP 分类 | | `3` | **Threat-Intel 浏览器** | IP 摘要表格 + IOC artifact feed | | `4` | **TTP 分解** | 带有 ASCII 条形图的 TTP 聚合分布 | | `5` | **审计日志** | 哈希链验证状态 + 事件表 | | `6` | **配置** | 活动配置和诱骗 endpoint 列表（只读） | LLM 咨询分类的会话始终以**暗色斜体黄色**显示，以区别于粗体绿色的确定性规则结果。 ## 运行测试 ``` pytest tests/ -v ``` 测试覆盖率包括： - `test_isolation.py` —— 遏制：针对伪造 shell 的命令注入 payload，ANSI 清理 - `test_ioc_ttp.py` —— 所有四个确定性 TTP 规则（credential stuffing、CVE probe、web shell、recon scan）+ IOC 提取 - `test_severity.py` —— 评分公式边界情况、TTP 危险排序、上限、边际效益递减 - `test_export_audit.py` —— STIX/CSV 导出验证、Sentrywatch DB 集成、哈希链篡改检测 ## 数据库 Schema 位于 `data/honeypot.db` 的 SQLite 数据库包含五个表：`sessions`、`iocs`、`ttp_classifications`、`ip_intel_summary`、`audit_log`。完整 schema 请参见 [TRD §5](../docs/)。 ## 项目结构 ``` Honeypot+Threat Intel Harvester/ ├── main.py # CLI entry point (run / query / export / audit-verify) ├── config.py # Pydantic config model ├── requirements.txt ├── conftest.py # pytest sys.path setup ├── core/ │ ├── models.py # Pydantic data models & enums │ ├── database.py # SQLite CRUD layer (DatabaseManager) │ ├── audit.py # SHA-256 hash-chained audit logger │ └── isolation.py # SandboxGuard (rate limiting, input sanitization) ├── services/ │ ├── manager.py # HoneypotManager (starts/stops both services) │ ├── ssh_honeypot.py # Paramiko-based SSH honeypot with fake shell │ └── http_honeypot.py # aiohttp-based HTTP honeypot with bait endpoints ├── analytics/ │ ├── ioc_extractor.py # Regex-based IOC extraction │ ├── ttp_classifier.py # Rule-first classifier with Claude fallback │ └── severity.py # Deterministic 0-100 severity score calculator ├── feed/ │ ├── exporter.py # JSON / CSV / STIX-lite export │ └── sentrywatch.py # Optional push to Sentrywatch ip_reputation ├── ui/ │ ├── app.py # HoneypotTUIApp (Rich Live + keyboard navigation) │ └── views/ │ ├── live_dashboard.py # View 1: real-time session + IP feed │ ├── session_detail.py # View 2: full session deep-dive │ ├── intel_browser.py # View 3: IP summary + IOC artifact table │ ├── ttp_breakdown.py # View 4: TTP distribution metrics │ ├── audit_viewer.py # View 5: audit log + chain verification │ └── config_view.py # View 6: read-only config display ├── tests/ │ ├── test_isolation.py │ ├── test_ioc_ttp.py │ ├── test_severity.py │ └── test_export_audit.py └── data/ └── honeypot.db # SQLite store (auto-created on first run) ```

标签：CISA项目, Python, SQLite, 威胁情报, 安全数据可视化, 开发者工具, 无后门, 蜜罐, 证书利用, 逆向工具