尧图网页设计 
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/wazuhWazuh 的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依赖版本用途pytest7.4.3测试执行框架docker6.1.3管理 OpenSearch 容器requests2.31.0OpenSearch REST API 调用建索引、清索引、健康检查jsonschema4.20.0响应结构 Schema 校验pycryptodome3.19.0AES / Blowfish 加解密代理与管理器通信flatbuffers23.5.26FlatBuffers 序列化/反序列化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-compilermacOS 可用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 15153.2 命令行选项run_tests.py 中除 README 列出的 5 个选项外还提供代理复用与数据目录配置等扩展选项选项说明默认值--managerWazuh 管理器 IP 地址127.0.0.1--port管理器通信端口1514--registration-port代理注册端口1515--test运行指定测试不带.json后缀全部测试--agent-id复用已有代理 ID无新注册--agent-name代理名称复用代理时可省略无--agent-key代理密钥复用代理时可省略无--test-data-dir测试数据目录test_data--expected-data-dir期望结果目录expected_data--verbose/-v输出详细结果含每条消息耗时与错误False--list-tests列出可用测试并退出False--list-tests会遍历test_data/*.json读取每个文件的description字段并检查expected_data/中是否存在对应文件缺失会标记⚠️。3.3 执行流程剖析run_tests.py的main()完整流程为对应 README 的 Usage 语义设置 OpenSearch调用InventorySyncIntegrationTester.setup_opensearch()先探测localhost:9200是否已有实例如 CI 中的 service container否则拉起名为opensearch-test的 Docker 容器opensearchproject/opensearch:latest单节点、禁用安全插件健康检查check_opensearch_health()请求/_cluster/health要求状态为green或yellow随后清空所有非系统索引并创建带映射的inventory_sync索引准备代理setup_agent()新建WazuhAgent实例——未指定--agent-id时调用register_agent()走 TLS 注册端口 1515、派生 AES 密钥、发送 startup 控制消息指定时则从wazuh_agents.json凭证文件恢复执行测试execute_test_sequence()按 JSON 定义的消息序列逐条发送并采集响应索引核验等待 2 秒后调用check_opensearch_indices()通过/_cat/indices检查 inventory/wazuh 相关索引是否落盘汇总退出全部通过返回 0否则返回 1--verbose下输出每条消息的状态、耗时与错误。四、协议基础FlatBuffers 消息与同步模式4.1 协议 Schema所有测试消息都封装为 FlatBuffersMessageunion其定义来自仓库协议文件 inventorySync.fbs核心枚举如下数值稳定不可重排枚举值说明ModeModuleFull0, ModuleDelta1, ModuleCheck2, MetadataDelta3, MetadataCheck4, GroupDelta5, GroupCheck6同步模式OperationUpsert0, Delete1文档操作类型StatusOk0, Error1, Offline2, ChecksumMismatch3, Processing4应答状态OptionSync0, VDFirst1, VDSync2会话选项MessageTypeunion 判别值覆盖Start1, StartAck2, End3, EndAck4, DataValue5, DataBatch6, DataClean7, ChecksumModule8, DataContext9, ReqRet10。其中Start表携带module/mode/size/option/indices及完整的代理元数据architecture/hostname/osname/osplatform/ostype/osversion/agentversion/agentname/agentid/groups/global_version/cluster_name/cluster_nodeDataValue携带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()序列化为Messageunionstart/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 5MetadataDelta3, GroupDelta5qa/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: 0ModuleFull、size: 1随后恰好发送 1 条 DataValuedata消息通过use_session_from_start: true自动携带 start 应答返回的会话 IDend关闭会话。expected_data/basic_flow.json 则断言start_ackstatus: 0即Ok且必须包含非空 session、data 消息不应有响应、end_ackstatus: 0并开启validate_session_consistency校验三条消息会话一致性。5.2 无数据流程nodata_flowREADME 将其描述为测试同步期间无数据发送的处理。实际用例test_data/nodata_flow.json仅发送一条startmode: 0、size: 0即结束对应的 expected_data/nodata_flow.json 期望收到start_ack且status: 1Error同时不校验 session——即声明size0的无效空同步被管理器拒绝。5.3 请求-返回机制reqret_end_flow/simple_reqret_testReqRet 是同步协议中针对序列号缺口的补传机制对应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_ackstatus: 0。其期望文件明确断言第一次end的应答类型为reqret。simple_reqret_test是同一机制的最小化版本size: 3仅跳过 seq1。5.4 元数据 Delta 同步metadata_delta_flowREADME 明确此模式用于代理元数据变化主机名、OS、架构等时批量更新所有既有文档。真实用例 test_data/metadata_delta_flow.json 使用mode: 3MetadataDelta、size: 0不发送任何数据消息start携带完整元数据与目标索引列表wazuh-states-fim-files、wazuh-states-sca、wazuh-states-inventory-system及global_version: 12345随后直接end期望EndAck状态为Okverification.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_flowREADME 说明此模式用于代理组归属变化时更新所有既有文档。真实用例 test_data/groups_delta_flow.json 使用mode: 5GroupDelta、size: 0start携带groups: [webservers, production, eu-west]与global_version: 54321同样直接end。期望更新字段为wazuh.agent.groups与state.document_version。实现侧对应 inventorySyncFacade.hpp 的buildGroupsUpdateQuery()executeUpdateByQuery()。5.6 其余内置用例速览test_data/中其余用例覆盖了更细粒度的协议边界可在掌握上述核心流程后逐一研读用例覆盖点module_check_match_flowChecksumModule 校验和匹配 →EndAck{Status_Ok}无需全量重同步module_check_mismatch_flow校验和不匹配 →EndAck{Status_ChecksumMismatch}触发全量重同步data_clean_single_index_flowDataClean 单索引 deleteByQuery 流程start→dataclean→enddata_clean_multiple_indices_flow多索引 DataClean序列号跟踪与多索引清理data_context_single_flow/data_context_multiple_flowDataContext 上下文数据入 RocksDBcontext_前缀不发送至索引器缺失序列走 ReqRetforbidden_index_in_*_flow非wazuh-states-*索引在 checksum/dataclean/datavalue 场景中被静默丢弃out_of_range_seq_rejection_flowseq size的 DataValue 被拒绝GapSet::observe 抛std::out_of_range被 WorkersQueue 吸收缺口仍触发 ReqRetdata_value_quota_exhausted_flow超出声明配额时的数据值处理这些用例与 inventorySyncFacade.hpp 中的handleData/handleDataClean/handleDataContext/handleChecksumModule分支一一对应可互为印证。六、测试数据格式与创建新测试6.1 JSON 结构测试数据test_data/与期望结果expected_data/均使用 JSONtest_data 文件顶层含description与messages[]每条消息含typestart/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_responsenull表示无响应否则校验type、data.type、data.status、validate_session与timeout。6.2 校验逻辑test_inventory_sync_integration.py 的validate_results()按以下规则比对对应 README 的Expected result validation消息条数与期望值一致若开启validate_session_consistency整个序列必须使用同一会话逐条检查状态与响应期望无响应却收到响应如 data 消息视为失败期望有响应却缺失也视为失败响应结构逐字段比对_validate_response包括超时timeout_exceeded判定。同一文件还提供了 pytest 集成opensearchfixture 负责容器生命周期test_inventory_sync_*系列函数直接断言result[validation][passed]可通过 pytest.ini 的-m not slow等标记筛选。6.3 创建步骤在test_data/中创建test_name.json定义消息序列在expected_data/中创建同名test_name.json定义期望响应执行python run_tests.py --manager ip --test test_name验证。七、故障排查README 给出的三类常见问题与排查思路Connection Refused连接被拒绝确认 Wazuh 管理器正在运行且--port默认 1514可达Agent Registration Failed代理注册失败检查管理端注册端口1515是否开放、认证配置是否允许新代理注册复用已有代理可加--agent-id跳过注册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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 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/wazuhWazuh 的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依赖版本用途pytest7.4.3测试执行框架docker6.1.3管理 OpenSearch 容器requests2.31.0OpenSearch REST API 调用建索引、清索引、健康检查jsonschema4.20.0响应结构 Schema 校验pycryptodome3.19.0AES / Blowfish 加解密代理与管理器通信flatbuffers23.5.26FlatBuffers 序列化/反序列化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-compilermacOS 可用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 15153.2 命令行选项run_tests.py 中除 README 列出的 5 个选项外还提供代理复用与数据目录配置等扩展选项选项说明默认值--managerWazuh 管理器 IP 地址127.0.0.1--port管理器通信端口1514--registration-port代理注册端口1515--test运行指定测试不带.json后缀全部测试--agent-id复用已有代理 ID无新注册--agent-name代理名称复用代理时可省略无--agent-key代理密钥复用代理时可省略无--test-data-dir测试数据目录test_data--expected-data-dir期望结果目录expected_data--verbose/-v输出详细结果含每条消息耗时与错误False--list-tests列出可用测试并退出False--list-tests会遍历test_data/*.json读取每个文件的description字段并检查expected_data/中是否存在对应文件缺失会标记⚠️。3.3 执行流程剖析run_tests.py的main()完整流程为对应 README 的 Usage 语义设置 OpenSearch调用InventorySyncIntegrationTester.setup_opensearch()先探测localhost:9200是否已有实例如 CI 中的 service container否则拉起名为opensearch-test的 Docker 容器opensearchproject/opensearch:latest单节点、禁用安全插件健康检查check_opensearch_health()请求/_cluster/health要求状态为green或yellow随后清空所有非系统索引并创建带映射的inventory_sync索引准备代理setup_agent()新建WazuhAgent实例——未指定--agent-id时调用register_agent()走 TLS 注册端口 1515、派生 AES 密钥、发送 startup 控制消息指定时则从wazuh_agents.json凭证文件恢复执行测试execute_test_sequence()按 JSON 定义的消息序列逐条发送并采集响应索引核验等待 2 秒后调用check_opensearch_indices()通过/_cat/indices检查 inventory/wazuh 相关索引是否落盘汇总退出全部通过返回 0否则返回 1--verbose下输出每条消息的状态、耗时与错误。四、协议基础FlatBuffers 消息与同步模式4.1 协议 Schema所有测试消息都封装为 FlatBuffersMessageunion其定义来自仓库协议文件 inventorySync.fbs核心枚举如下数值稳定不可重排枚举值说明ModeModuleFull0, ModuleDelta1, ModuleCheck2, MetadataDelta3, MetadataCheck4, GroupDelta5, GroupCheck6同步模式OperationUpsert0, Delete1文档操作类型StatusOk0, Error1, Offline2, ChecksumMismatch3, Processing4应答状态OptionSync0, VDFirst1, VDSync2会话选项MessageTypeunion 判别值覆盖Start1, StartAck2, End3, EndAck4, DataValue5, DataBatch6, DataClean7, ChecksumModule8, DataContext9, ReqRet10。其中Start表携带module/mode/size/option/indices及完整的代理元数据architecture/hostname/osname/osplatform/ostype/osversion/agentversion/agentname/agentid/groups/global_version/cluster_name/cluster_nodeDataValue携带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()序列化为Messageunionstart/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 5MetadataDelta3, GroupDelta5qa/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: 0ModuleFull、size: 1随后恰好发送 1 条 DataValuedata消息通过use_session_from_start: true自动携带 start 应答返回的会话 IDend关闭会话。expected_data/basic_flow.json 则断言start_ackstatus: 0即Ok且必须包含非空 session、data 消息不应有响应、end_ackstatus: 0并开启validate_session_consistency校验三条消息会话一致性。5.2 无数据流程nodata_flowREADME 将其描述为测试同步期间无数据发送的处理。实际用例test_data/nodata_flow.json仅发送一条startmode: 0、size: 0即结束对应的 expected_data/nodata_flow.json 期望收到start_ack且status: 1Error同时不校验 session——即声明size0的无效空同步被管理器拒绝。5.3 请求-返回机制reqret_end_flow/simple_reqret_testReqRet 是同步协议中针对序列号缺口的补传机制对应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_ackstatus: 0。其期望文件明确断言第一次end的应答类型为reqret。simple_reqret_test是同一机制的最小化版本size: 3仅跳过 seq1。5.4 元数据 Delta 同步metadata_delta_flowREADME 明确此模式用于代理元数据变化主机名、OS、架构等时批量更新所有既有文档。真实用例 test_data/metadata_delta_flow.json 使用mode: 3MetadataDelta、size: 0不发送任何数据消息start携带完整元数据与目标索引列表wazuh-states-fim-files、wazuh-states-sca、wazuh-states-inventory-system及global_version: 12345随后直接end期望EndAck状态为Okverification.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_flowREADME 说明此模式用于代理组归属变化时更新所有既有文档。真实用例 test_data/groups_delta_flow.json 使用mode: 5GroupDelta、size: 0start携带groups: [webservers, production, eu-west]与global_version: 54321同样直接end。期望更新字段为wazuh.agent.groups与state.document_version。实现侧对应 inventorySyncFacade.hpp 的buildGroupsUpdateQuery()executeUpdateByQuery()。5.6 其余内置用例速览test_data/中其余用例覆盖了更细粒度的协议边界可在掌握上述核心流程后逐一研读用例覆盖点module_check_match_flowChecksumModule 校验和匹配 →EndAck{Status_Ok}无需全量重同步module_check_mismatch_flow校验和不匹配 →EndAck{Status_ChecksumMismatch}触发全量重同步data_clean_single_index_flowDataClean 单索引 deleteByQuery 流程start→dataclean→enddata_clean_multiple_indices_flow多索引 DataClean序列号跟踪与多索引清理data_context_single_flow/data_context_multiple_flowDataContext 上下文数据入 RocksDBcontext_前缀不发送至索引器缺失序列走 ReqRetforbidden_index_in_*_flow非wazuh-states-*索引在 checksum/dataclean/datavalue 场景中被静默丢弃out_of_range_seq_rejection_flowseq size的 DataValue 被拒绝GapSet::observe 抛std::out_of_range被 WorkersQueue 吸收缺口仍触发 ReqRetdata_value_quota_exhausted_flow超出声明配额时的数据值处理这些用例与 inventorySyncFacade.hpp 中的handleData/handleDataClean/handleDataContext/handleChecksumModule分支一一对应可互为印证。六、测试数据格式与创建新测试6.1 JSON 结构测试数据test_data/与期望结果expected_data/均使用 JSONtest_data 文件顶层含description与messages[]每条消息含typestart/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_responsenull表示无响应否则校验type、data.type、data.status、validate_session与timeout。6.2 校验逻辑test_inventory_sync_integration.py 的validate_results()按以下规则比对对应 README 的Expected result validation消息条数与期望值一致若开启validate_session_consistency整个序列必须使用同一会话逐条检查状态与响应期望无响应却收到响应如 data 消息视为失败期望有响应却缺失也视为失败响应结构逐字段比对_validate_response包括超时timeout_exceeded判定。同一文件还提供了 pytest 集成opensearchfixture 负责容器生命周期test_inventory_sync_*系列函数直接断言result[validation][passed]可通过 pytest.ini 的-m not slow等标记筛选。6.3 创建步骤在test_data/中创建test_name.json定义消息序列在expected_data/中创建同名test_name.json定义期望响应执行python run_tests.py --manager ip --test test_name验证。七、故障排查README 给出的三类常见问题与排查思路Connection Refused连接被拒绝确认 Wazuh 管理器正在运行且--port默认 1514可达Agent Registration Failed代理注册失败检查管理端注册端口1515是否开放、认证配置是否允许新代理注册复用已有代理可加--agent-id跳过注册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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考