fengyun-agent-task 0.1.1

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

task 本地 WSS 缓存与离线分析工具

该目录是独立实现。

目标

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

什么是 --dry-run

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

适用命令：

• submit-alerts
• submit-chains
• replay

开启 --dry-run 后会：

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

例如：

python -m task.cli replay --kind all --status failed --dry-run

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

命令总览

命令	用途	是否修改本地数据	是否提交到服务端
init-db	初始化 SQLite WAL 数据库	是	否
reset-db	删除并重建 SQLite 数据库	是，破坏性操作	否
status	查看本地缓存统计	否	否
receive	接收 WSS 日志并写入 JSONL + SQLite	是	否
import-sample-data	导入样例日志	是	否
analyze-alerts	分析日志并生成本地告警候选	是	否
analyze-chains	基于告警候选生成攻击链候选	是	否
submit-alerts	提交本地告警 outbox	会更新提交状态	是，除非 --dry-run
submit-chains	提交本地攻击链 outbox	会更新提交状态	是，除非 --dry-run
replay	重放 pending/failed/submitted 提交记录	会更新提交状态	是，除非 --dry-run
show-log	打印本地日志原文	否	否
show-chain	打印本地攻击链日志列表	否	否

命令参考

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

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

init-db

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

python -m task.cli init-db

参数：无。

reset-db

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

python -m task.cli reset-db --yes

参数：

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

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

• task/data/log_cache.sqlite3
• task/data/log_cache.sqlite3-wal
• task/data/log_cache.sqlite3-shm

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

status

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

python -m task.cli status

参数：无。

receive

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

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

参数：

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

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

import-sample-data

导入样例 JSON 日志到 SQLite。

python -m task.cli import-sample-data
python -m task.cli import-sample-data --data-dir data
python -m task.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 原始备份。

analyze-alerts

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

python -m task.cli analyze-alerts
python -m task.cli analyze-alerts --limit 10000

参数：

参数	类型	默认值	说明
--limit	int	1000	本轮最多分析多少条未处理日志。

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

analyze-chains

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

python -m task.cli analyze-chains
python -m task.cli analyze-chains --max-logs-per-chain 2000

参数：

参数	类型	默认值	说明
--max-logs-per-chain	int	2000	每条候选攻击链最多纳入多少条日志。

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

submit-alerts

提交本地告警 outbox。

python -m task.cli submit-alerts
python -m task.cli submit-alerts --dry-run --limit 10
python -m task.cli submit-alerts --status failed --limit 100

参数：

参数	类型	默认值	可选值	说明
--limit	int	100	任意整数	本次最多处理多少条告警提交记录。
--status	string	pending	pending / failed / submitted	筛选要处理的提交状态。
--dry-run	flag	False	-	只打印 payload，不实际提交、不更新提交状态。

submit-chains

提交本地攻击链 outbox。

python -m task.cli submit-chains
python -m task.cli submit-chains --dry-run
python -m task.cli submit-chains --status failed --limit 50

参数：

参数	类型	默认值	可选值	说明
--limit	int	100	任意整数	本次最多处理多少条攻击链提交记录。
--status	string	pending	pending / failed / submitted	筛选要处理的提交状态。
--dry-run	flag	False	-	只打印 payload，不实际提交、不更新提交状态。

replay

重放已保存在本地 outbox 中的提交 payload。

python -m task.cli replay
python -m task.cli replay --kind all --status failed --dry-run
python -m task.cli replay --kind alert --status failed --limit 20
python -m task.cli replay --kind chain --status pending

参数：

参数	类型	默认值	可选值	说明
--kind	string	all	all / alert / chain	重放类型：全部、仅告警、仅攻击链。
--status	string	failed	pending / failed / submitted	选择要重放的提交状态。
--limit	int	100	任意整数	每类最多处理多少条；--kind all 时 alert 和 chain 分别最多处理该数量。
--dry-run	flag	False	-	只打印将要重放的 payload，不实际提交、不更新状态。

说明：replay 是“重试/重放已保存的提交 payload”，不是重新接收日志，也不是重新分析日志。

常用失败重放流程：

python -m task.cli replay --kind all --status failed --dry-run
python -m task.cli replay --kind all --status failed

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

show-log

打印本地日志原文 JSON。

python -m task.cli show-log WAF0000001

参数：

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

show-chain

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

python -m task.cli show-chain CHAIN_LOCAL_xxx

参数：

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

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

推荐使用流程

1. 初始化

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

2. 导入样例数据

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

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

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

SQLite schema 和字段含义见：

task/SQLITE_SCHEMA.md

3. 接收日志

python -m task.cli receive

联调只接收一批：

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

4. 分析和提交

python -m task.cli analyze-alerts --limit 10000
python -m task.cli submit-alerts --dry-run --limit 10
python -m task.cli submit-alerts --limit 200

python -m task.cli analyze-chains
python -m task.cli submit-chains --dry-run
python -m task.cli submit-chains

5. 复盘查询

python -m task.cli show-log WAF0000001
python -m task.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=task/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：命令行入口，包含 replay、show-log、show-chain 等 CLI 辅助命令。

重要说明

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