Simon
44e77dcb0e
**重构前**（旧编号 02-11）： - docs/02-产品需求/ → 00 产品规划/PRD - docs/03-技术架构/ → 01-05 子目录散落 - docs/04-原型设计/ → 01-02 产品设计（HTML 原型） - docs/05-原型设计/ → screens/ - docs/06-测试素材/ → 02-E2E / 03-功能 / 04-版本测试 - docs/07-项目管理/ → 任务说明书/日报/计划 - docs/08-安全审计/ → 审计报告 - docs/09-堡垒运维/ → toolbox / deploy - docs/10-项目管理/ → 任务说明书（重复） - docs/11-历史归档/ → deploy-nas-archived **重构后**（新编号 00-07，语义化）： - docs/00-产品开发流程与文档管理规范.md - docs/00-版本迭代总览.md - docs/01-产品文档/ （PRD/原型/认证/会话/AI 服务/坐席/集成） - docs/02-技术文档/ （技术方案/架构图/重构记录/前端改造/实现配置） - docs/03-测试文档/ （E2E/功能用例/版本报告/缺陷单） - docs/04-运维文档/ （部署运维/运维指南） - docs/05-运营文档/ （品牌推广/用户手册） - docs/06-安全审计/ （审计报告） - docs/07-项目管理/ （任务说明书/日报/计划/看板） **净收益**： - 目录编号与产品文档管理规范对齐（按文档阶段 01-07 编号） - 消除 02-产品需求 与 10-项目管理 的编号重叠 - 子目录按文档类型分组（如 01-产品文档/00-产品规划、01-产品文档/01-认证与登录） - 把运维/安全/项目管理从 0X 散落改为 04/06/07 合计 494 文件 + 78495 行 / - 14076 行
292 lines
8.7 KiB
Markdown
292 lines
8.7 KiB
Markdown
# Token多IP异常检测 - 技术设计文档
|
|
|
|
> **任务ID**: 待分配
|
|
> **模块**: 威胁检测
|
|
> **优先级**: P1
|
|
> **ATT&CK**: T1078 (有效账户), T1552 (非安全凭据)
|
|
|
|
---
|
|
|
|
## 1. 需求概述
|
|
|
|
### 1.1 业务背景
|
|
|
|
当前系统已废弃密码登录，仅支持企微OAuth2/扫码登录。Token是用户身份的唯一凭证，当Token被泄露后，攻击者可能从不同IP使用同一Token访问系统。本功能旨在检测此类异常行为。
|
|
|
|
### 1.2 功能目标
|
|
|
|
- 记录每个Token使用的IP地址
|
|
- 检测同一Token在短时间内被多个IP使用的情况
|
|
- 触发告警通知安全管理员
|
|
- 可选：自动禁用异常Token
|
|
|
|
---
|
|
|
|
## 2. 技术方案
|
|
|
|
### 2.1 架构设计
|
|
|
|
```
|
|
┌─────────────────────────────────────────────────────────┐
|
|
│ 检测流程 │
|
|
├─────────────────────────────────────────────────────────┤
|
|
│ │
|
|
│ 用户API请求 │
|
|
│ │ │
|
|
│ ▼ │
|
|
│ ┌─────────────────┐ │
|
|
│ │ record_token_ip │ ← 每次请求记录IP │
|
|
│ │ (埋点) │ │
|
|
│ └────────┬────────┘ │
|
|
│ │ │
|
|
│ ▼ │
|
|
│ ┌─────────────────┐ │
|
|
│ │ Redis Set │ ← token_ips:{hash} │
|
|
│ │ IP集合(1h TTL) │ │
|
|
│ └────────┬────────┘ │
|
|
│ │ │
|
|
│ ▼ 定时任务(每分钟) │
|
|
│ ┌─────────────────┐ │
|
|
│ │ detect_anomaly │ ← 扫描异常Token │
|
|
│ │ (定时任务) │ │
|
|
│ └────────┬────────┘ │
|
|
│ │ │
|
|
│ ▼ │
|
|
│ ┌─────────────────┐ │
|
|
│ │ send_alert │ ← 企微机器人告警 │
|
|
│ └─────────────────┘ │
|
|
│ │
|
|
└─────────────────────────────────────────────────────────┘
|
|
```
|
|
|
|
### 2.2 数据结构
|
|
|
|
#### Redis Key设计
|
|
|
|
| Key格式 | 类型 | TTL | 说明 |
|
|
|---------|------|-----|------|
|
|
| `token_ips:{token_hash}` | Set | 3600秒 | 记录Token使用的IP集合 |
|
|
| `token_ips:alerted:{token_hash}` | String | 3600秒 | 已告警标记，避免重复 |
|
|
|
|
#### Token存储（现有）
|
|
|
|
| Key格式 | 类型 | 说明 |
|
|
|---------|------|------|
|
|
| `user:token:{token}` | JSON | 用户信息，含employee_id |
|
|
|
|
### 2.3 接口设计
|
|
|
|
#### 2.3.1 记录Token使用IP (埋点)
|
|
|
|
```python
|
|
# 在 token_service.py 中增加
|
|
async def record_token_ip(token: str, ip: str):
|
|
"""
|
|
记录Token使用的IP地址
|
|
|
|
Args:
|
|
token: 用户Token
|
|
ip: 客户端IP (X-Forwarded-For 或 request.client.host)
|
|
"""
|
|
import hashlib
|
|
token_hash = hashlib.sha256(token.encode()).hexdigest()
|
|
|
|
redis = await get_redis()
|
|
key = f"token_ips:{token_hash}"
|
|
|
|
# 添加IP到Set (自动去重)
|
|
redis.sadd(key, ip)
|
|
# 设置1小时过期
|
|
redis.expire(key, 3600)
|
|
```
|
|
|
|
#### 2.3.2 异常检测定时任务
|
|
|
|
```python
|
|
# 在 tasks/token_anomaly_detection.py
|
|
async def detect_token_anomaly():
|
|
"""
|
|
检测Token异常使用
|
|
|
|
扫描所有 token_ips:* keys
|
|
当 IP数量 >= 阈值 时触发告警
|
|
"""
|
|
# 配置
|
|
THRESHOLD = 3 # IP数量阈值
|
|
WINDOW_SECONDS = 3600 # 时间窗口
|
|
|
|
redis = await get_redis()
|
|
alerted_key_prefix = "token_ips:alerted:"
|
|
|
|
# 扫描所有 token_ips:* keys
|
|
async for key in redis.scan_iter("token_ips:*"):
|
|
# 跳过 alerted keys
|
|
if key.startswith(alerted_key_prefix):
|
|
continue
|
|
|
|
token_hash = key.replace("token_ips:", "")
|
|
ip_count = await redis.scard(key)
|
|
|
|
if ip_count >= THRESHOLD:
|
|
# 检查是否已告警
|
|
alerted_key = f"{alerted_key_prefix}{token_hash}"
|
|
if await redis.get(alerted_key):
|
|
continue # 已告警，跳过
|
|
|
|
# 获取用户信息
|
|
token = await redis.get(f"user:token:{token_hash}")
|
|
if token:
|
|
user_data = json.loads(token)
|
|
employee_id = user_data.get("employee_id")
|
|
|
|
# 发送告警
|
|
await send_security_alert(
|
|
title="Token异常告警",
|
|
content=f"员工 {employee_id} 的Token被 {ip_count} 个IP使用\nToken: {token_hash[:8]}..."
|
|
)
|
|
|
|
# 标记已告警
|
|
await redis.setex(alerted_key, WINDOW_SECONDS, "1")
|
|
```
|
|
|
|
#### 2.3.3 获取客户端IP
|
|
|
|
```python
|
|
def get_client_ip(request) -> str:
|
|
"""获取客户端真实IP"""
|
|
# 优先从 X-Forwarded-For 获取
|
|
forwarded = request.headers.get("X-Forwarded-For")
|
|
if forwarded:
|
|
return forwarded.split(",")[0].strip()
|
|
|
|
# 降级到 request.client.host
|
|
return request.client.host if request.client else ""
|
|
```
|
|
|
|
---
|
|
|
|
## 3. 配置项
|
|
|
|
### 3.1 环境变量
|
|
|
|
| 变量名 | 类型 | 默认值 | 说明 |
|
|
|--------|------|--------|------|
|
|
| `TOKEN_ANOMALY_THRESHOLD` | int | 3 | 触发告警的IP数量阈值 |
|
|
| `TOKEN_ANOMALY_WINDOW` | int | 3600 | 时间窗口(秒) |
|
|
| `TOKEN_ANOMALY_AUTO_DISABLE` | bool | false | 是否自动禁用Token |
|
|
| `CONTENT_AUDIT_WEBHOOK` | string | - | 企微机器人webhook(现有) |
|
|
|
|
---
|
|
|
|
## 4. 告警内容
|
|
|
|
### 4.1 告警模板
|
|
|
|
```json
|
|
{
|
|
"msgtype": "markdown",
|
|
"markdown": {
|
|
"content": "🔴 **Token异常告警**\n\n"
|
|
"> 员工ID: {employee_id}\n"
|
|
"> 异常Token: {token_hash[:8]}...\n"
|
|
"> IP数量: {ip_count}\n"
|
|
"> 时间: {timestamp}\n\n"
|
|
"> **请及时确认是否为本人操作**"
|
|
}
|
|
}
|
|
```
|
|
|
|
---
|
|
|
|
## 5. 集成点
|
|
|
|
### 5.1 现有组件复用
|
|
|
|
| 组件 | 用途 |
|
|
|------|------|
|
|
| Redis | IP存储 |
|
|
| APScheduler | 定时任务 |
|
|
| content_audit_webhook | 企微告警 |
|
|
| token_service.py | Token管理 |
|
|
|
|
### 5.2 侵入点
|
|
|
|
| 文件 | 修改内容 |
|
|
|------|----------|
|
|
| `app/services/token_service.py` | 增加 record_token_ip() |
|
|
| `app/main.py` | 注册定时任务 |
|
|
| `app/tasks/token_anomaly_detection.py` | 新建检测任务 |
|
|
|
|
---
|
|
|
|
## 6. 性能与容错
|
|
|
|
### 6.1 性能估算
|
|
|
|
| 指标 | 估算值 |
|
|
|------|---------|
|
|
| Redis存储 | ~50KB (1000活跃Token) |
|
|
| 定时任务耗时 | < 100ms |
|
|
| 定时任务间隔 | 60秒 |
|
|
|
|
### 6.2 容错设计
|
|
|
|
- 告警发送失败：记录日志，不阻塞主流程
|
|
- Redis连接失败：跳过本次检测，下个周期重试
|
|
- Token不存在：跳过，不影响其他检测
|
|
|
|
---
|
|
|
|
## 7. 测试用例
|
|
|
|
### 7.1 单元测试
|
|
|
|
| 用例ID | 描述 | 预期结果 |
|
|
|--------|------|----------|
|
|
| T001 | 单IP使用Token | 不触发告警 |
|
|
| T002 | 3个IP使用Token | 触发告警 |
|
|
| T003 | 5个IP使用Token | 触发告警(严重) |
|
|
| T004 | 同一IP多次使用 | 不触发告警 |
|
|
|
|
### 7.2 集成测试
|
|
|
|
| 用例ID | 描述 | 预期结果 |
|
|
|--------|------|----------|
|
|
| I001 | 真实Token请求 | IP被记录 |
|
|
| I002 | 定时任务执行 | 异常Token被检测 |
|
|
| I003 | 告警发送 | 企微收到消息 |
|
|
|
|
---
|
|
|
|
## 8. 部署清单
|
|
|
|
### 8.1 文件变更
|
|
|
|
| 操作 | 文件 |
|
|
|------|------|
|
|
| 新增 | `app/tasks/token_anomaly_detection.py` |
|
|
| 修改 | `app/services/token_service.py` |
|
|
| 修改 | `app/main.py` |
|
|
|
|
### 8.2 配置变更
|
|
|
|
| 操作 | 变量 |
|
|
|------|------|
|
|
| 新增(可选) | `TOKEN_ANOMALY_THRESHOLD` |
|
|
| 新增(可选) | `TOKEN_ANOMALY_AUTO_DISABLE` |
|
|
|
|
---
|
|
|
|
## 9. 回滚方案
|
|
|
|
如需回滚：
|
|
1. 移除定时任务注册 (main.py)
|
|
2. 删除 record_token_ip() 调用
|
|
3. Redis keys 会在1小时后自动过期
|
|
|
|
---
|
|
|
|
> **编制人**: 威胁检测工程师
|
|
> **日期**: 2026-07-14
|
|
> **审核人**: 待定
|