# 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
> **审核人**: 待定