# AttackFlow

**Repository Path**: didiplus/AttackFlow

## Basic Information

- **Project Name**: AttackFlow
- **Description**: AttackFlow 实时采集 SSH 认证失败日志，解析攻击源 IP，通过 IP 地理定位在世界地图上以飞线图动态展示攻击来源与目标服务器的关系，同时将攻击事件持久化到 ClickHouse，并提供告警引擎与多服务器节点支持。
- **Primary Language**: Python
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 1
- **Forks**: 0
- **Created**: 2026-08-20
- **Last Updated**: 2026-08-22

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# AttackFlow

> SSH 攻击实时监控与飞线图可视化系统

AttackFlow 实时采集 SSH 认证失败日志，解析攻击源 IP，通过 IP 地理定位在世界地图上以飞线图动态展示攻击来源与目标服务器的关系，同时将攻击事件持久化到 ClickHouse，并提供告警引擎与多服务器节点支持。

---
![](images/login.png)
![](images/index.png)
## 目录

- [功能特性](#功能特性)
- [系统架构](#系统架构)
- [项目结构](#项目结构)
- [快速部署（Docker Compose）](#快速部署docker-compose)
- [手动部署](#手动部署)
- [配置说明](#配置说明)
- [API 接口](#api-接口)
- [安全特性](#安全特性)
- [多服务器部署](#多服务器部署)
- [数据清除](#数据清除)
- [开发指南](#开发指南)

---

## 功能特性

| 模块 | 能力 |
|------|------|
| 实时采集 | tail 方式监控 `/var/log/auth.log`，正则匹配多种 SSH 攻击模式（Failed password / Invalid user / Connection closed / Disconnected / authentication failure） |
| IP 定位 | 调用 ipinfo.io 获取攻击源地理位置，Redis 缓存 1 小时，私有 IP 自动识别 |
| 自动定位 | Agent 启动时通过公网 IP 自动获取本机坐标，无需手动配置（可关闭） |
| 飞线图可视化 | ECharts 世界地图，攻击源 → 目标服务器动态飞线 + 涟漪散点，支持缩放漫游 |
| 实时推送 | WebSocket 长连接，Redis Pub/Sub 跨实例广播，自动指数退避重连 |
| 持久化 | ClickHouse MergeTree 引擎，按月分区，90 天 TTL 自动过期 |
| 告警引擎 | 滑动窗口规则（N 秒内 >= M 次触发），冷却去重，支持 Webhook / Telegram 多通道推送 |
| 多服务器 | Agent 注册表 + 30s 心跳 + 90s 过期，前端多节点 scatter 渲染 + 服务器列表卡片 |
| 时间范围切换 | 实时模式 / 24 小时历史模式一键切换，历史模式查询 ClickHouse |
| 安全认证 | bcrypt 密码哈希 + JWT 会话 + CSRF Token + 登录速率限制 + 图形验证码 |
| 退出确认 | 自定义模态框确认退出，替代原生 confirm，与主题风格一致 |
| 双主题 | Linear Aesthetic 暗黑 / 明亮主题，玻璃拟态 + 微光呼吸动画，localStorage 持久化 |

---

## 系统架构

![](images/17872239407441.png)

**数据流：**

1. Agent tail 日志 → 正则提取 `(时间戳, IP)` → ipinfo.io 定位 → 组装攻击事件（含 `coords=[经度,纬度]`、`target_coords`）
2. 事件写入 Redis Stream + 最近列表（sorted set）+ Pub/Sub 广播
3. 事件同步写入 ClickHouse（失败不阻断采集）
4. FastAPI WebSocket 订阅 Pub/Sub，实时推送给前端
5. 前端 ECharts 渲染飞线图（攻击源 → 目标服务器）+ 涟漪散点 + 服务器节点
6. 告警引擎独立订阅 Stream，滑动窗口判定后推送 Webhook / Telegram

---

## 项目结构

```
AttackFlow/
├── app/                        # 主应用包（Web API + 告警）
│   ├── main.py                 # FastAPI 入口 + lifespan + /healthz
│   ├── config.py               # 环境变量配置 + load_dotenv + 容错解析
│   ├── models.py               # Pydantic 数据模型
│   ├── captcha.py              # 验证码图片生成
│   ├── clean_data.py           # 数据清除脚本（Redis+ClickHouse）
│   ├── storage/
│   │   ├── redis_store.py      # Redis: 会话/验证码/IP缓存/Stream/Pub-Sub/限流/服务器注册表
│   │   └── clickhouse_store.py # ClickHouse: 建表/插入/查询/计数
│   ├── auth/
│   │   ├── security.py         # bcrypt + JWT + CSRF
│   │   └── rate_limit.py       # 登录速率限制
│   ├── api/
│   │   ├── routes.py           # REST: 登录/验证码/攻击查询/统计/服务器列表
│   │   └── websocket.py        # WebSocket: 订阅 Pub/Sub 跨实例广播
│   └── alert/
│       └── engine.py           # 告警引擎: 滑动窗口 + 多通道推送
├── agent/                      # 独立采集 Agent（可单独部署到被监控服务器）
│   ├── collector.py            # 采集主入口: 日志解析+IP定位+注册心跳+自动定位
│   ├── config.py               # Agent 专属配置（不依赖 app 包）
│   ├── storage.py              # 精简 Redis + ClickHouse 客户端
│   ├── pyproject.toml          # 独立依赖（仅 16 个包）
│   ├── Dockerfile              # 独立构建镜像
│   └── .env.example            # 环境变量示例
├── templates/
│   ├── index.html              # 主页（飞线图+攻击播报+统计+服务器列表+退出模态框）
│   └── login.html              # 登录页（玻璃拟态 + CSRF）
├── static/
│   ├── css/index.css           # Linear Aesthetic 主题系统
│   └── js/index.js             # 主题切换+ECharts+WS重连+多服务器+时间范围切换
├── images/                     # 项目截图
├── pyproject.toml              # uv 项目配置（package=false）
├── uv.lock                     # 依赖锁文件
├── requirements.txt            # pip 兼容依赖（uv export 生成）
├── docker-compose.yml          # Redis+ClickHouse+API+Agent+Alert 编排
├── Dockerfile                  # 主应用构建（API + Alert）
├── .env / .env.example         # 环境变量配置
└── start.sh                    # 三进程启动脚本
```

---

## 快速部署（Docker Compose）

### 前置条件

- Docker 20.10+
- Docker Compose v2
- 宿主机存在 SSH 日志（Linux 默认 `/var/log/auth.log`）

### 步骤

1. **克隆并配置**

   ```bash
   git clone <repo-url> AttackFlow
   cd AttackFlow
   cp .env.example .env
   ```

2. **编辑 `.env`**（至少修改以下项）

   ```dotenv
   JWT_SECRET=<生成一个随机字符串>
   ATTACKFLOW_USER=<你的用户名>
   ATTACKFLOW_PASS=<你的密码>
   SERVER_LON=<服务器经度>    # 例如 -77.0369（华盛顿）
   SERVER_LAT=<服务器纬度>    # 例如 38.9072
   ```

   > **坐标顺序非常重要**：`SERVER_LON` 是经度，`SERVER_LAT` 是纬度。填反会导致飞线图终点位置错乱。

3. **启动全部服务**

   ```bash
   docker compose up -d --build
   ```

   将启动 5 个服务：

   | 服务 | 说明 | 端口 |
   |------|------|------|
   | redis | Redis 7（持久化） | 6379 |
   | clickhouse | ClickHouse 24 | 8123 (HTTP), 9000 (TCP) |
   | api | FastAPI 服务 | 8000 |
   | agent | 日志采集 Agent | - |
   | alert | 告警引擎 | - |

4. **访问**

   打开 `http://localhost:8000`，使用 `.env` 中配置的账号密码登录。

5. **查看日志 / 停止**

   ```bash
   docker compose logs -f api        # 查看 API 日志
   docker compose logs -f agent      # 查看采集日志
   docker compose down               # 停止全部服务
   ```

---

## 手动部署

适用于开发环境或无法使用 Docker 的场景。

### 前置条件

- Python >= 3.10
- [uv](https://docs.astral.sh/uv/) 包管理器
- Redis 6+（本地或云服务如 Upstash）
- ClickHouse 22+（本地 Docker 或云服务）

### 安装 uv

```bash
curl -LsSf https://astral.sh/uv/install.sh | sh
```

### 安装依赖

```bash
cd AttackFlow
uv sync
```

### 配置环境变量

```bash
cp .env.example .env
# 编辑 .env，填入 Redis/ClickHouse 连接信息、认证信息、服务器坐标
```

### 启动 Redis / ClickHouse（如需本地）

```bash
docker run -d --name redis -p 6379:6379 redis:7-alpine
docker run -d --name clickhouse -p 8123:8123 -p 9000:9000 clickhouse/clickhouse-server:24-alpine
```

> ClickHouse 使用原生 TCP 协议端口 **9000**（非 HTTP 8123）。

### 一键启动三进程

```bash
chmod +x start.sh
./start.sh
```

将启动 API（:8000）+ Agent + Alert 三个进程。

### 或分别启动（开发模式）

```bash
# 终端 1：API 服务
uv run uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload

# 终端 2：采集 Agent（独立项目，在 agent/ 目录运行）
cd agent && uv sync && uv run python collector.py

# 终端 3：告警引擎
uv run python -m app.alert.engine
```

---

## 配置说明

所有配置通过环境变量（`.env` 文件）管理，`python-dotenv` 自动加载。

| 变量 | 默认值 | 说明 |
|------|--------|------|
| `REDIS_URL` | `redis://localhost:6379/0` | Redis 连接地址。云服务用 `rediss://` 前缀 |
| `CLICKHOUSE_HOST` | `localhost` | ClickHouse 主机 |
| `CLICKHOUSE_PORT` | `9000` | ClickHouse **TCP** 端口（非 8123） |
| `CLICKHOUSE_USER` | `default` | ClickHouse 用户名 |
| `CLICKHOUSE_PASSWORD` | _(空)_ | ClickHouse 密码 |
| `CLICKHOUSE_DATABASE` | `attackflow` | ClickHouse 数据库名 |
| `JWT_SECRET` | `change-me-in-production` | JWT 签名密钥，**生产必须修改** |
| `TOKEN_EXPIRE_MINUTES` | `30` | JWT 会话有效期（分钟） |
| `ATTACKFLOW_USER` | `didiplus` | 管理员用户名 |
| `ATTACKFLOW_PASS` | _(空)_ | 管理员密码（bcrypt 哈希存储） |
| `LOGIN_MAX_ATTEMPTS` | `5` | 登录最大尝试次数 |
| `LOGIN_WINDOW_SECONDS` | `60` | 登录速率限制窗口（秒） |
| `ALERT_WINDOW_SECONDS` | `60` | 告警滑动窗口（秒） |
| `ALERT_THRESHOLD` | `10` | 窗口内触发告警的攻击次数 |
| `ALERT_COOLDOWN_SECONDS` | `300` | 告警冷却时间（秒） |
| `PUSH_CHANNELS` | `[]` | 推送通道 JSON 数组，见下方格式 |
| `LOG_FILE_PATH` | `/var/log/auth.log` | SSH 日志路径 |
| `IPINFO_TOKEN` | _(空)_ | ipinfo.io API Token（免费额度无需） |
| `SERVER_LOCATION_AUTO` | `1` | 自动获取服务器坐标（通过公网 IP 定位），关闭后使用手动配置 |
| `SERVER_LON` | `113.2500` | 服务器**经度**（飞线图终点，自动定位关闭时生效） |
| `SERVER_LAT` | `23.1167` | 服务器**纬度**（飞线图终点，自动定位关闭时生效） |

### 推送通道配置

`PUSH_CHANNELS` 为 JSON 数组，支持多通道：

```json
[
  {"type": "webhook", "url": "https://example.com/hook"},
  {"type": "telegram", "token": "bot-token", "chat_id": "123456"}
]
```

---

## API 接口

所有 `/api/*` 接口需登录认证（JWT Cookie）。

| 方法 | 路径 | 说明 |
|------|------|------|
| `GET` | `/login` | 登录页面 |
| `POST` | `/login` | 登录处理（表单：username, password, captcha, csrf） |
| `GET` | `/logout` | 登出 |
| `GET` | `/captcha` | 验证码图片（PNG） |
| `GET` | `/` | 主页（需认证） |
| `GET` | `/healthz` | 健康检查 |
| `GET` | `/api/attacks?limit=50&hours=24` | 攻击数据。`hours` 指定时查 ClickHouse 历史，否则查 Redis 最近列表 |
| `GET` | `/api/stats` | 统计信息（总攻击数、最近 1 小时、国家数） |
| `GET` | `/api/servers` | 服务器节点列表（多服务器注册表） |
| `WS` | `/ws/attacks` | WebSocket 实时攻击推送 |

### 示例

```bash
# 获取最近 20 条攻击
curl -b cookies.txt http://localhost:8000/api/attacks?limit=20

# 获取最近 24 小时历史（ClickHouse）
curl -b cookies.txt "http://localhost:8000/api/attacks?hours=24&limit=100"

# 获取统计
curl -b cookies.txt http://localhost:8000/api/stats

# 获取服务器节点
curl -b cookies.txt http://localhost:8000/api/servers
```

---

## 安全特性

| 特性 | 实现 |
|------|------|
| 密码存储 | bcrypt（rounds=12），不存明文 |
| 会话认证 | JWT（HS256），HttpOnly + SameSite=Strict Cookie |
| CSRF 防护 | 登录表单 CSRF Token，Cookie 绑定校验 |
| 登录限流 | IP 维度速率限制，窗口内最大尝试次数 |
| 图形验证码 | Pillow 生成，Redis 存储 5 分钟 TTL |
| WebSocket 认证 | 连接时校验 JWT，失效返回 4000 关闭码触发前端跳转登录 |

---

## 多服务器部署

AttackFlow 支持多台服务器同时采集，汇聚到同一 Redis + ClickHouse。Agent 已独立为自包含项目，不依赖主应用。

1. **每台被监控服务器**部署一个 Agent，配置相同的 `REDIS_URL` 和 `CLICKHOUSE_*`
2. Agent 启动时自动定位本机坐标（通过公网 IP），向 Redis Hash `servers:registry` 注册
3. 每 30s 心跳续期，90s 无心跳自动过期
4. 攻击事件携带 `target_coords`（该 Agent 的服务器坐标）
5. 前端通过 `/api/servers` 获取所有在线节点，在地图上渲染多个服务器散点 + 服务器列表卡片

### 方式一：直接运行

```bash
cd agent
uv sync
cp .env.example .env  # 编辑 Redis/ClickHouse 连接信息
uv run python collector.py
```

### 方式二：Docker 部署

```bash
cd agent
docker build -t attackflow-agent .
docker run -d --env-file .env \
  -v /var/log/auth.log:/var/log/auth.log:ro \
  attackflow-agent
```

### Agent `.env` 示例

```dotenv
REDIS_URL=rediss://default:xxx@upstash-host:6379/0
CLICKHOUSE_HOST=ch-host
CLICKHOUSE_PORT=9000
LOG_FILE_PATH=/var/log/auth.log
SERVER_LOCATION_AUTO=1          # 自动定位（无需手动配坐标）
# SERVER_LON=-77.0369           # 自动定位关闭时使用
# SERVER_LAT=38.9072
```

---

## 数据清除

提供 `app/clean_data.py` 脚本清除 Redis 与 ClickHouse 中的数据：

```bash
# 交互确认清除两端
uv run python -m app.clean_data

# 跳过确认
uv run python -m app.clean_data --yes

# 仅清 Redis
uv run python -m app.clean_data --redis --yes

# 仅清 ClickHouse
uv run python -m app.clean_data --ch --yes
```

清除范围：

- **Redis 精确 key**：`attacks:stream`、`attacks:recent`、`servers:registry`
- **Redis 前缀扫描**：`geo:ip:`、`auth:session:`、`auth:captcha:`、`rate:login:`、`alert:state:`、`alert:window:`
- **ClickHouse**：`TRUNCATE TABLE attackflow.attacks`

> 脚本内置连接超时（Redis 10s、ClickHouse 15s），防止云服务连接卡死。

---

## 开发指南

### 技术栈

- **后端**：Python 3.10+ / FastAPI / Pydantic v2 / uvicorn
- **存储**：Redis 7（实时层）/ ClickHouse 24（持久化层）
- **采集**：aiohttp（异步 IP 定位）/ 原生文件 tail
- **认证**：bcrypt / PyJWT / CSRF
- **前端**：原生 CSS 变量主题 / ECharts 5 / FontAwesome 6 / Inter + JetBrains Mono
- **包管理**：[uv](https://docs.astral.sh/uv/)（`package = false` 应用模式）

### 常用命令

```bash
uv sync                       # 安装主应用依赖
uv sync --dev                 # 安装开发依赖（pytest, httpx）
uv run pytest                 # 运行测试
cd agent && uv sync           # 安装 Agent 独立依赖
uv run python -m app.clean_data --yes  # 清除所有数据
```

### 坐标格式约定

全链路统一使用 **`[经度, 纬度]`** 顺序（与 ECharts geo 坐标系一致）：

- ipinfo.io 返回 `"lat,lon"`，Agent 转换为 `[lon, lat]`
- `target_coords = [SERVER_LON, SERVER_LAT]`
- ClickHouse 存储 `lon` / `lat` 独立列，查询重组为 `[lon, lat]`
- 前端 `coords[0]` = 经度，`coords[1]` = 纬度

### ClickHouse 表结构

```sql
CREATE TABLE attackflow.attacks
(
    timestamp      DateTime64(3),
    ip             String,
    country_code   LowCardinality(String),
    country_name   String,
    city           String,
    lon            Float64,
    lat            Float64,
    source_host    String,
    attack_count   UInt32
)
ENGINE = MergeTree
PARTITION BY toYYYYMM(timestamp)
ORDER BY (timestamp, ip)
TTL timestamp + INTERVAL 90 DAY;
```

---

## 许可证

私有项目，作者 didiplus。