fengyun-agent-task 0.2.2

CncertAgent local WSS log cache, offline analysis, submission, and API adapter toolkit.

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。

例如：

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	打印本地攻击链日志列表	否	否

命令参考

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

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

init-db

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

python -m fengyun_ai_agent.cli init-db

参数：无。

reset-db

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

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 格式统计信息。

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。

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 本地缓存。

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 原始备份会按日期拆分写入，默认文件名如下：

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。

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，适合重置数据库后用原始日志备份恢复本地缓存。

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 日志。本地规则队列完成一批后会立即取下一批，不等待邮件大模型队列。

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

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

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 项：

# 默认双向关联
('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 状态记录。

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，不做同步。

常用失败重试流程：

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。

python -m fengyun_ai_agent.cli show-log WAF0000001

参数：

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

show-chain

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

python -m fengyun_ai_agent.cli show-chain CHAIN_LOCAL_xxx

参数：

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

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

推荐使用流程

1. 初始化

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

2. 导入样例数据

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

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

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

SQLite schema 和字段含义见：

src/fengyun_ai_agent/SQLITE_SCHEMA.md

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

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

4. 接收日志

python -m fengyun_ai_agent.cli receive

联调只接收一批：

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

5. 分析和提交

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. 复盘查询

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

常用环境变量

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、规则删减或字段黑名单控制。