Wazuh Engine 源码架构解析wazuh-manager-analysisd 的事件管线、模块分层与启动依赖注入【免费下载链接】wazuhWazuh - The Open Source Security Platform. Unified XDR and SIEM protection for endpoints and cloud workloads.项目地址: https://gitcode.com/GitHub_Trending/wa/wazuhWazuh Manager 的安全事件处理核心wazuh-engine随发行包以wazuh-manager-analysisd守护进程形式交付负责把 agent 上报的原始安全日志解码为 Wazuh Common SchemaWCS、做 GeoIP/IOC/KVDB 富化、按策略policy编排处理并转发到 Wazuh Indexer。本文以 src/engine/source/README.md 为骨架结合 main.cpp 等真实源码完整梳理source/源码树的模块分层、事件数据流、启动阶段的依赖注入顺序与领域术语体系帮助开发者在动手阅读或修改引擎代码前建立一张准确的全局地图。引擎的定位从 remoted 到 Indexer 的中间处理层wazuh-engine是 Wazuh manager 的解码、富化与路由引擎它接收来自remoted及外部生产者的原始安全事件使用用户定义的decoder解码器将其解析为Wazuh Common Schema (WCS)进行富化GeoIP、IOC、KVDB 查询让事件流经一个或多个policy策略最终把归一化后的 JSON 文档转发到 Wazuh Indexer以及可选的文件输出。引擎同时支持**独立模式standalone**运行设置环境变量WAZUH_ENGINE_STANDALONEtrue即可用于开发、测试或作为 wazuh-indexer 内的独立内容处理器/校验器。从源码可以确认这一开关的定义位置——process.hpp 中声明了constexpr auto ENV_ENGINE_STANDALONE WAZUH_ENGINE_STANDALONE;而独立运行的启动脚本 run_engine.sh 正是通过export WAZUH_ENGINE_STANDALONEtrue开启该模式。两种运行方式的差异会体现在日志初始化、索引器连接配置来源等启动分支上见下文“启动与依赖注入”一节。面向操作员的文档配置编写、规则集作者指南、CLI 用法位于 引擎用户手册本文只覆盖开发者/源码树视角每个目录做什么、模块间如何依赖、进程启动时如何组装。事件管线Event Pipeline数据流原文档给出的管线图完整呈现了事件从入口到出口的走向这里原样保留并做逐段解读---------------------- remoted / VD / Others ─►│ httpsrv (events) │ UDS HTTP ingestion ---------┬------------ │ JSON event ▼ ---------------------- │ router/Orchestrator │ fan-out per active policy ----------┬----------- │ ┌──────────────────────┼──────────────────────┐ ▼ ▼ ▼ Policy A (std) Policy B (custom) Tester session │ │ │ └──────────┬───────────┴──────────┬───────────┘ ▼ ▼ ┌────────────────────────────────────────────────┐ │ bk::IController (Rx or Taskflow) │ │ │ │ pre-filter → decoders → pre-enrichment │ │ → enrichment (geo, ioc, kvdb) │ │ → post-filter → outputs │ └──────────────────────┬─────────────────────────┘ │ ▼ wazuh-indexer / file outputs (streamlog)关键语义有三点一条入站事件 ⇒ 每个激活策略一次独立遍历。策略之间互不阻塞标准空间standard与自定义空间custom的策略可以并发处理同一条事件。策略内部解码器按层级组织根解码器root decoder向子解码器分发解码器再被归组到integration集成中。每个解码器恰好属于一个 integration。builder模块负责编译把策略资产编译成表达式树bk后端再把这棵树物化为可执行管线RxCpp 可观察图或 Taskflow DAG 二选一。这条链路在源码中的锚点可以一一对应事件经 UDS HTTP 进入后main.cpp中创建的第二台httpsrv::ServerEvent services注册了POST /events/enriched路由其处理函数api::event::handlers::pushEvent(orchestrator, dumper, agentMetadataCache)直接把事件推入router::Orchestrator见 main.cpp管线末端则统一收敛到wiconnectorIndexer 出口与streamlog文件输出。source/模块分层地图source/目录下的模块按角色分组依赖方向为“下层依赖上层”箭头指向被依赖方。各模块若自带 README深度说明在其自身文档中此处只做索引。当前仓库中这些目录均已确认存在agentcache、api、base、bk、builder、cmcrud、cmstore、cmsync、conf、confremote、defs、dumper、fastmetrics、fastqueue、geo、hlp、httpsrv、iockvdb、iocsync、kvdbstore、logicexpr、logpar、parsec、proto、rawevtindexer、router、scheduler、schemf、store、streamlog、wiconnector、yml外加 main.cpp 与 stackExecutor.hpp。Foundation基础层几乎被所有其他模块使用模块职责base/共享原语日志spdlog、JSON 包装、错误类型base::Error、RespOrError、表达式树base::Term、base::Event、进程与时间工具。无 README接口在 base/include/base/proto/API 的 Protobuf*.proto契约。线上格式是 JSON但 protobuf 是 C handler 与 Python 客户端共享的唯一事实来源yml/yaml-cpp 与 RapidJSON 的相互转换用于配置与内容加载conf/三级配置环境变量 → JSON 文件 → 默认值与类型化校验所有模块启动时读取hlp/类型专属解析器库IP、日期、JSON、CSV 等是解码器的构建基础parsec/无头文件header-only的 parser-combinator 库支撑logpar与logicexprlogicexpr/布尔表达式解析/求值器Shunting-Yard 算法用于check阶段logpar/把声明式日志格式串编译成组合式hlp解析器defs/$variable替换带环检测用于资产定义schemf/WCS schema 与字段类型校验构建期与运行期fastqueue/有界线程安全队列无锁CQueue、互斥StdQueue支持可选速率限制fastmetrics/无锁计数器/仪表/拉取回调周期性 JSON dumpStorage存储层store/ — 可插拔驱动的 JSON 文档存储内置FileDriver即引擎的持久化 KV。cmstore/ — 内容仓库持有 decoder、filter、output、integration、KVDB 与 policy 的命名空间并维护双向 UUID↔名称缓存。kvdbstore/ — 从cmstore物化的内存 KVDB 缓存供 decoder/filter 查询无 handler 持有即过期。iockvdb/ — 基于 RocksDB 的 IOC 数据库支持整体数据库实例的 RCU 风格原子热切换。Compilation execution backend编译与执行后端builder/ — 编译中枢从cmstore读取资产产出可执行的IPolicy表达式树通过BuilderDeps结构体拉入logpar、schemf、kvdbstore、iockvdb、geo、streamlog与wiconnector。无 README公共接口在 builder/include/builder/。bk/ — 两种可互换的执行后端RxCpp 可观察图或 Taskflow DAG支持节点追踪与热加载。Enrichment I/O services富化与 I/O 服务geo/ — MaxMind GeoIP/ASN 查询基于哈希的数据库热加载。streamlog/ — 异步滚动日志通道按大小时间、gzip、保留策略供文件输出与dumper使用。dumper/ — 可开关的原始事件转储器激活时经streamlog落盘。scheduler/ — 优先级线程池任务调度器负责周期性同步与指标刷新。wiconnector/ — Wazuh Indexer 的线程安全客户端事件、策略资源、IOC 与远端配置。这是通往 indexer 的唯一出口。Synchronization remote configuration同步与远端配置confremote/ — 从 indexer 拉取远端运行时配置被拒绝时回滚。cmcrud/ — API 与cmstore变更之间的校验/适配层强制规范化的变更顺序与命名空间导入的原子性。cmsync/ — 周期性从 indexer 同步内容策略变化时热替换 router 路由。iocsync/ — 周期性把 IOC 同步进iockvdb原子热切换。rawevtindexer/ — 可开关的原始预处理前事件取证式索引。Routing runtime路由与运行时router/ — 生产用Routerworker 池 同步式TesterOrchestrator门面。持有事件队列、策略Environment与路由热替换。API gatewayAPI 网关httpsrv/ — 基于 cpp-httplib 的 UDS HTTP 服务器。main中创建两个实例管理 API 与可选的远端事件接收器。api/ — 按域划分的 handler 工厂router、tester、cmcrud、geo、ioccrud、dumper、rawevtindexer、metrics、event。负责 JSON↔protobuf 互转并委托给对应域接口。Entry point入口main.cpp — 进程入口信号/守护进程处理、依赖注入装配、用于 LIFO 关闭的StackExecutor。stackExecutor.hpp — 按构造顺序记录关闭回调执行时倒序LIFO运行。高层模块依赖图原文档的依赖图省略了base、conf、proto等基础库它们到处被使用突出运行时中枢┌──────────────┐ │ api │ (handlers per domain) └──────┬───────┘ │ ┌──────▼───────┐ │ httpsrv │ └──────────────┘ ┌──────────────┐ ┌─────────────────┐ ┌──────────────┐ │ cmsync │───►│ router │◄───│ fastqueue │ └──────┬───────┘ │ (Orchestrator) │ └──────────────┘ │ └────────┬────────┘ │ │ ▼ ▼ ┌──────────────┐ ┌──────────────┐ │ cmcrud │ │ builder │ ─── compilation hub └──────┬───────┘ └──┬───┬───┬───┘ │ │ │ │ ▼ │ │ └────────────► geo, streamlog ┌──────────────┐ │ │ │ cmstore │◄───────┘ └────► logpar ──► hlp ──► parsec └──────┬───────┘ schemf │ kvdbstore ▼ iockvdb ◄── iocsync ┌────────┐ │ store │◄── confremote, rawevtindexer └────────┘ ┌────────────────┐ │ wiconnector │ ──► wazuh-indexer (sole egress) └────────────────┘ ▲ cmsync, iocsync, confremote, rawevtindexer, streamlog, builder四个关键关系值得牢记router是运行时中枢拥有事件队列、worker 线程与路由生命周期cmsync负责热替换其路由。builder是编译中枢所有参与事件处理的依赖都经BuilderDeps汇聚于此。store与cmstore是数据中枢持久化状态schema、允许字段、规则集、同步状态经由它们流动。wiconnector是通往 indexer 的唯一出口一切出站 OpenSearch 流量都经过它。启动与依赖注入main.cpp 的八个装配阶段引擎模块在 main.cpp 中通过std::shared_ptr与StackExecutor装配后者记录拆除回调并按 LIFO 执行关机。构造分阶段进行每个阶段只依赖它之上的阶段部分阶段受conf::key::SERVER_ENABLE_EVENT_PROCESSING门控。逐段对照源码验证如下1. 进程引导。解析命令行选项main.cpp 支持-f前台运行、-t测试配置、-d调试级别可重复、-h帮助随后按模式初始化日志standalone 走logging::getStandaloneLoggingConfig()支持按日/按大小滚动manager 模式则经base::libwazuhshared::init()复用 wazuh-shared 日志并chdir到 Wazuh home非 standalone 模式下若未加-f则goDaemon()守护化。信号处理上SIGINT/SIGTERM仅置位g_shutdown_requestedmain.cppSIGPIPE直接忽略SIG_IGNmain.cpp。2. 配置加载。conf::Conf从etc/wazuh-manager-internal-options.conf加载main.cpp之后所有模块经confManager.getT(key::…)读取。3. 核心数据层。store::StoreFileDriver→cmstore::CMStore→kvdbstore::KVDBManager→iockvdb::KVDBManager(store)→geo::Manager(store, downloader)→fastmetrics::registerManager()→schemf::Schema。其中 schema 从 store 读取schema/engine-schema/0读取失败只告警而不终止——引擎会以“无 schema”降级运行日志提示与 indexer mapping 的一致性不再受保证main.cpp。4. 解析层。hlp::initTZDB(...)初始化时区数据库后构建logpar::Logpar——它需要 store 中的schema/wazuh-logpar-overrides/0文档该文档读取失败会直接抛异常终止启动与 schema 的降级策略不同main.cpp——最后hlp::registerParsers(logpar)。5. 调度与 I/O。scheduler::Scheduler始终创建并且第一个注册进退出栈注释明确说明它必须在所有模块之前终止以确保已调度的任务在关机前停止main.cpp。此后读取enableProcessing开关开启时创建wiconnector::WIndexerConnector并注册队列指标拉取回调INDEXER_QUEUE_SIZE、INDEXER_EVENTS_DROPPED、INDEXER_QUEUE_USAGE_PERCENT再创建streamlog::LogManager(store, scheduler)。6. 编译中枢。组装builder::BuilderDeps携带logpar、kvdbManager、IOCkvdb、geoManager、streamLogger、indexerConnector以及文件输出的streamlog::RotationConfig——基础路径、命名模式、最大大小、缓冲、是否压缩、压缩级别、最大文件数、累积上限均取自conf::key::STREAMLOG_*系列键随后构建builder::Builder(cmStore, schemaValidator, defs, allowedFields, builderDeps, store)与cmcrud::CrudService(cmStore, builder)main.cpp。allowedFields同样从 store 的schema/allowed-fields/0加载缺失时仅告警并退化为“不限制字段”。7. 后台服务受enableProcessing门控。依次创建confremote::ConfRemoteManager、rawevtindexer::RawEventIndexer可经confremote的index_raw_events触发器热重载开关、router::Orchestrator立即启动并注册关闭回调、cmsync::CMSync、iocsync::IocSync经 scheduler 按IOC_SYNC_INTERVAL周期调度interval 为 0 时禁用、Geo 同步任务GEO_SYNC_INTERVAL从 manifest 拉取 GeoLite2-City/ASN 数据库与dumper::Dumper(streamLogger)。8. API 面。创建httpsrv::ServerAPI servicespayload 上限由SERVER_API_PAYLOAD_MAX_BYTES控制且对负值有防回绕校验按域注册 handlermetrics、geo、router、tester、dumper、rawevtindexer、cmcrud、ioccrud、status最后apiServer-start(SERVER_API_SOCKET)。若enableProcessing开启还会创建第二台httpsrv::ServerEvent services作为远端事件接收器监听SERVER_ENRICHED_EVENTS_SOCKETmain.cpp。主循环与关机的一个细节进入运行态后首次内容同步任务cm-sync-task在synchronize()之前调用orchestrator-expandWorkerPool()——目的是让首次同步修改完整的 worker 池而不仅是主 workermain.cpp。此外还有“无可用路由”的状态监控内容同步完成前入站事件会被丢弃首次启动时该情况记 INFO、后续记 WARNING。关机是严格逆序StackExecutor按 LIFO 执行回调——API 服务器先停并 join 客户端连接、后台服务请求 shutdown 并 join、orchestrator 排空队列、streamlog与wiconnector冲刷缓冲、scheduler 停止日志最后拆除。StackExecutor的实现本身极简std::deque存回调execute()从栈顶弹出执行单个回调抛异常不会中断后续回调stackExecutor.hpp。一个值得注意的注册顺序细节wiconnector先注册shutdown()再注册requestShutdown()借助 LIFO 保证破坏性关闭先于协作式关闭执行确保进行中的分页循环中止并释放共享锁main.cpp。从源码可验证的关键参数约束启动过程中对 indexer 连接器参数有硬性范围校验超出即抛异常终止main.cpp这些是调参时的硬边界配置项conf key校验范围说明INDEXER_BULK_MAX_BYTES64 KB ~ 100 MB批量写入最大字节数默认 8 MB见 conf.cpp 中WAZUH_INDEXER_BULK_MAX_BYTES默认值0x1 23INDEXER_FLUSH_INTERVAL1 ~ 3600 秒刷盘间隔0 非法INDEXER_LOGGER_QUEUE_SIZE1 ~ 1024错误日志有界队列长度INDEXER_LOGGER_THREADS1 ~ 16错误日志线程数INDEXER_MAX_RETRY_DELAY1 ~ 3600 秒最大重试退避事件队列侧Orchestrator的入站队列是带字节上限的fastqueue::CQueuerouter::IngestEventEVENT_QUEUE_SIZE/EVENT_QUEUE_EPS/EVENT_QUEUE_MAX_BYTES默认队列长度 131072见 conf.cpp测试事件走独立的StdQueue。内容同步周期CM_SYNC_INTERVAL默认 120 秒conf.cpp。领域术语表引擎使用一套小型但高度专有的词汇贯穿代码、API 与用户手册阅读源码前务必对齐Event事件— 代表一条安全日志行的 JSON 文档携带 agent/cluster 元数据是流经引擎的工作单元。Wazuh Common Schema (WCS)— 所有输出事件必须遵循的权威类型化字段 schema。由 indexer 拥有引擎启动时从 store 获取schema/engine-schema/0。Asset资产— 最小内容单元decoder、filter、output、integration、KVDB、schema以type/name/version寻址例如decoder/aws-cloudtrail/0。Decoder解码器— 解析并归一化事件到 WCS 字段的资产。层级组织root → children归组到 integration 中。Integration集成— 属于同一产品或日志源的解码器 KVDB 的有序组。每个解码器恰好属于一个 integration。Filter过滤器— 对事件的布尔谓词。pre-filter 在解码前丢弃事件post-filter 在事件到达输出前丢弃事件。Output输出— 处理完事件的目的地Wazuh Indexer、文件。随 manager 打包分发不从内容源同步。Policy策略— 命名的处理管线pre-filter → decoders → pre-enrichment → enrichment → post-filter → outputs。多个策略并发运行。Namespace / Space命名空间/空间— 逻辑内容分区。出厂两个空间standardWazuh 维护与custom用户。indexer 是事实来源引擎在本地镜像它。KVDB— 处理期间供 decoder/filter 查询的轻量键值存储。普通 KVDB 按空间隔离IOC 与 Geo 数据库是全局的。Helper辅助函数— 可从 decoder/filter 阶段调用的可复用函数条件类用于check与映射类用于map。Stage阶段— decoder 内部的操作块check布尔、parse|field抽取、normalize含嵌套map。Route / Environment路由/环境— orchestrator 持有的已编译策略的运行时实例。路由可热替换无需重启。代码约定与继续深入的路径全引擎使用C17clang-format与clang-tidy配置在engine/下。每个模块在include/module/暴露IModule接口实现在src/GoogleTest 单元测试在test/src/unit/供其他模块测试使用的 mock 在test/mocks/。依赖一律以std::shared_ptr注入且只在 main.cpp 中装配——模块自身从不实例化依赖。API handler 遵循工厂模式api::xxx::handlers::registerHandlers(...)的统一签名在 main.cpp 中一目了然。继续深入的入口引擎用户手册 — 面向操作员的文档与快速上手附数据流 mermaid 图同目录还有 架构、配置、API 参考 等。router/README.md — 运行时编排、路由生命周期与 tester 语义。builder/include/builder/builder.hpp — 策略编译的公共接口builder 无 README从这里入手。bk/README.md — Rx 与 Taskflow 两种执行后端。CMakeLists.txt —add_executable(wazuh-engine .../main.cpp)与 RPATH 配置说明了 manager 安装$ORIGIN/../lib与独立包$ORIGIN/lib两种运行布局。小结wazuh-engine的源码树组织体现了清晰的“编译—执行—同步”三分法builder把内容资产编译为表达式树bk把树物化为可执行管线router持有运行时路由并可被cmsync热替换而wiconnector把一切出站流量收敛为单一出口。所有装配集中在main.cpp的八个阶段中完成StackExecutor保证 LIFO 安全关机WAZUH_ENGINE_STANDALONEtrue则让同一份代码能以独立进程形态服务于开发、测试与 indexer 内的内容校验场景。掌握这份地图后无论是排查事件丢包、调整队列与批量参数还是新增一个富化维度都可以按图索骥地定位到具体模块。【免费下载链接】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），仅供参考