04arvind/Network-Traffic-Analyzer

# 网络流量分析器 基于规则的本地抓包与检测演示 (FastAPI + Scapy)。 ## 架构 (工作原理) - **数据包捕获** (`app/capture.py`)：Scapy 从选定的接口嗅探 TCP/UDP 数据包。 - **解析**：每个数据包都被转换为结构化的 `PacketEvent` (Pydantic model)。 - **规则** (`app/analyzer.py`)：透明的规则引擎会评估端口扫描、SYN flood 和敏感端口访问。 - **存储** (`app/storage.py`)：数据包和告警被写入本地 SQLite DB (`data/traffic.db`)。 - **交付** (`app/main.py` + `app/realtime.py`)： - REST：`/stats`, `/alerts`, `/demo/scan` - WebSocket：`/traffic/live` 将实时数据包 + 告警流式传输到 UI。 - **UI** (`app/templates/index.html`, `app/static/*`)：显示滚动统计数据、实时表格和告警。 ## 功能说明 - 捕获 **仅限本地** 的流量 (如果可用，首选 loopback)。 - 将 **IP** (IPv4/IPv6)、**TCP** 和 **UDP** 头解析为结构化事件。 - 应用 **可解释的规则**： - 端口扫描 (时间窗口内每个源 IP 的不同目标端口数) - SYN flood (时间窗口内每个源 IP 的纯 SYN 数据包计数) - 异常/敏感端口访问 (被标记为低严重级别的关注端口) - 将所有内容持久化到 **SQLite**，并将检测结果流式传输到浏览器。 ## 检测逻辑 (透明、基于规则) ### 1) 端口扫描 (`rule="port_scan"`，严重级别 `high`) 触发条件： - 对于给定的 `source_ip`，引擎观察到 **≥ `N` 个不同的目标端口** - 在 **`port_scan_window_seconds`** 秒内。 `app/config.py` 中的默认值： - `NTA_PORT_SCAN_THRESHOLD` (默认 **10**) - `NTA_PORT_SCAN_WINDOW_SECONDS` (默认 **10**) 告警中记录的证据： - `distinct_ports` (排序后的列表) - `window_seconds` 冷却时间： - 告警通过 `NTA_ALERT_COOLDOWN_SECONDS` (默认 **30**) 针对每个 `(rule, source_ip)` 进行速率限制。 ### 2) SYN flood (`rule="syn_flood"`，严重级别 `critical`) 触发条件： - 对于给定的 `source_ip`，引擎观察到 **≥ `syn_flood_threshold` 个 SYN 数据包** - 在 **`syn_flood_window_seconds`** 秒内。 重要细节： - 仅统计 TCP 标志包含 **SYN** 且 **不包含 ACK** 的数据包 (纯 SYN)。 默认值： - `NTA_SYN_FLOOD_THRESHOLD` (默认 **20**) - `NTA_SYN_FLOOD_WINDOW_SECONDS` (默认 **5**) 记录的证据： - `syn_count` - `window_seconds` 冷却时间： - 使用 `NTA_ALERT_COOLDOWN_SECONDS` (默认 **30**) 对每个 `(rule, source_ip`) 采用相同的冷却机制。 ### 3) 异常/敏感端口访问 (`rule="unusual_port"`，严重级别 `low`) 触发条件： - 数据包目标为以下任何目标端口： - **23**, **3389**, **445**, **5900**, **3306**, **1433** 默认值：在 `app/analyzer.py` 中硬编码的集合。 记录的证据： - `port` - `protocol` 注意： - 此规则不受冷却逻辑限制 (每个匹配的数据包都会生成一个告警)。 ## 配置 (环境变量) 这些配置由 `app/config.py` 通过 `os.getenv("NTA_...")` 读取。在面试和演示期间非常有用。 - `NTA_INTERFACE` (可选)：用于嗅探的接口名称 (例如 `lo`, `Ethernet`, `Npcap Loopback Adapter`)。 - 如果未设置，应用会尝试自动选择 loopback 接口，否则回退到第一个可用的接口。 - `NTA_DB_PATH` (默认 `data/traffic.db`)：SQLite DB 的存储位置。 - `NTA_BPF` (默认 `tcp or udp`)：Scapy 捕获过滤器 (在 `app/capture.py` 中作为 `bpf_filter` / BPF 传递给嗅探器)。 - `NTA_RETENTION_SECONDS` (默认 `3600`)：配置的保留窗口。 - **注意：** 当前代码库在 `TrafficStore` 中包含了 `purge_older_than(...)`，但并未将定期清理接入运行中的应用。 - `NTA_PORT_SCAN_THRESHOLD` (默认 `10`) - `NTA_PORT_SCAN_WINDOW_SECONDS` (默认 `10`) - `NTA_SYN_FLOOD_THRESHOLD` (默认 `20`) - `NTA_SYN_FLOOD_WINDOW_SECONDS` (默认 `5`) - `NTA_ALERT_COOLDOWN_SECONDS` (默认 `30`) ## 前置条件 - Python **3.11+** - `requirements.txt` / `pyproject.toml` 中列出的依赖包 - **Windows 注意事项 (Scapy)：** - 使用 Scapy 进行数据包捕获/发送通常需要 **Npcap**。 - 可能需要提升的权限。 ## 在本地运行 ### 1) 安装 如果可用，使用 `uv`： ``` uv sync ``` 或者使用 pip： ``` pip install -r requirements.txt ``` ### 2) 启动服务器 ``` uv run python main.py ``` 这会通过 `uvicorn` 在端口 **8000** 上运行 `app.main:app`。 ### 3) 打开仪表盘 - UI：**http://127.0.0.1:8000** - 实时数据流：WebSocket 地址 **/traffic/live** ## 使用虚假攻击流量进行演示 (推荐) 模拟器支持两种模式： - `demo` (默认)：调用 `POST /demo/scan` 将合成的 `PacketEvent` 对象注入后端。 - **不**需要原始数据包发送权限。 - `raw`：使用 Scapy 发送真实构造的 SYN 数据包。 - 需要 Scapy/Npcap 权限。 - `both`：同时执行发送原始 SYN 数据包和注入 demo API。 ### 示例 (默认 demo 注入) ``` python simulate_attack.py --target 127.0.0.1 --start-port 20 --end-port 60 ``` ### 示例 (原始构造的 SYN 数据包) ``` python simulate_attack.py --target 127.0.0.1 --start-port 20 --end-port 60 --mode raw ``` ### 示例 (原始 + demo) ``` python simulate_attack.py --target 127.0.0.1 --start-port 20 --end-port 60 --mode both ``` 有用的参数： - `--delay` (默认 `0.05`)：数据包发送间的休眠时间 - `--iface`：覆盖 Scapy 接口选择 - `--api-url` (默认 `http://127.0.0.1:8000/demo/scan`) ## REST API ### `GET /stats` 返回用于填充仪表盘的聚合统计数据： - `total_packets` - `total_alerts` - `packets_per_second` - `top_talkers` (按数据包数量和字节数统计的顶级源 IP) - `protocol_breakdown` - `recent_packets` (PacketEvent 列表) - `recent_alerts` (AlertEvent 列表) ### `GET /alerts?limit=50` 返回最新的告警 (按时间戳降序排列)。 响应结构： - `{ "items": [ ... ] }` ### `POST /demo/scan` 将合成数据包注入后端，以便在没有原始捕获的情况下触发规则。 Payload 模型 (`DemoScanRequest`)： - `source_ip` (默认 `10.0.0.5`) - `target_ip` (默认 `127.0.0.1`) - `start_port` (默认 `20`) - `end_port` (默认 `60`) - `delay` (默认 `0.0`) ## WebSocket API (实时流) ### `WS /traffic/live` - 发送初始消息： - `{ "type": "snapshot", "payload": }` - 然后流式传输增量消息： - `{ "type": "packet", "payload": }` - `{ "type": "alert", "payload": }` 页面加载时 UI 会自动订阅。 ## 数据库 / 存储细节 (SQLite) DB 位置： - 默认为 `data/traffic.db` (可以通过 `NTA_DB_PATH` 覆盖)。 表： ### `packets` - `id` (自增) - `ts` (实数 / epoch 秒) - `iso_ts` (ISO 时间戳字符串) - `src_ip`, `dst_ip` - `src_port`, `dst_port` - `protocol` (`TCP`, `UDP` 或 `OTHER`) - `tcp_flags` - `packet_size` - `summary` (packet.summary()) - `interface` ### `alerts` - `id` (自增) - `ts` (实数 / epoch 秒) - `iso_ts` - `rule` (例如 `port_scan`, `syn_flood`, `unusual_port`) - `severity` (`low`, `high`, `critical`) - `source_ip`, `destination_ip`, `destination_port` - `message` - `evidence_json` (JSON 字符串；加载回 `AlertEvent.evidence`) ## 捕获注意事项 / 故障排除 - 仅嗅探您拥有/控制的接口。 - **Windows + Scapy：** - 安装 **Npcap**。 - 如果捕获/发送失败，请以管理员身份运行终端。 - 接口选择： - 如果未设置 `NTA_INTERFACE`，应用会优先选择类似 loopback 的接口；否则使用第一个可用的接口。 - 实时仪表盘： - 如果 WebSocket 断开连接，UI 会显示 **Offline** 并停止实时更新。 - 数据保留： - 已配置 `NTA_RETENTION_SECONDS`，但自动清理目前并未作为运行时的一部分执行。

标签：AV绕过, FastAPI, Scapy, SQLite, WebSocket, 依赖分析, 插件系统, 网络流量分析, 逆向工具