Metadata-Version: 2.4
Name: fengyun-agent-task
Version: 0.2.2
Summary: CncertAgent local WSS log cache, offline analysis, submission, and API adapter toolkit.
Author: Fengyun-AI-Agent
License-Expression: MIT
Keywords: fengyun,security,logs,sqlite,wss
Classifier: Development Status :: 3 - Alpha
Classifier: Environment :: Console
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Security
Classifier: Topic :: System :: Logging
Requires-Python: >=3.9
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: requests>=2.31.0
Requires-Dist: websocket-client>=1.7.0
Requires-Dist: pycryptodome>=3.23.0
Requires-Dist: hyperscan>=0.7.0
Requires-Dist: pandas>=2.0.0
Dynamic: license-file

# Fengyun AI Agent 本地 WSS 缓存与离线分析工具

该目录是独立实现。

## 目标

- WSS 接收大量日志时优先本地落盘，避免在线分析拖慢接收。
- SQLite WAL 保存结构化缓存，支持多进程读写。
- JSONL 保存原始 WSS 消息和原始日志，便于灾备和复盘。
- 告警与攻击链先进入本地 outbox，再由提交命令统一提交。
- 单包研判会抽取 `entities` 实体；攻击链分析基于实体索引表做关联。
- 攻击链关联会保存结构化 `association`，前端链下日志可展示上级日志、两侧实体和关联关系。

## 什么是 `--dry-run`

`--dry-run` 表示“试运行 / 演练模式”。

适用命令：

- `submit`

开启 `--dry-run` 后会：

- 按 `--status`、`--kind`、`--limit` 等参数筛选本地待提交记录。
- 打印将要提交的 payload，便于检查内容。
- 不向远端服务端发送提交请求。
- 不更新 `submission_records` 中的提交状态。
- 不记录提交尝试。
- 不写入 `submitted_alerts.jsonl` / `submitted_chains.jsonl`。

例如：

```bash
python -m fengyun_ai_agent.cli submit --kind all --status failed --dry-run
```

含义是：从本地提交记录中找出 `failed` 状态的告警和攻击链，打印将要提交的 payload，但不实际提交、不改变本地提交状态。

## 命令总览

| 命令 | 用途 | 是否修改本地数据 | 是否提交到服务端 |
|---|---|---|---|
| `init-db` | 初始化 SQLite WAL 数据库 | 是 | 否 |
| `reset-db` | 删除并重建 SQLite 数据库 | 是，破坏性操作 | 否 |
| `status` | 查看本地缓存统计 | 否 | 否 |
| `team-status` | 查询远端队伍状态/比赛服务端收发进度 | 否 | 是，只读查询 |
| `receive` | 接收 WSS 日志并写入 JSONL + SQLite | 是 | 否 |
| `import-sample-data` | 导入样例日志 | 是 | 否 |
| `import-raw-logs` | 从 `raw_logs.jsonl` 或 `raw_logs_YYYYMMDD.jsonl` 回灌日志到 SQLite | 是 | 否 |
| `analyze-alerts` | 分析日志并生成本地告警候选 | 是 | 否 |
| `analyze-chains` | 基于告警候选生成攻击链候选 | 是 | 否 |
| `submit` | 统一提交本地告警/攻击链 outbox | 会更新提交状态 | 是，除非 `--dry-run` |
| `show-log` | 打印本地日志原文 | 否 | 否 |
| `show-chain` | 打印本地攻击链日志列表 | 否 | 否 |

## 命令参考

所有命令都通过以下入口执行：

```bash
python -m fengyun_ai_agent.cli <command> [args]
```

### `init-db`

初始化本地 SQLite 数据库，并启用 WAL 模式。

```bash
python -m fengyun_ai_agent.cli init-db
```

参数：无。

### `reset-db`

删除并重建本地 SQLite 数据库。

```bash
python -m fengyun_ai_agent.cli reset-db --yes
```

参数：

| 参数 | 默认值 | 说明 |
|---|---:|---|
| `--yes` | `False` | 确认删除旧 SQLite 数据库；不提供会拒绝执行。 |

注意：这是破坏性操作，会删除：

- `fengyun_ai_agent/data/log_cache.sqlite3`
- `fengyun_ai_agent/data/log_cache.sqlite3-wal`
- `fengyun_ai_agent/data/log_cache.sqlite3-shm`

不会删除 JSONL 原始备份文件。

### `status`

查看本地缓存统计，输出 JSON 格式统计信息。

```bash
python -m fengyun_ai_agent.cli status
```

参数：无。

### `team-status`

查询远端队伍状态，接口为 `GET /api/v1/team/status`。连接参数来自环境变量 `GX_SERVER`、`GX_TEAM_ID`、`GX_TOKEN`、`GX_VERIFY_CERT`、`GX_REQUEST_TIMEOUT`。

```bash
python -m fengyun_ai_agent.cli team-status
python -m fengyun_ai_agent.cli team-status --full
python -m fengyun_ai_agent.cli team-status --json
```

参数：

| 参数 | 类型 | 默认值 | 说明 |
|---|---|---:|---|
| `--full` | flag | `False` | 打印所有阶段和播放次数的详细进度。 |
| `--json` | flag | `False` | 输出服务端原始 JSON；存在时忽略 `--full`。 |

说明：该命令是只读远端查询，不写 SQLite、不写 JSONL、不提交 outbox。`status` 查看本地缓存统计，`team-status` 查看比赛服务端队伍进度。

### `receive`

接收 WSS 日志流，写入 JSONL 原始备份和 SQLite 本地缓存。

```bash
python -m fengyun_ai_agent.cli receive
python -m fengyun_ai_agent.cli receive --max-batches 1
```

参数：

| 参数 | 类型 | 默认值 | 说明 |
|---|---|---:|---|
| `--max-batches` | int | `0` | 最多接收多少个 `log_batch`；`0` 表示不限。 |

说明：该命令只负责接收和落盘，不做告警分析，也不提交到服务端。

WSS 原始备份会按日期拆分写入，默认文件名如下：

```text
src/fengyun_ai_agent/data/raw_messages_YYYYMMDD.jsonl
src/fengyun_ai_agent/data/raw_logs_YYYYMMDD.jsonl
```

日期使用执行命令机器的本地日期，不读取日志内时间字段。`TASK_RAW_MESSAGES_JSONL` 和 `TASK_RAW_LOGS_JSONL` 仍作为基础文件名配置，例如基础路径为 `raw_logs.jsonl` 时，实际备份文件为 `raw_logs_20260704.jsonl`。

### `import-sample-data`

导入样例 JSON 日志到 SQLite。

```bash
python -m fengyun_ai_agent.cli import-sample-data
python -m fengyun_ai_agent.cli import-sample-data --data-dir data
python -m fengyun_ai_agent.cli import-sample-data --data-dir data --no-raw-backup
```

参数：

| 参数 | 类型 | 默认值 | 说明 |
|---|---|---:|---|
| `--data-dir` | string | `data` | 样例数据目录。 |
| `--no-raw-backup` | flag | `False` | 只导入 SQLite，不追加写入按日期拆分的 raw messages/raw logs JSONL。 |

默认会同时追加写入 `raw_messages_YYYYMMDD.jsonl` 和 `raw_logs_YYYYMMDD.jsonl` 原始备份，日期使用执行导入命令时机器的本地日期。

### `import-raw-logs`

从本地 `raw_logs.jsonl` 或按日期拆分的 `raw_logs_YYYYMMDD.jsonl` 批量写入 SQLite，适合重置数据库后用原始日志备份恢复本地缓存。

```bash
python -m fengyun_ai_agent.cli import-raw-logs
python -m fengyun_ai_agent.cli import-raw-logs --path src/fengyun_ai_agent/data/raw_logs.jsonl
python -m fengyun_ai_agent.cli import-raw-logs --path src/fengyun_ai_agent/data/raw_logs_20260704.jsonl
python -m fengyun_ai_agent.cli import-raw-logs --batch-size 500 --strict
```

参数：

| 参数 | 类型 | 默认值 | 说明 |
|---|---|---:|---|
| `--path` | string | 配置中的 `RAW_LOGS_JSONL` | JSONL 文件路径；按日期备份时传 `raw_logs_YYYYMMDD.jsonl`。 |
| `--batch-size` | int | `1000` | 每批写入 SQLite 的日志数量。 |
| `--strict` | flag | `False` | 遇到坏行立即失败；默认跳过坏行并打印统计。 |

该命令会初始化/迁移 SQLite schema，但不会删除旧数据库；如需干净重建，先执行 `reset-db --yes`。

### `analyze-alerts`

分析未处理日志，生成本地告警候选。

`--watch` 模式会启动两个互不阻塞的队列：`local` 只取非 `mail` 日志，`llm` 只取 `mail` 日志。本地规则队列完成一批后会立即取下一批，不等待邮件大模型队列。

```bash
python -m fengyun_ai_agent.cli analyze-alerts
python -m fengyun_ai_agent.cli analyze-alerts --limit 100
python -m fengyun_ai_agent.cli analyze-alerts --limit 100 --local-workers 30 --llm-workers 4 --watch --sleep-seconds 5
```

参数：

| 参数 | 类型 | 默认值 | 说明 |
|---|---|---:|---|
| `--limit` | int | `1000` | 每批最多取多少条未处理日志。 |
| `--local-workers` | int | 环境变量 | 本地规则队列并发数；默认读取 `TASK_ALERT_LOCAL_WORKERS`，未设置或为 `0` 时沿用 `--limit`；`hids` 也在这个队列。 |
| `--llm-workers` | int | 环境变量 | 大模型队列并发数；默认读取 `TASK_ALERT_LLM_WORKERS`，未设置时读取 `LLM_CONCURRENCY`，再默认 `4`；当前只有 `mail` 进入这个队列。 |
| `--watch` | flag | `False` | 持续运行；有未分析日志时连续处理，空批时等待。 |
| `--sleep-seconds` | float | `5.0` | watch 模式空批等待秒数。 |
| `--max-rounds` | int | `0` | watch 模式最多执行多少轮；`0` 表示不限。 |
| `--max-idle-rounds` | int | `0` | watch 模式连续空批多少轮后退出；`0` 表示不限。 |

说明：该命令只生成本地告警候选，不直接提交到服务端。

### `analyze-chains`

基于本地告警候选生成攻击链候选。

```bash
python -m fengyun_ai_agent.cli analyze-chains
python -m fengyun_ai_agent.cli analyze-chains --limit 1000 --watch --sleep-seconds 5
python -m fengyun_ai_agent.cli analyze-chains --no-chain-rescan
python -m fengyun_ai_agent.cli analyze-chains --rescan-limit 50
```

说明：该命令只生成本地攻击链候选，不直接提交到服务端。

参数：

| 参数 | 类型 | 默认值 | 说明 |
|---|---|---:|---|
| `--limit` | int | `1000` | 每轮最多处理多少条 seed 黑告警；`0` 表示不限。 |
| `--no-chain-rescan` | flag | `False` | 本次攻击链分析不执行到期回扫。 |
| `--rescan-limit` | int | `0` | 每轮最多回扫多少条到期攻击链；`0` 表示不限。 |
| `--watch` | flag | `False` | 持续运行；有待链分析黑告警时连续处理，空批时等待。 |
| `--sleep-seconds` | float | `5.0` | watch 模式空批等待秒数。 |
| `--max-rounds` | int | `0` | watch 模式最多执行多少轮；`0` 表示不限。 |
| `--max-idle-rounds` | int | `0` | watch 模式连续空批多少轮后退出；`0` 表示不限。 |

攻击链关联规则维护在 `src/fengyun_ai_agent/chain_rules.py`。规则格式支持 7 或 8 项：

```python
# 默认双向关联
('waf', 'srcIp', 'str', 'web', 'webfReqIps', 'list', 'in')

# 只允许 left -> right
('waf', 'srcIp', 'str', 'web', 'webfReqIps', 'list', 'in', 'forward')

# 只允许 right -> left
('waf', 'srcIp', 'str', 'web', 'webfReqIps', 'list', 'in', 'backward')
```

`direction` 可选值：

| 值 | 含义 |
|---|---|
| `bidirectional` | 双向，默认值；兼容旧规则。 |
| `forward` | 只允许左侧日志关联右侧日志。 |
| `backward` | 只允许右侧日志关联左侧日志。 |

链内由关联拉黑的日志会写入 `association_json`，API 输出为 `association`；前端会展示上级日志 ID、上级实体、本告警关联实体、关联关系。单包研判本身已判黑、同时又被关联命中的日志，会保留原研判说明并追加“关联逻辑”。

当新告警导致已有攻击链发生合并时，系统会保留合并前的旧链记录并标记为 `invalid`；合并后的 canonical 攻击链会更新 payload 并重置为 `pending`，等待重新提交到服务端。

已成功提交的攻击链会进入定期回扫：默认按链生成时间后的 10/20/30 分钟最多回扫 3 次。回扫由 `analyze-chains` 在攻击链分析阶段触发，从 `seed_log_id` 起始告警重新递归关联。如果没有新增告警，只更新 `rescan_count/last_rescan_at/next_rescan_at`；如果发现新增告警或触发链合并，会重建链 payload 并把攻击链 outbox 重置为 `pending`，等待后续 `submit` 提交。

### `submit`

统一提交本地 outbox。默认提交告警和攻击链中所有 `pending`、`failed` 状态记录。

```bash
python -m fengyun_ai_agent.cli submit
python -m fengyun_ai_agent.cli submit --dry-run
python -m fengyun_ai_agent.cli submit --kind alert --status pending
python -m fengyun_ai_agent.cli submit --kind chain --status failed --limit 50
python -m fengyun_ai_agent.cli submit --kind all --status pending,failed
python -m fengyun_ai_agent.cli submit --kind alert,chain --status all --dry-run
python -m fengyun_ai_agent.cli submit --watch --sleep-seconds 4
python -m fengyun_ai_agent.cli submit --watch --sleep-seconds 4 --max-idle-rounds 10
```

参数：

| 参数 | 类型 | 默认值 | 可选值 | 说明 |
|---|---|---:|---|---|
| `--kind` | string | `all` | `alert` / `chain` / `all` / 逗号分隔组合 | 控制提交告警、攻击链或两者。 |
| `--status` | string | `pending,failed` | `pending` / `failed` / `submitted` / `all` / 逗号分隔组合 | 控制提交哪些状态的 outbox。 |
| `--limit` | int | `0` | 任意整数 | 每个 kind/status 组合最多提交多少条；`0` 或负数表示不限。 |
| `--dry-run` | flag | `False` | - | 只打印 payload，不实际提交、不更新提交状态、不写 submitted JSONL。 |
| `--watch` | flag | `False` | - | 持续运行；每轮扫描并提交新出现的 outbox。 |
| `--sleep-seconds` | float | `5.0` | 任意非负数 | watch 模式每轮提交后的等待秒数。 |
| `--max-rounds` | int | `0` | 任意非负整数 | watch 模式最多执行多少轮；`0` 表示不限。 |
| `--max-idle-rounds` | int | `0` | 任意非负整数 | watch 模式连续空轮多少次后退出；`0` 表示不限。 |

说明：`submit` 不会重新接收日志，也不会重新分析日志。真实提交时会先把本地已标黑但缺失 outbox 的日志、以及缺失 outbox 的攻击链同步到 `submission_records`；`--dry-run` 只打印当前已有 outbox，不做同步。

常用失败重试流程：

```bash
python -m fengyun_ai_agent.cli submit --kind all --status failed --dry-run
python -m fengyun_ai_agent.cli submit --kind all --status failed
```

第一条先预览将要提交的内容；第二条才真正提交。

### `show-log`

打印本地日志原文 JSON。

```bash
python -m fengyun_ai_agent.cli show-log WAF0000001
```

参数：

| 参数 | 类型 | 说明 |
|---|---|---|
| `log_id` | positional string | 要查看的本地日志 ID。 |

### `show-chain`

打印本地攻击链关联日志列表。

```bash
python -m fengyun_ai_agent.cli show-chain CHAIN_LOCAL_xxx
```

参数：

| 参数 | 类型 | 说明 |
|---|---|---|
| `chain_id` | positional string | 要查看的本地攻击链 ID。 |

输出内容包括时间、日志源、日志 ID。

## 推荐使用流程

### 1. 初始化

```bash
python -m fengyun_ai_agent.cli init-db
python -m fengyun_ai_agent.cli status
```

### 2. 导入样例数据

```bash
python -m fengyun_ai_agent.cli import-sample-data --data-dir data
```

如果只想导入 SQLite，不追加 JSONL 备份：

```bash
python -m fengyun_ai_agent.cli import-sample-data --data-dir data --no-raw-backup
```

SQLite schema 和字段含义见：

```text
src/fengyun_ai_agent/SQLITE_SCHEMA.md
```

### 3. 接收日志前查看远端队伍状态

```bash
python -m fengyun_ai_agent.cli team-status
python -m fengyun_ai_agent.cli team-status --full
```

### 4. 接收日志

```bash
python -m fengyun_ai_agent.cli receive
```

联调只接收一批：

```bash
python -m fengyun_ai_agent.cli receive --max-batches 1
```

### 5. 分析和提交

```bash
python -m fengyun_ai_agent.cli analyze-alerts --limit 100

python -m fengyun_ai_agent.cli analyze-chains --limit 10000
python -m fengyun_ai_agent.cli submit --dry-run
python -m fengyun_ai_agent.cli submit
```

### 6. 复盘查询

```bash
python -m fengyun_ai_agent.cli show-log WAF0000001
python -m fengyun_ai_agent.cli show-chain CHAIN_LOCAL_xxx
```

## 常用环境变量

```bash
export GX_SERVER=https://172.17.35.21:18080
export GX_TEAM_ID=TEAM001
export GX_TOKEN=team001-demo-token
export GX_VERIFY_CERT=0
export TASK_DATA_DIR=./data

# 打印真实提交的最终 payload 和服务端响应，默认关闭
export TASK_SUBMIT_VERBOSE_LOG=1

# 攻击链分析阶段回扫默认开启；默认按 600 秒间隔最多回扫 3 次
export TASK_CHAIN_RESCAN_ENABLED=1
export TASK_CHAIN_RESCAN_INTERVAL_SECONDS=600
export TASK_CHAIN_RESCAN_MAX_COUNT=3

# LLM 返回缓存默认开启；关闭后每次都真实请求模型，不读写缓存
export TASK_LLM_CACHE_ENABLED=1

# LLM 返回缓存默认写入 $TASK_DATA_DIR/llm_cache.jsonl，可读 JSONL，不使用数据库
# export TASK_LLM_CACHE_JSONL=./data/llm_cache.jsonl

# 默认使用外网 LLM；生产/比赛本地环境只需设置为 1 后切到固定的内网 vLLM
export LLM_PRODUCTION=1

# 最高优先级覆盖；设置后无论 LLM_PRODUCTION 取值都会使用这里的配置
# export LLM_BASE_URL=http://127.0.0.1:8100/v1
# export LLM_MODEL=qwen3-8B
# export LLM_API_KEY=EMPTY
```

LLM 配置统一在 `src/fengyun_ai_agent/judge/llm_config.py` 中维护，`mail_judge` 会读取这套配置；HIDS 默认关闭大模型兜底，只有设置 `TASK_HIDS_LLM_ENABLED=1` 时才会使用 `LLMClient`。`LLM_PRODUCTION=1` 时默认使用 `http://127.0.0.1:8100/v1`、`qwen3-8B`、`EMPTY`。

LLM 缓存统一在 `src/fengyun_ai_agent/judge/llm_cache.py` 中维护，缓存键为 `场景名 + MD5(最终 system/user 输入文本)`。默认开启并写入 `$TASK_DATA_DIR/llm_cache.jsonl`，每行保留场景、MD5、完整输入文本、模型原始返回和模型元信息，方便人工排查；命中后直接复用返回内容，不访问外网/内网模型，也不写入业务数据库。设置 `TASK_LLM_CACHE_ENABLED=0` 后会完全跳过缓存读写。当前场景名包括 `mail_judge`、`mail_judge_retry`、`hids_judge`、`waf_judge`、`web_judge`。

## 代码结构

- `config.py`：环境变量、默认路径、接口 endpoint 和日志源映射。
- `db.py`：SQLite 连接、初始化、轻量迁移和重置。
- `json_utils.py`：JSON/JSONL 序列化、容错解析和原始备份写入。
- `repository.py`：本地 SQLite 仓储主入口，负责日志入库/查询、攻击链元数据、payload 构造和 CLI/离线流程兼容导出。
- `submission_repository.py`：`submission_records` outbox、提交记录和重试状态。
- `analyze.py`：本地黑日志分析和攻击链聚合入口。
- `chain_rules.py`：实体关联规则定义，支持双向/单向关联方向。
- `api_repository.py`：Flask 后端和前端 API 兼容适配层。
- `cli.py`：命令行入口，包含 team-status、submit、show-log、show-chain 等 CLI 辅助命令。
- `judge/llm_config.py`：统一管理外网 LLM 和本地 vLLM 的 OpenAI 兼容参数。
- `judge/llm_cache.py`：基于可读 JSONL 的 LLM 返回缓存，按场景名和最终输入文本 MD5 复用结果。

## 大数据量性能建议

当前实现已针对百万级日志做了几项基础优化：

- `logs.id` 主键 upsert，重复导入不会重复插入。
- `log_entities` 实体索引表保存 `(log_source, entity_type, entity_value)`，攻击链不再全表扫描 `logs.entities_json`。
- `analyze-chains` 只从当前 seed 的相关规则出发，并通过 `idx_log_entities_lookup` 查候选。
- SQLite 使用 WAL、`busy_timeout` 和按命令独立连接，适合本地单机批处理。

150 万日志量下建议：

- 导入使用较大批次，例如 `import-raw-logs --batch-size 5000` 或 `10000`，减少事务次数。
- 分析分批跑，例如 `analyze-alerts --limit 100 --watch --max-idle-rounds 1`、`analyze-chains --limit 0 --watch --max-idle-rounds 1`，避免单次长事务和过长输出。
- 保持数据库在 SSD 本地盘，不要放在网络盘或同步目录。
- 重建库前停止 Flask 后端，避免 SQLite 文件被长连接占用。
- 原始备份会按日期拆成 `raw_logs_YYYYMMDD.jsonl`，按比赛日分别回灌和分析，避免三天数据混在一个大文件里。

后续仍可优化的方向：

- 已新增 `idx_logs_alert_pending(alert_analyzed, received_at)`，减少 150 万日志下的未分析扫描成本。
- 链分析 seed 已有 `(label, chain_analyzed, timestamp, received_at)` 组合索引，可根据真实查询计划继续补充其它组合索引。
- `import-raw-logs` 当前逐行 JSON 解析和每批 upsert，150 万量级可接受；如果导入时间成为瓶颈，可以在导入期间临时调大 batch、关闭控制台逐批输出，或增加 bulk import 专用命令。
- 如果攻击链关联候选过多，需要对高基数字段保留、低信息量字段限流，例如过于常见的 `reqUserAgent` 可按规则降权或单向化。

## 重要说明

- `raw_messages*.jsonl` 和 `raw_logs*.jsonl` 是 append-only 原始备份，不会自动删除；WSS 接收和样例导入默认写入按日期拆分的 `raw_messages_YYYYMMDD.jsonl` / `raw_logs_YYYYMMDD.jsonl`。
- WSS 接收命令只做落盘，不做网络提交。
- 每个命令单独打开 SQLite 连接，不要跨进程共享连接。
- 攻击链规则中低信息量实体可能放大关联范围，应优先通过 `direction`、规则删减或字段黑名单控制。
