fengyun-agent-task 0.1.2

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

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

该目录是独立实现。

目标

• WSS 接收大量日志时优先本地落盘，避免在线分析拖慢接收。
• SQLite WAL 保存结构化缓存，支持多进程读写。
• JSONL 保存原始 WSS 消息和原始日志，便于灾备和复盘。
• 告警与攻击链先进入本地 outbox，再由提交命令统一提交。

什么是 --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 回灌日志到 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 表示不限。

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

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.jsonl / raw_logs.jsonl。

默认会同时追加写入 raw_messages.jsonl 和 raw_logs.jsonl 原始备份。

import-raw-logs

从本地 raw_logs.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 --batch-size 500 --strict

参数：

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

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

analyze-alerts

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

python -m fengyun_ai_agent.cli analyze-alerts
python -m fengyun_ai_agent.cli analyze-alerts --limit 100 --watch --sleep-seconds 5

参数：

参数	类型	默认值	说明
--limit	int	1000	本轮最多分析多少条未处理日志。
--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

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

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 10000

python -m fengyun_ai_agent.cli analyze-chains
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

代码结构

• config.py：环境变量、默认路径、接口 endpoint 和日志源映射。
• db.py：SQLite 连接、初始化、轻量迁移和重置。
• json_utils.py：JSON/JSONL 序列化、容错解析和原始备份写入。
• repository.py：本地 SQLite 仓储主入口，负责日志入库/查询、攻击链元数据、payload 构造和 CLI/离线流程兼容导出。
• submission_repository.py：submission_records outbox、提交记录和重试状态。
• analyze.py：本地黑日志分析和攻击链聚合入口。
• api_repository.py：Flask 后端和前端 API 兼容适配层。
• cli.py：命令行入口，包含 team-status、submit、show-log、show-chain 等 CLI 辅助命令。

重要说明

• raw_messages.jsonl 和 raw_logs.jsonl 是 append-only 原始备份，不会自动删除。
• WSS 接收命令只做落盘，不做网络提交。
• 每个命令单独打开 SQLite 连接，不要跨进程共享连接。
• 第一版检测和攻击链逻辑是基线启发式，后续可以在 detectors.py 和 analyze.py 中增强。