wecom_it_smart_desk/docs/09-部署运维/技术设计-Token多IP异常检测.md

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 (埋点)

# 在 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 异常检测定时任务

# 在 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

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 告警模板

{
  "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 审核人: 待定