Wazuh inventory_sync 集成测试框架实战：用 FlatBuffers 协议与 JSON 用例验证端到端资产同步

Wazuh inventory_sync 集成测试框架实战：用 FlatBuffers 协议与 JSON 用例验证端到端资产同步

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

Wazuh 的inventory_sync模块负责将代理端采集的资产清单（系统、软件包、文件完整性、SCA、漏洞等状态）通过可靠的同步协议写入 OpenSearch。本文以 inventory_sync QA 集成测试文档 为主体，结合 inventorySync.fbs 协议定义 与 C++ 同步实现，系统讲解这套测试框架的架构、环境准备、命令用法、内置用例与 JSON 用例编写规范，帮助你在本地复现"真实代理 → 管理器 → OpenSearch"的完整同步链路验证。

一、框架定位：为什么需要协议级集成测试

inventory_sync的同步过程横跨三层：代理端（采集与序列化）、管理器端（agentSession.hpp 中基于 GapSet 的会话管理）以及索引端（OpenSearch/Indexer Connector）。仅靠单元测试（如 tests/unit/agentSession_test.cpp）无法覆盖真实网络链路、加密通信与异步索引行为。

qa/目录下的集成测试框架正是为填补这一空白而设计，其核心能力（对应 README 的 Overview）：

• 自动化测试：使用真实 Wazuh 代理协议驱动inventory_sync模块，而非 mock 桩；
• JSON 化测试数据：test_data/与expected_data/以 JSON 定义场景与预期，用例创建和长期维护成本低；
• 顺序化消息执行：按"start → data → end"的顺序逐条发送，忠实还原同步算法对消息次序的强依赖；
• 预期结果校验：把管理器的真实响应与expected_data/中的期望结构逐字段比对，并可在测试后核验 OpenSearch 索引落盘结果。

二、环境准备

2.1 系统要求

• Python 3.8 或更高版本；
• 一台可访问、正在运行的 Wazuh 管理器；
• Docker（用于拉起 OpenSearch 测试实例）。

2.2 Python 依赖

requirements.txt 明确锁定了以下版本：

pip install -r requirements.txt

2.3 FlatBuffers 生成

测试框架通过 generate_flatbuffers.py 从仓库内的协议文件src/shared_modules/utils/flatbuffers/schemas/inventorySync.fbs生成 Python 类。该脚本在测试运行需要时会自动执行，也可手动运行：

python3 generate_flatbuffers.py

前提是系统已安装flatc编译器（Ubuntu/Debian 可用sudo apt-get install flatbuffers-compiler，macOS 可用brew install flatbuffers）。脚本会依次探测/usr/local/bin/flatc、/usr/bin/flatc与PATH，成功后以--python --gen-object-api输出到qa/generated/目录，并补齐Wazuh/SyncSchema包结构。加载与解析逻辑见 flatbuffers_manager.py 中的FlatBuffersManager。

三、运行集成测试

3.1 基本用法与常用命令

对本地管理器运行全部测试：

python run_tests.py --manager 127.0.0.1

运行单个测试：

python run_tests.py --manager 127.0.0.1 --test basic_flow

复用已注册代理（跳过注册等待）与自定义端口：

python run_tests.py --manager 127.0.0.1 --agent-id 001 --agent-name "test-agent" python run_tests.py --manager 127.0.0.1 --port 1514 --registration-port 1515

3.2 命令行选项

run_tests.py 中除 README 列出的 5 个选项外，还提供代理复用与数据目录配置等扩展选项：

--list-tests会遍历test_data/*.json，读取每个文件的description字段，并检查expected_data/中是否存在对应文件（缺失会标记⚠️）。

3.3 执行流程剖析

run_tests.py的main()完整流程为（对应 README 的 Usage 语义）：

1. 设置 OpenSearch：调用InventorySyncIntegrationTester.setup_opensearch()，先探测localhost:9200是否已有实例（如 CI 中的 service container），否则拉起名为opensearch-test的 Docker 容器（opensearchproject/opensearch:latest，单节点、禁用安全插件）；
2. 健康检查：check_opensearch_health()请求/_cluster/health，要求状态为green或yellow；随后清空所有非系统索引并创建带映射的inventory_sync索引；
3. 准备代理：setup_agent()新建WazuhAgent实例——未指定--agent-id时调用register_agent()走 TLS 注册（端口 1515）、派生 AES 密钥、发送 startup 控制消息；指定时则从wazuh_agents.json凭证文件恢复；
4. 执行测试：execute_test_sequence()按 JSON 定义的消息序列逐条发送并采集响应；
5. 索引核验：等待 2 秒后调用check_opensearch_indices()通过/_cat/indices检查 inventory/wazuh 相关索引是否落盘；
6. 汇总退出：全部通过返回 0，否则返回 1；--verbose下输出每条消息的状态、耗时与错误。

四、协议基础：FlatBuffers 消息与同步模式

4.1 协议 Schema

所有测试消息都封装为 FlatBuffersMessageunion，其定义来自仓库协议文件 inventorySync.fbs，核心枚举如下（数值稳定，不可重排）：

MessageTypeunion 判别值覆盖：Start=1, StartAck=2, End=3, EndAck=4, DataValue=5, DataBatch=6, DataClean=7, ChecksumModule=8, DataContext=9, ReqRet=10。其中Start表携带module/mode/size/option/indices及完整的代理元数据（architecture/hostname/osname/osplatform/ostype/osversion/agentversion/agentname/agentid/groups/global_version/cluster_name/cluster_node），DataValue携带session/seq/operation/id/index/data载荷。更详细的逐消息参考见 benchmark/tool_simulator/docu/06-flatbuffers-messages.md。

4.2 消息构造与应答处理

测试端在 wazuh_agent_controller.py 中完成协议仿真：

• 载荷格式为s:inventory_sync:<json>，由 flatbuffers_manager.py 的create_message()序列化为Messageunion（start/data/end 等类型走各自的 FlatBuffer Builder 分支）；
• 封装阶段执行"MD5 摘要 + 随机数 + 全局/本地计数器 → zlib 压缩 → Wazuh 自定义!填充 → AES/Blowfish CBC 加密 →!<agentid>!#AES:头"的标准代理打包；
• 响应解析支持startup_response、control_ack、flatbuffer、text、binary等类型，并实现了关键的EndAck(Processing) 跳过逻辑（_PROCESSING_STATUS = 4）：当管理器返回EndAck{Processing}表示会话已入队但尚未完成索引时，_receive_final_response()会透明地继续读取，直到收到最终的EndAck{Ok/Error}，避免把中间态误判为结果。

4.3 同步模式与测试用例的对应

需要特别说明：README 中将 MetadataDelta 标注为"Mode 4"、GroupDelta 标注为"Mode 6"，而仓库实际 schema 中二者分别为Mode 3 与 Mode 5（MetadataDelta=3, GroupDelta=5），qa/test_data/中的 JSON 用例也使用mode: 3与mode: 5。下文以 schema 权威数值为准。

五、内置测试用例详解

test_data/与expected_data/目录中各有 17 个同名 JSON 文件。以下结合 README 说明与各用例的实际内容展开。

5.1 基础流程测试（basic_flow）

覆盖最基础的同步会话：start → data → end。对应的 test_data/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" }, "description": "Start synchronization session", "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"} }, "description": "Send data message using session from start response", "delay": 0.5, "use_session_from_start": true }, { "type": "end", "data": {}, "description": "End synchronization session using session from start response", "delay": 0.5, "use_session_from_start": true } ] }

要点：start消息声明mode: 0（ModuleFull）、size: 1（随后恰好发送 1 条 DataValue）；data消息通过use_session_from_start: true自动携带 start 应答返回的会话 ID；end关闭会话。

expected_data/basic_flow.json 则断言：start_ack（status: 0，即Ok）且必须包含非空 session、data 消息不应有响应、end_ack（status: 0），并开启validate_session_consistency校验三条消息会话一致性。

5.2 无数据流程（nodata_flow）

README 将其描述为"测试同步期间无数据发送的处理"。实际用例（test_data/nodata_flow.json）仅发送一条start（mode: 0、size: 0）即结束；对应的 expected_data/nodata_flow.json 期望收到start_ack且status: 1（Error），同时不校验 session——即声明size=0的无效空同步被管理器拒绝。

5.3 请求-返回机制（reqret_end_flow/simple_reqret_test）

ReqRet 是同步协议中针对序列号缺口的补传机制，对应MessageType.ReqRet与Pair(begin, end)区间表。底层由 agentSession.hpp 中的GapSet追踪已收/缺失分片；会话结束时若仍有缺口，管理器返回ReqRet请求代理补传。

test_data/reqret_end_flow.json 演示了完整闭环：start声明size: 5（期望 seq 0–4），随后发送 seq 0、2、4（跳过 1、3 制造双缺口），end触发ReqRet，补传 seq 1、3 后再次end才收到end_ack（status: 0）。其期望文件明确断言第一次end的应答类型为reqret。simple_reqret_test是同一机制的最小化版本（size: 3，仅跳过 seq=1）。

5.4 元数据 Delta 同步（metadata_delta_flow）

README 明确：此模式用于代理元数据变化（主机名、OS、架构等）时批量更新所有既有文档。真实用例 test_data/metadata_delta_flow.json 使用mode: 3（MetadataDelta）、size: 0，不发送任何数据消息：

• start携带完整元数据与目标索引列表（wazuh-states-fim-files、wazuh-states-sca、wazuh-states-inventory-system）及global_version: 12345；
• 随后直接end，期望EndAck状态为Ok；
• verification.expected_updates定义索引核验字段：wazuh.agent.name、wazuh.agent.version、wazuh.agent.host.architecture/hostname/os.name/os.platform/os.type/os.version与state.document_version。

其底层实现在 inventorySyncFacade.hpp：开始元数据更新前会锁定该代理（lockAgent，拒绝并发会话）、flush 挂起的 bulk 操作、等待该代理其他会话完成（最长 60 秒，超时会自动清理"僵尸会话"），随后由InventorySyncQueryBuilder::buildMetadataUpdateQuery()构造 update-by-query，经executeUpdateByQuery()对所有指定索引批量执行，回调中解锁代理并发送EndAck{Ok}。

5.5 组 Delta 同步（groups_delta_flow）

README 说明：此模式用于代理组归属变化时更新所有既有文档。真实用例 test_data/groups_delta_flow.json 使用mode: 5（GroupDelta）、size: 0，start携带groups: ["webservers", "production", "eu-west"]与global_version: 54321，同样直接end。期望更新字段为wazuh.agent.groups与state.document_version。实现侧对应 inventorySyncFacade.hpp 的buildGroupsUpdateQuery()+executeUpdateByQuery()。

5.6 其余内置用例速览

test_data/中其余用例覆盖了更细粒度的协议边界，可在掌握上述核心流程后逐一研读：

这些用例与 inventorySyncFacade.hpp 中的handleData/handleDataClean/handleDataContext/handleChecksumModule分支一一对应，可互为印证。

六、测试数据格式与创建新测试

6.1 JSON 结构

测试数据（test_data/）与期望结果（expected_data/）均使用 JSON：

• test_data 文件：顶层含description与messages[]；每条消息含type（start/data/end/dataclean/datacontext/checksum_module）、data（消息体）、description、delay（发送间隔，秒）、use_session_from_start（复用 start 应答会话，支持命名会话如"reqret_test"）、expect_session_response、expect_end_response等控制字段；部分用例带verification块定义 OpenSearch 索引核验。
• expected_data 文件：含expected_message_count、validate_session_consistency与expected_messages[]；每条期望消息声明expected_status、expected_response（null表示无响应；否则校验type、data.type、data.status、validate_session与timeout）。

6.2 校验逻辑

test_inventory_sync_integration.py 的validate_results()按以下规则比对（对应 README 的"Expected result validation"）：

1. 消息条数与期望值一致；
2. 若开启validate_session_consistency，整个序列必须使用同一会话；
3. 逐条检查状态与响应：期望无响应却收到响应（如 data 消息）视为失败；期望有响应却缺失也视为失败；
4. 响应结构逐字段比对（_validate_response），包括超时（timeout_exceeded）判定。

同一文件还提供了 pytest 集成：opensearchfixture 负责容器生命周期，test_inventory_sync_*系列函数直接断言result["validation"]["passed"]，可通过 pytest.ini 的-m "not slow"等标记筛选。

6.3 创建步骤

1. 在test_data/中创建<test_name>.json，定义消息序列；
2. 在expected_data/中创建同名<test_name>.json，定义期望响应；
3. 执行python run_tests.py --manager <ip> --test <test_name>验证。

七、故障排查

README 给出的三类常见问题与排查思路：

1. Connection Refused（连接被拒绝）：确认 Wazuh 管理器正在运行，且--port（默认 1514）可达；
2. Agent Registration Failed（代理注册失败）：检查管理端注册端口1515是否开放、认证配置是否允许新代理注册；复用已有代理可加--agent-id跳过注册；
3. Import Errors（导入错误）：执行pip install -r requirements.txt补齐依赖；FlatBuffers 相关错误则确认flatc已安装（必要时手动运行python3 generate_flatbuffers.py）。

可结合源码进一步定位：若出现大量EndAck(Processing)后无最终应答，说明会话已入队但索引未完成，可检查 wazuh_agent_controller.py 的_receive_final_response超时（默认 10 秒）与期望文件中的timeout字段。

八、许可

本测试框架隶属于 Wazuh 项目（开源安全平台），沿用项目相同的许可条款（GPL v2，详见 agentSession.hpp 文件头与仓库根目录 LICENSE）。测试执行期间请遵守对目标管理器与 OpenSearch 实例的合规使用要求。

小结：通过这套框架，你可以在数分钟内验证"代理注册 → FlatBuffers 消息序列化 → 加密链路传输 → 管理器 GapSet 会话管理 → OpenSearch 索引落盘"的完整闭环，并通过新增 JSON 用例覆盖 ReqRet 补传、元数据/组 Delta、校验和比对等高级同步模式，为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