Wazuh Inventory Sync 测试工具链实战：单元测试、协议集成测试与端到端会话仿真工具-平芜编程栈

Wazuh Inventory Sync 测试工具链实战：单元测试、协议集成测试与端到端会话仿真工具

【免费下载链接】wazuhWazuh - The Open Source Security Platform. Unified XDR and SIEM protection for endpoints and cloud workloads.项目地址: https://gitcode.com/GitHub_Trending/wa/wazuh

Inventory Sync 是 Wazuh 5.x 中负责将 Agent 侧资产清单（软件包、热修复、操作系统信息等）增量同步到 Manager 并写入 Wazuh Indexer 的核心模块，它基于 FlatBuffer 协议完成 Agent-Manager 之间的会话式数据交换。本文围绕源码树中的三类测试面展开：核心会话逻辑的 gtest 单元测试、基于真实协议的 QA 集成测试套件，以及可模拟完整 Manager 侧会话的端到端工具inventory_sync_testtool。读完后你将掌握如何为 Inventory Sync 的会话逻辑、协议语义（ReqRet 重传、DataClean、校验和匹配）以及 Inventory Sync 到漏洞扫描（Vulnerability Scanner）的完整管道选择合适的测试手段并实际运行它们。

三个测试面的总体分工

Inventory Sync 在源码树src/wazuh_modules/inventory_sync/下提供了三层互补的测试能力：

测试面	位置	测试对象	典型触发场景
单元测试	tests/unit/	`AgentSession`、`GapSet`、`ResponseDispatcher`、`DataBatch`处理	修改会话逻辑、排队行为、协议解析时
QA 协议集成测试	qa/	真实 FlatBuffer 协议下的 Agent-Manager 交互	修改协议语义、确认应答、重传、元数据/分组对账、校验和逻辑时
端到端测试工具	testtool/	Manager 侧完整会话：Inventory Sync + Indexer + 漏洞扫描联动	验证真实的 Manager 侧索引行为或触发漏洞扫描的会话时

三层测试共享同一个 FlatBuffer 协议契约——即生产环境使用的活动模式文件 inventorySync.fbs。这意味着一旦协议变更，必须同步更新qa/与testtool/两个消费方，否则测试与生产行为会脱节。

单元测试：会话、缺块与派发逻辑的白盒验证

单元测试位于 src/wazuh_modules/inventory_sync/tests/unit/，基于 GoogleTest/GoogleMock 构建，覆盖组件包括：

AgentSession：单次同步会话的完整生命周期；
GapSet：稀疏序号（seq）缺块的追踪结构；
ResponseDispatcher：StartAck/EndAck 等回包派发；
DataBatch的处理行为（分块跟踪、校验和、重传）。

这些测试在无完整 Manager 部署的前提下验证了会话生命周期、块跟踪、校验和处理、重传行为与响应派发逻辑。

从源码结构看，其隔离性来自模板化的依赖注入：agentSession_test.cpp 中测试夹具直接使用

using AgentSessionForTest = AgentSessionImpl<MockStore, MockIndexerQueue, MockResponseDispatcher>;

即把AgentSessionImpl的存储（Store）、索引器队列（IndexerQueue）、响应派发器（ResponseDispatcher）三个依赖全部替换为StrictMock，从而能够逐条断言如sendStartAck恰好被调用一次的协议行为（见 agentSession_test.cpp 的Constructor_Success用例：构造一个合法的StartFlatBuffer 消息后，预期EXPECT_CALL(mockResponseDispatcher, sendStartAck(_, _, _, _)).Times(1)）。测试同时验证非法输入路径，例如Constructor_NullData与Constructor_InvalidSize会抛出AgentSessionException。

GapSet 是为稀疏序列号追踪设计的高效结构（类注释描述为 "Efficient GapSet for sparse sequence number tracking"），它支撑了协议中的缺块检测与ReqRet重传语义，对应的 gapSet_test.cpp 提供 16 个TEST_F用例；agentSession_test.cpp 有 40 个，responseDispatcher_test.cpp 有 3 个，此外 datavalue_null_validation_test.cpp 专门验证 DataValue 的空值校验行为。各测试文件的用例数从源码统计可确认：40/7/16/3 个TEST_F。

协议集成测试：用真实 FlatBuffer 消息走通完整会话

QA 集成测试位于 src/wazuh_modules/inventory_sync/qa/，其核心特征是使用真实 FlatBuffer 协议模拟 Agent-Manager 交互，而非 mock 字节流。框架由四个文件组成：

run_tests.py：命令行入口，负责 OpenSearch 环境准备、Agent 注册与测试执行编排；
test_inventory_sync_integration.py：核心测试器InventorySyncIntegrationTester，执行 JSON 描述的消息序列并校验预期结果；
wazuh_agent_controller.py：WazuhAgent模拟 Agent，完成注册（1515 端口）、会话加解密与消息收发；
flatbuffers_manager.py：加载或按需生成 FlatBuffer 消息类（Start、DataValue、End、ReqRet、DataClean、DataContext、ChecksumModule等），并封装create_message/parse_message。

运行方式与参数

前置要求：Python 3.8+、可访问的 Wazuh Manager、Docker（用于拉起 OpenSearch）。依赖安装与手工生成 FlatBuffer 类：

pip install -r requirements.txt python3 generate_flatbuffers.py # 需要系统安装 flatc

常用运行方式：

# 对本地 Manager 运行全部测试 python run_tests.py --manager 127.0.0.1 # 只跑单个用例 python run_tests.py --manager 127.0.0.1 --test basic_flow # 使用已注册的 Agent、自定义端口 python run_tests.py --manager 127.0.0.1 --agent-id 001 --agent-name "test-agent" \ --port 1514 --registration-port 1515

run_tests.py 支持的完整参数：

选项	说明	默认值
`--manager`	Wazuh Manager 地址	`127.0.0.1`
`--port`	Manager 通信端口	`1514`
`--registration-port`	Agent 注册端口	`1515`
`--test`	只运行指定测试（不含 .json 后缀）	全部测试
`--agent-id`/`--agent-name`/`--agent-key`	复用已注册的 Agent 凭据	每次新注册
`--test-data-dir`/`--expected-data-dir`	测试数据/期望结果目录	`test_data`/`expected_data`
`--verbose`/`-v`	输出每个用例的耗时与错误明细	关闭
`--list-tests`	列出可用测试后退出	关闭

测试用例覆盖矩阵

test_data/与expected_data/采用"输入 JSON + 期望结果 JSON"的成对设计，当前 qa/test_data/ 下共有 17 个流程文件，覆盖：

基础 start/data/end 同步（basic_flow）与无数据会话（nodata_flow）；
ReqRet重传处理（reqret_end_flow、simple_reqret_test）；
DataClean处理（单索引/多索引：data_clean_single_index_flow、data_clean_multiple_indices_flow）；
仅DataContext的会话（data_context_single_flow、data_context_multiple_flow）；
校验和匹配/不匹配（module_check_match_flow、module_check_mismatch_flow）；
元数据 delta 更新（metadata_delta_flow）与分组 delta 更新（groups_delta_flow）；
边界与防御性场景：data_value_quota_exhausted_flow、out_of_range_seq_rejection_flow，以及forbidden_index_in_*三个非法索引拒绝用例。

以 basic_flow.json 为例，一个测试文件就是一个有序消息脚本：

{ "description": "Basic inventory sync flow test: start -> data -> end", "messages": [ { "type": "start", "data": { "module": "inventory_sync", "mode": 0, "size": 1, "agentid": "001", "agentname": "test-agent", "agentversion": "4.8.0", "cluster_name": "wazuh" }, "delay": 0.5, "expect_session_response": true }, { "type": "data", "data": { "seq": 0, "operation": 0, "id": "doc123", "index": "wazuh-states-inventory-system", "data": { "message": "Hello", "timestamp": "2025-08-20T10:00:00Z" } }, "delay": 0.5, "use_session_from_start": true }, { "type": "end", "data": {}, "delay": 0.5, "use_session_from_start": true } ] }

其中use_session_from_start: true体现了协议的关键细节：Start消息会收到StartAck，后续data/end消息必须携带StartAck返回的会话句柄，消息间的delay则利用同步算法的有序性逐条验证。

QA 套件还专门覆盖了两个纯 Manager 侧更新的同步模式（见 qa/README.md）：

Metadata Delta（Mode 4）：Agent 元数据（主机名、OS、架构等）变化时使用。会话不发送任何 data 消息，Manager 在指定索引中批量更新wazuh.agent.id/name/version、wazuh.agent.host.*、state.document_version、state.modified_at，最后以Status_Ok结束；
Groups Delta（Mode 6）：Agent 分组归属变化时使用，仅更新wazuh.agent.groups及文档版本字段。

值得注意的是WazuhAgent并非简单的 TCP 客户端：从 wazuh_agent_controller.py 可以看到它实现了与 Manager 一致的 AES-CBC/Blowfish-CBC 加解密（含固定 IV 与 PKCS 填充），因此 QA 套件验证的是包含真实链路加密在内的完整协议栈；凭据会持久化到本地wazuh_agents.json，支持跨轮次复用注册。

inventory_sync_testtool：Manager 侧端到端会话仿真

端到端测试工具位于 src/wazuh_modules/inventory_sync/testtool/，可执行文件名为inventory_sync_testtool（构建配置见 CMakeLists.txt），其 README.md 给出了完整的架构说明。它用一个 JSON 输入文件描述整个会话，用真实 FlatBuffer 消息仿真 Agent 库存数据摄入，并无需完整 Wazuh 部署即可触发漏洞扫描工作流，特别适用于验证 Inventory Sync、Wazuh Indexer 与 Vulnerability Scanner 三者之间的集成。

架构与消息流

┌──────────────────┐ │ Test Tool │ ← 读取 JSON 输入（Start + data_values + data_context） └────────┬─────────┘ ▼ ┌──────────────────┐ │ InventorySync │ ← 处理库存消息，写入 RocksDB │ Facade │ └────────┬─────────┘ │ 触发 VulnerabilityScanner ▼ ┌──────────────────┐ │ Vulnerability │ ← 扫描软件包 CVE，结果发送 Indexer │ Scanner │ └────────┬─────────┘ ▼ ┌──────────────────┐ │ Response Server │ ← 接收 StartAck/EndAck，校验工作流完成 │ (Test Tool) │ └──────────────────┘

初始化阶段，RouterModule 在queue/inventory-states创建 UNIX 域套接字，InventorySync 打开 RocksDB，VulnerabilityScanner 加载 CVE 数据库，工具自带的 ResponseServer 绑定queue/sockets/ar接收确认消息。整体流程为：

Start 消息：由 JSON 的Start对象直接构建 FlatBuffer Start，经 RouterModule 进入 InventorySync，创建扫描会话并把基础上下文存入 RocksDB，随后收到StartAck（含会话 ID）；
Data 消息：DataValue（包/热修复，携带operation与 JSON 原始data载荷）和DataContext（OS 等上下文，恒为 upsert）依次写入 RocksDB。VD 后续读取这些数据构建ScanContext：Agent 数据来自 Start，OS 数据来自 Start + OS 的 DataContext，包与热修复来自对应索引的 DataValue；
End 消息：会话终结并触发 VD 扫描——从 RocksDB 装载包、合并 OS 信息、解析 CNA/Feed 源、版本匹配、平台与厂商校验、Windows 热修复缓解校验，最终生成 ECS 事件写入 Indexer，工具收到EndAck后打印统计并退出。

Manager 侧回包走 ResponseDispatcher 协议，格式为：

"(msg_to_agent) [] N!s <agentId> <size> <module>_sync <flatbuffer>"

配置与输入格式

运行inventory_sync_testtool需要两个文件：

config.json（Indexer 连接与 mTLS 证书，示例见 test_data/config.json）：

{ "indexer": { "hosts": ["https://ChangeMe:9200"], "ssl": { "certificate_authorities": ["/var/wazuh-manager/etc/certs/root-ca.pem"], "certificate": "/var/wazuh-manager/etc/certs/manager.pem", "key": "/var/wazuh-manager/etc/certs/manager-key.pem" } } }

indexer.hosts、indexer.ssl.certificate_authorities、indexer.ssl.certificate、indexer.ssl.key四项均为必填。

INPUT_XXX.json（单文件完整描述一次会话），仓库内提供了 INPUT_000.json、INPUT_001.json、INPUT_002.json 三个样例。高层结构：

{ "Start": { "... Start 消息字段 ..." }, "data_values": [ { "operation": "upsert | delete", "payload": { "... 索引文档 ..." } } ], "data_context": [ { "payload": { "... OS 或额外上下文文档 ..." } } ] }

Start对象与 FlatBuffer Start 一一对应，字段要求：

字段	类型	必填	说明
`agentid`	string	是	Agent 标识（`Start.agentid`）
`mode`	string	是	`"full"`或`"delta"`，映射为`Mode_ModuleFull/Delta`
`option`	string	是	`"VDFirst"`、`VDSync`或`"Sync"`
`agentname`/`agentversion`/`architecture`/`hostname`	string	是	Agent 基本属性
`osname`/`osplatform`/`ostype`/`osversion`	string	是	OS 四元组
`groups`	string[]	否	缺省为`["default"]`
`indices`	string[]	是	本次会话涉及的库存索引
`size`	integer	是	消息总数（`data_values + data_context`）

一个典型的 VDSync 场景：

"Start": { "agentid": "001", "mode": "delta", "option": "VDSync", "agentname": "ubuntu22", "agentversion": "v5.0.0", "architecture": "aarch64", "hostname": "ubuntu22", "osname": "Ubuntu", "osplatform": "ubuntu", "ostype": "linux", "osversion": "22.04.5 LTS (Jammy Jellyfish)", "groups": ["default"], "indices": ["wazuh-states-inventory-packages", "wazuh-states-inventory-system"], "size": 3 }

data_values[]的每条记录被转换为一个 DataValue 消息：工具读取operation（upsert/delete）与payload，以payload._index作为 FlatBuffer 索引字段，把payload._source作为原始 JSON 序列化为 FlatBuffer 的data字段。data_context[]则全部按 upsert 处理，用于补充 OS 细节（内核、codename、major/minor 等），与 Start 中已有字段互补。

命令行用法

inventory_sync_testtool [FLAGS] FLAGS: --input <file> 输入 JSON 文件（必填） --config <file> Indexer 配置 JSON（必填） --wait <seconds> End 之后的等待时间（默认 10） --verbose 开启详细日志

示例：

./inventory_sync_testtool \ --input src/wazuh_modules/inventory_sync/testtool/test_data/INPUT_000.json \ --config src/wazuh_modules/inventory_sync/testtool/test_data/config.json \ --wait 15 \ --verbose

正常完成的输出示例：

[INFO] ✓ StartAck received - Session: 14039769528377457750 [INFO] Scanning package [1/1]: 'grafana' - Vendor: 'grafana' - Version: '8.5.5' [INFO] Analyzing CVE: CVE-2022-23498 - Package 'grafana' (v.8.5.5) is VULNERABLE [INFO] Scan for package 'grafana' ended - Found 21 vulnerabilities (analyzed 143 CVE candidates) [INFO] Agent '001' - Scan completed in 245 ms: 1 packages scanned, 1 vulnerable packages, 21 total vulnerabilities found [INFO] ✓ EndAck received - Session: 14039769528377457750 [INFO] Test completed successfully!

排障与使用限制

README 列出了四类常见问题与处理方向：

收不到 StartAck/EndAck：检查 InventorySync 是否正常运行、queue/sockets/ar套接字路径是否正确、ResponseDispatcher 是否发送、JSON 中Start对象是否合法；
Indexer 连接失败/SSL 校验失败：核对 indexer URL、证书文件及 Indexer 集群健康状态；
FlatBuffer 解析错误（如INSUFFICIENT_PADDING）：检查消息格式与 simdjson 载荷构造，必要时清理上一次运行残留的 RocksDB 数据；
零漏洞（假阴性）：确认 CVE 数据库已填充且为最新、用--verbose查看 VD 日志、核对data_values中的版本/厂商/平台字段，以及版本匹配是否因"已装版本高于受影响版本"而正确判为不受影响。

使用上需注意：工具状态保存在 RocksDB，重复使用同一 agent/session 可能需要清理数据库目录（如/tmp/wazuh_inventorysync_test.db）；进程崩溃后 UNIX 套接字（queue/inventory-states、queue/sockets/ar）可能残留需手动删除；并行测试请使用不同 Agent ID 避免 RocksDB 与 Indexer 冲突。已知限制包括：仅仿真 Agent 消息、不连接真实 Agent；面向单 Agent 流程设计；CVE Feed 更新不会自动触发，需走 VulnerabilityScanner 自身的 feed 更新机制。

如何为变更选择测试手段

你改动的内容	首选测试面
会话逻辑、排队行为、协议解析	单元测试（tests/unit/）
协议语义、确认应答、重传、元数据/分组对账、校验和	QA 集成套件（qa/）
真实的 Manager 侧索引、漏洞扫描触发的会话	`inventory_sync_testtool`（testtool/）

最后一条维护约定值得写进团队的检查清单：测试所用的 FlatBuffer 模式就是生产使用的活动模式inventorySync.fbs，更新协议时必须同步更新qa/（含 generate_flatbuffers.py 生成链路）与testtool/两个消费方，保持测试与生产的行为一致性。

【免费下载链接】wazuhWazuh - The Open Source Security Platform. Unified XDR and SIEM protection for endpoints and cloud workloads.项目地址: https://gitcode.com/GitHub_Trending/wa/wazuh

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考