企业官网建设流程全解析

Wazuh InventorySync 与漏洞检测集成测试工具：JSON 驱动的端到端 VD 测试实践

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

本篇文章围绕 Wazuh 仓库中 InventorySync VD 测试工具（inventory_sync_testtool）展开，深入讲解其架构、配置、输入数据格式、端到端消息流与调试方法。该工具用于在不部署完整 Wazuh 环境的情况下，以真实 FlatBuffer 协议模拟 agent 库存数据（包、热补丁、OS 信息）的注入，并驱动 InventorySync → VulnerabilityScanner → Indexer 的完整漏洞扫描链路。读完本文，你将掌握如何编写测试输入 JSON、配置 Indexer 连接、运行该工具并通过 StartAck/EndAck 校验扫描结果。

架构总览

核心组件

inventory_sync_testtool是一个独立的 C++ 可执行程序，位于 src/wazuh_modules/inventory_sync/testtool/main.cpp，其构建规则定义在 testtool/CMakeLists.txt 中，链接了inventory_sync、rocksdb、router、indexer_connector、vulnerability_scanner等核心库。它采用单个 JSON 输入文件完整描述一次 InventorySync 会话：

• 初始Start消息（agent 元数据、mode、option、indices 等）；
• 所有DataValue消息（包、热补丁等）；
• 所有DataContext消息（OS 文档与附加上下文）。

整体数据流如下：

┌──────────────────┐ │ Test Tool │ ← Reads JSON input (Start + data_values + data_context) └────────┬─────────┘ │ ├─► RouterModule (queue/inventory-states) │ ▼ ┌──────────────────┐ │ InventorySync │ ← Processes inventory messages │ Facade │ Stores in RocksDB └────────┬─────────┘ │ ├─► Triggers VulnerabilityScanner │ ▼ ┌──────────────────┐ │ Vulnerability │ ← Scans packages for CVEs │ Scanner │ Sends results to Indexer └────────┬─────────┘ │ ├─► ResponseDispatcher (queue/sockets/ar) │ ▼ ┌──────────────────┐ │ Response Server │ ← Receives StartAck/EndAck │ (Test Tool) │ Validates workflow completion └──────────────────┘

从源码看，测试工具内部实现了两个关键的网络角色（见 main.cpp）：

• ResponseServer（main.cpp 第 263 行起）：绑定queue/sockets/arUNIX 数据报套接字，接收 StartAck/EndAck/ReqRet 消息，并通过std::promise同步通知主流程；
• FakeReportServer（main.cpp 第 169 行起）：模拟 manager 的 alert 接收套接字（queue/sockets/queue），打印 VD 产出的告警消息。

关键特性

• 端到端测试：验证 InventorySync → VD → Indexer 完整流水线；
• 真实协议模拟：使用基于 FlatBuffer 的认证协议（Start、DataValue、DataContext、End），消息定义来自inventorySync_generated.h；
• JSON 驱动会话：单个 JSON 文件描述整个同步会话；
• 实时监控：跟踪 StartAck/EndAck 消息实现同步；
• 全面日志：详细的[INFO]/[DEBUG]/[WARN]/[ERROR]输出便于分析与调试；
• 灵活输入：JSON 化测试数据覆盖包、OS 信息与热补丁。

前置条件

1. Indexer 实例

测试工具需要一个正在运行的 Wazuh Indexer 实例来验证端到端功能——最终漏洞事件要写入 OpenSearch/索引。

2. SSL/TLS 证书

工具需要有效的 SSL 证书与 Indexer 进行安全通信：

注意证书必须与 Indexer 配置匹配。相关路径在测试工具使用的config.json中设置。

证书文件清单：

/path/to/certs/ ├── root-ca.pem # Root CA certificate ├── manager.pem # Client certificate └── manager-key.pem # Client private key

仓库内置的示例配置 test_data/config.json 使用默认证书路径/var/wazuh-manager/etc/certs/。

配置

配置文件格式（config.json）

测试工具通过 JSON 配置文件指定 Indexer 连接参数。仓库示例：

{ "vulnerability-detection": { "enabled": "yes" }, "indexer": { "hosts": [ "https://127.0.0.1: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" } }, "clusterName": "cluster01" }

配置参数说明：

除文档列出的indexer配置外，仓库示例还包含vulnerability-detection.enabled: "yes"（开启 VD 模块）与clusterName，说明该配置文件直接复用了 Wazuh 管理端配置结构，由 main.cpp 的parseArgs与启动逻辑 解析并传给 InventorySync 与 VulnerabilityScannerFacade。

输入数据格式

输入文件结构（INPUT_XXX.json）

测试工具接受单个 JSON 文件完整描述一次 InventorySync 会话，顶层结构为：

{ "Start": { "... Start message fields ..." }, "data_values": [ { "operation": "upsert | delete", "payload": { "... Indexer document for a package/hotfix ..." } } ], "data_context": [ { "payload": { "... Indexer document for OS or extra context ..." } } ] }

• Start→ 用于构建 FlatBufferStart消息；
• data_values[]→ 转换为DataValue消息；
• data_context[]→ 转换为DataContext消息（始终按 upsert 处理）。

仓库中提供了三份可直接运行的示例输入：INPUT_000.json、INPUT_001.json、INPUT_002.json。

Start对象

Start块与 InventorySync Start 消息 1:1 映射。完整示例（可对照 INPUT_000.json 的Start字段）：

{ "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 } }

字段说明：

源码级细节：从 main.cpp 的MessageBuilder::buildStart可以看到，实际解析比表格更宽松——agentid、agentname、agentversion、architecture、hostname、osname、ostype、osplatform、osversion、mode、option、groups都有默认值兜底；module默认syscollector（示例输入中为syscollector_vd）；indices缺失时会根据data_values/data_context自动推断；size缺省时按消息数自动计算。parseMode与parseOption（main.cpp 第 513-539 行）实现了字符串到 FlatBuffer 枚举的映射，未知值默认回退到Mode_ModuleDelta/Option_VDSync。

data_values[]数组

每个条目代表一条DataValue消息。工具的处理步骤：

1. 读取operation（"upsert"或"delete"）；
2. 读取payload；
3. 用payload._index（或自动推断的index字段）作为 FlatBuffer 索引；
4. 将payload._source序列化为 FlatBuffer 的data字段（原始 JSON）。

完整示例（包 upsert 与 delete 对，风格同 INPUT_000.json）：

"data_values": [ { "operation": "upsert", "payload": { "_index": "wazuh-states-inventory-packages", "_id": "wazuh_001_f033cfe690b80a478fb4c832934f0fd55927c349", "_score": 2, "_source": { "wazuh": { "agent": { "id": "001", "name": "ubuntu22", "version": "v5.0.0", "groups": ["default"], "host": { "architecture": "aarch64", "hostname": "ubuntu22", "os": { "name": "Ubuntu", "platform": "ubuntu", "type": "linux", "version": "22.04.5 LTS (Jammy Jellyfish)" } } } }, "checksum": { "hash": { "sha1": "c665573c916b60392f615c71742948813a60bbf5" } }, "package": { "category": "misc", "installed": null, "path": null, "priority": "optional", "source": null, "type": "deb", "vendor": "contact@grafana.com", "version": "8.5.5", "architecture": "amd64", "checksum": "eec8dacdb7087b2e0cc5ccebeb1259a6ec7a731e", "description": "Grafana", "name": "grafana", "size": 334, "format": "deb", "item_id": "f97fcd7d34fa7b705241093e8ba47c4458017285", "multiarch": "foreign", "groups": "utils" }, "state": { "document_version": 1, "modified_at": "2025-11-27T12:31:07.733Z" } } } }, { "operation": "delete", "payload": { "_index": "wazuh-states-inventory-packages", "_id": "wazuh_001_f033cfe690b80a478fb4c832934f0fd55927c350", "_score": 2, "_source": { "wazuh": { "agent": { "id": "001", "name": "ubuntu22", "version": "v5.0.0", "groups": ["default"], "host": { "architecture": "aarch64", "hostname": "ubuntu22", "os": { "name": "Ubuntu", "platform": "ubuntu", "type": "linux", "version": "22.04.5 LTS (Jammy Jellyfish)" } } } }, "checksum": { "hash": { "sha1": "82771c9b43f0434021653ebaa811e26e0ebaed5d" } }, "package": { "architecture": "all", "category": "oldlibs", "description": "transitional package for https support", "installed": null, "multiarch": "foreign", "name": "apt-transport-https", "path": null, "priority": "optional", "size": 169984, "source": "apt", "type": "deb", "vendor": "Ubuntu Developers <ubuntu-devel-discuss@lists.ubuntu.com>", "version": "2.4.14" }, "state": { "document_version": 1, "modified_at": "2025-11-27T12:31:07.733Z" } } } } ]

• 对于包，编排器读取_source.package并提取所有包字段；
• operation: "upsert"→ElementOperation::Upsert；
• operation: "delete"→ElementOperation::Delete。

源码级细节：inferIndex（main.cpp 第 551-582 行）实现了索引自动推断——条目显式给出index时直接使用；否则根据 payload 内容判断：payload.package.hotfix→ 热补丁索引、payload.package→ 包索引、payload.host→ OS 索引。每条 DataValue 通过 MessageBuilder::buildDataValue 构造，将完整 payload JSON 作为原始字节写入data字段。

data_context[]数组

每个条目成为一条DataContext消息。没有 operation 标志，全部按 upsert 处理，用于补充 OS/上下文数据。

示例（OS 上下文文档）：

"data_context": [ { "payload": { "_index": "wazuh-states-inventory-system", "_id": "wazuh_001_f033cfe690b80a478fb4c832934f0fd55927c351", "_score": 2, "_source": { "wazuh": { "agent": { "id": "001", "name": "ubuntu22", "version": "v5.0.0", "groups": ["default"], "host": { "architecture": "aarch64", "hostname": "ubuntu22", "os": { "name": "Ubuntu", "platform": "ubuntu", "type": "linux", "version": "22.04.5 LTS (Jammy Jellyfish)" } } } }, "checksum": { "hash": { "sha1": "996d7206f3c607c7aa377702d65c1b53a80d2a84" } }, "host": { "architecture": "aarch64", "hostname": "ubuntu24", "os": { "build": null, "codename": "noble", "distribution": { "release": null }, "full": null, "kernel": { "name": "Linux", "release": "6.8.0-71-generic", "version": "#71-Ubuntu SMP PREEMPT_DYNAMIC Tue Jul 22 16:44:45 UTC 2025" }, "major": "24", "minor": "04", "name": "Ubuntu", "patch": "2", "platform": "ubuntu", "type": "linux", "version": "24.04.2 LTS (Noble Numbat)" } }, "state": { "document_version": 1, "modified_at": "2025-11-27T11:57:48.873Z" } } } } ]

• 编排器读取_source.host.os填充OsContextData；
• Start 已提供基础 OS 字段，DataContext OS 文档补齐缺失细节（kernel、codename、major/minor 等）。

源码级细节：DataContext 通过 MessageBuilder::buildDataContext 构造，与 DataValue 的区别是不含 operation 字段，消息类型为MessageType_DataContext。另外仓库示例还展示了 DataContext 与wazuh-states-inventory-packages索引组合的用法（如 INPUT_000.json 中补充内核镜像linux-image、zlib1g、libssl3、ubuntu-pro-client等包上下文），用于模拟包安装/卸载场景。

扫描类型

由Start.option控制：

此外mode字段（full/delta）控制会话模式：full对应Mode_ModuleFull（如 INPUT_000 使用），delta对应Mode_ModuleDelta（如 INPUT_001 使用），分别代表全量同步与增量同步两种 InventorySync 会话。

工作流与消息流

1. 初始化阶段

[Test Tool] → Initialize components ├─► RouterModule::start() ├─► InventorySync::start() └─► VulnerabilityScanner::start()

关键动作：

• RouterModule 在queue/inventory-states创建 UNIX 域套接字——该主题常量INVENTORY_SYNC_TOPIC {"inventory-states"}在 inventorySyncFacade.hpp 第 45 行 中定义；
• InventorySync 打开 RocksDB 数据库（路径queue/inventory_sync，见 inventorySyncFacade.hpp 第 44 行）；
• VulnerabilityScanner 加载 CVE feed 数据库；
• ResponseServer 绑定queue/sockets/ar等待 ack 消息（对应 responseDispatcher.hpp 第 46 行 的ARQUEUE_PATH）。

源码级细节：main.cpp 的main函数 依次启动RouterModule::instance().start()、RouterProvider（主题inventory-states）、InventorySync::instance().start(...)、VulnerabilityScannerFacade::instance().start(...)，再启动FakeReportServer与ResponseServer。值得注意的特殊处理是vdTesttoolSkipPostUpdateScan()弱钩子（main.cpp 第 34-37 行）——返回true跳过 feed 更新后自动触发的全量扫描，避免测试过程中被意外全量扫描干扰。

2. Start 消息

[Test Tool] ───Start(FlatBuffer)──► [RouterModule] │ ▼ [InventorySync] │ ├─► Create scan session ├─► Store basic context in RocksDB └──StartAck──► [ResponseServer]

Start 消息直接由 JSON 文件中的Start对象构建。概念结构如下：

Start { agentid: "001", module: "syscollector", mode: Mode_ModuleDelta, option: Option_VDSync, size: <data_values + data_context>, index: ["wazuh-states-inventory-system", "wazuh-states-inventory-packages"], 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"] }

ResponseDispatcher 协议（manager 侧，供参考）：

"(msg_to_agent) [] N!s <agentId> <size> <module>_sync <flatbuffer>" Example: "(msg_to_agent) [] N!s 001 48 syscollector_sync <binary_data>"

源码级细节：ResponseServer::handleMessage 先判断消息是否以(msg_to_agent)开头——若是则按上述文本协议解析出 FlatBuffer 载荷（提取_sync后的二进制段），否则视为纯 FlatBuffer 消息直接校验；随后用flatbuffers::Verifier校验并用Wazuh::SyncSchema::GetMessage反序列化，按content_type()分派到handleStartAck/handleEndAck/handleReqRet。handleStartAck会记录 session ID 并通过std::promise解除主流程的阻塞等待。

3. 数据消息（OS、包、热补丁）

[Test Tool] ───DataValue(Package)──► [RouterModule] ──► [InventorySync] │ [Test Tool] ───DataValue(Hotfix)───► [RouterModule] ──► [InventorySync] │ [Test Tool] ───DataContext(OS)─────► [RouterModule] ──► [InventorySync] │ ▼ Store in RocksDB

消息格式（概念）：

DataValue { session: <StartAck.session>, seq: <0..N-1>, operation: INSERT | UPSERT | DELETE, index: "wazuh-states-inventory-packages" / "wazuh-states-inventory-hotfixes", data: <JSON payload stored as raw bytes> } DataContext { session: <StartAck.session>, seq: <...>, index: "wazuh-states-inventory-system" / ..., data: <JSON payload stored as raw bytes> }

InventorySync 将这些文档写入 RocksDB。随后 VD 读取并构建ScanContext：

• Agent 数据来自Start；
• OS 数据来自Start+DataContext (OS)；
• 包与操作来自DataValue (packages)；
• 热补丁来自DataValue/DataContext (hotfix index)。

源码级细节：发送循环位于 main.cpp 的sendEvent函数——发送 Start 后等待最多 10 秒的 StartAck（超时抛异常Timeout waiting for StartAck），随后按序发送 DataValue（seq递增，operation == "delete"时映射为Operation_Delete，否则Operation_Upsert）与 DataContext（无 operation），最后发送 End。

4. End 消息与漏洞扫描

[Test Tool] ───End(FlatBuffer)─────► [RouterModule] │ ▼ [InventorySync] │ ├─► Finalize session ├─► Trigger VD scan │ ▼ [VulnerabilityScanner] │ ├─► Build ScanContext from RocksDB ├─► Scan packages for CVEs ├─► Generate ECS events ├─► Send to Indexer │ └──EndAck───► [ResponseServer]

漏洞扫描流水线：

1. 包加载：从 RocksDB 读取库存文档；
2. OS 解析：合并 Start 与 OS DataContext 中的 OS 数据；
3. CNA/Feed 解析：确定 CVE feed 来源（NVD、Debian、Ubuntu 等）；
4. 版本匹配：比较已安装版本与受影响版本；
5. 平台与厂商校验：确保 OS 与包厂商匹配 CVE 条件；
6. 热补丁验证：检查 CVE 是否已被已安装热补丁修复（Windows）；
7. 事件生成：构建 ECS 格式漏洞事件；
8. Indexer 分发：发送 ECS 事件到 OpenSearch。

5. 完成与统计

[ResponseServer] ─── Receives EndAck ───► [Test Tool] │ ▼ Print scan statistics Show vulnerability count Exit

预期输出示例：

[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!

源码级细节：工具支持两种输入模式——传入文件时只处理单个 JSON，传入目录时（S_IFDIR判断，见 main.cpp 第 1087-1121 行）会遍历目录下所有.json文件依次执行sendEvent。每次迭代前会重置std::promise与 ack 标志，实现同一进程内的多会话测试。

使用示例

命令行参数

CLI完全由 JSON 输入驱动——不再需要以位置参数传agent_id、mode或option，这些全部位于Start对象中。

inventory_sync_testtool [FLAGS] FLAGS: --input <file> Input JSON file with test data (required) --config <file> Configuration file with indexer settings (required) --wait <seconds> Wait time after End message (default: 10) --verbose Enable verbose logging

源码级细节：parseArgs（main.cpp 第 799-834 行）还支持可选的--logFile <file>参数（将 InventorySync/VD 日志追加写入文件）以及目录形式的--input；默认等待时间在源码中为 30 秒（TestConfig::waitTime = 30），README 中标注的 10 秒为文档默认描述，运行时可通过--wait显式覆盖。

运行示例：

./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

构建产物路径由 testtool/CMakeLists.txt 的RUNTIME_OUTPUT_DIRECTORY指定为${CMAKE_BINARY_DIR}/bin，即src/build/bin/inventory_sync_testtool。

VSCodelaunch.json示例

适用于 gdb 断点调试的配置：

{ "version": "0.2.0", "configurations": [ { "name": "(gdb) InventorySync VD Test - Session from JSON", "type": "cppdbg", "request": "launch", "program": "${workspaceFolder}/src/build/bin/inventory_sync_testtool", "args": [ "--input", "${workspaceFolder}/src/wazuh_modules/inventory_sync/testtool/test_data/INPUT_000.json", "--config", "${workspaceFolder}/src/wazuh_modules/inventory_sync/testtool/test_data/config.json", "--wait", "15", "--verbose" ], "stopAtEntry": false, "cwd": "${workspaceFolder}", "environment": [], "externalConsole": false, "MIMode": "cppdbg", "setupCommands": [ { "description": "Enable pretty-printing for gdb", "text": "-enable-pretty-printing", "ignoreFailures": true } ], "preLaunchTask": "build", "miDebuggerPath": "/usr/bin/gdb" } ] }

调试与排障

常见问题

1. 未收到 StartAck/EndAck

症状：

[ERROR] Timeout waiting for StartAck

可能原因：

• InventorySync 未正确运行；
• socket 路径错误（queue/sockets/ar）；
• ResponseDispatcher 未发送消息；
• Start 消息格式错误（检查 JSON 中的Start对象）。

2. Indexer 连接失败

症状：

[ERROR] Failed to connect to indexer [ERROR] SSL certificate verification failed

原因：

• Indexer URL 错误；
• SSL 证书无效或缺失；
• Indexer 处于 red/yellow 状态或不可用。

3. FlatBuffer 解析错误

症状：

[WARN] Invalid FlatBuffer message, skipping [WARN] Failed to iterate JSON: INSUFFICIENT_PADDING

原因：

• 消息格式错误；
• 构建 JSON payload 时 simdjson padding 问题；
• 之前运行留下的 RocksDB 数据损坏。

4. 未发现漏洞（假阴性）

症状：

[INFO] Agent '001' - Scan completed: 0 vulnerabilities found

排查步骤：

1. 检查 CTI 数据库：确保 CVE feed 已填充且为最新；

2. 开启调试日志：以--verbose运行并查看 VD 日志；

3. 核对包数据：确认data_values中的版本、厂商、平台；

4. 检查版本匹配逻辑：

   [DEBUG] Analyzing CVE: CVE-2022-23498 - Package 'grafana' (v.8.5.5) is NOT vulnerable [DEBUG] Installed version is HIGHER than affected version '8.5.3'

注意事项与限制

重要提示

1. 有状态测试工具在 RocksDB 中维护状态。多次使用同一 agent/会话运行测试可能需要清理数据库：

   rm -rf /tmp/wazuh_inventorysync_test.db

2. Socket 清理UNIX 域套接字在崩溃后可能残留：

   rm -f queue/inventory-states queue/sockets/ar

3. Indexer 索引运行测试前确保 inventory 索引存在且健康（见前置条件）。

4. CVE Feed漏洞检测需要填充完整的 CVE feed 数据库（ADP/NVD + 厂商 feed）。

5. Agent ID 唯一性并行测试时使用不同 agent ID，避免 RocksDB 与 Indexer 中的冲突。

已知限制

1. 无真实 Agent 通信该工具仅模拟 agent 消息，不连接真实 agent。

2. 单 Agent 流程主要为单 agent 测试设计。多 agent 场景需要分开运行或自行修改。

3. Feed 更新CVE feed 更新不会自动触发，需使用 VulnerabilityScanner 的 feed 更新机制单独处理。

4. 网络策略在防火墙规则严格的环境下可能失败（需要与 Indexer 的网络连通）。

5. Windows 热补丁覆盖基于热补丁的修复测试需要 feed 中存在 Windows 特定的 CVE 数据。

总结

inventory_sync_testtool以"单 JSON 文件驱动一次完整同步会话"的方式，把 InventorySync 与 VulnerabilityScanner 两大模块的端到端链路收敛为一条可复现、可调试、可断点的测试路径。它复用了生产级的 FlatBuffer 协议、RocksDB 状态存储与 Indexer 分发逻辑，因此在开发、回归验证与 CI 场景中都能真实反映模块行为。上手时只需三步：准备可用的 Indexer 与证书（参考 test_data/config.json）、编写或复用一份输入 JSON（参考 test_data/INPUT_000.json）、执行inventory_sync_testtool --input ... --config ...并观察 StartAck/EndAck 与漏洞统计输出即可。

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