网络安全网络IDS【免费下载链接】zeekZeek is a powerful network analysis framework that is much different from the typical IDS you may know.项目地址https://gitcode.com/gh_mirrors/ze/zeek点击查看免费下载导读本文以 Zeek 仓库中 base/protocols/mqtt/main.zeek 的官方文档doc/scripts/base/protocols/mqtt/main.zeek.rst为主体结合其源码实现、binpac 解析器与 btest 测试用例完整剖析 Zeek 对 MQTTv3.1.1协议的检测能力。读完本文你将掌握 MQTT 分析模块的日志流设计、ConnectInfo / PublishInfo / SubscribeInfo 三种日志记录的字段语义、QoS 0/1/2 的完成度判定状态机、端口与常量表的可配置方式以及如何通过策略钩子PolicyHook对日志做二次加工。一、模块概览功能定位与加载方式base/protocols/mqtt/main.zeek实现了 Zeek 对MQTTv3.1.1协议的基础分析能力核心产出是三类日志文件。模块归属命名空间MQTT其自身导入base/protocols/mqtt/consts.zeekconsts.zeek后者提供了协议常量定义。整个 MQTT 支持目录的结构如下scripts/base/protocols/mqtt/load.zeek加载入口依次加载consts、main并通过load-sigs加载 DPD 签名scripts/base/protocols/mqtt/main.zeek日志框架、类型定义、事件处理器scripts/base/protocols/mqtt/consts.zeekMQTT 控制报文类型、协议版本、QoS 等级、CONNACK 返回码的常量映射表scripts/base/protocols/mqtt/dpd.sig用于动态端口检测DPD的签名。加载方式为load base/protocols/mqtt__load__.zeek中实际展开为加载consts、main与dpd.sig三部分。DPD 签名非常简洁signature dpd_mqtt { payload /^.{4,7}MQ/ enable mqtt }该签名匹配报文前 4~7 字节之后紧跟MQ字样的载荷用于在没有注册端口的情况下识别 MQTT 流量mqtt为 mqtt.pac 中定义的 analyzer 名称。需要说明的是binpac 解析器src/analyzer/protocol/mqtt/mqtt.pac目前标注为支持v3.1.1不含 v5.0且流量以 datagram 方式送入MQTT_PDU解析单元。二、日志设计三个独立的日志流模块在zeek_init事件中优先级 5注册了三个日志流main.zeekevent zeek_init() priority5 { Log::create_stream(MQTT::CONNECT_LOG, Log::Stream($columnsConnectInfo, $evlog_mqtt, $pathmqtt_connect, $policylog_policy_connect)); Log::create_stream(MQTT::SUBSCRIBE_LOG, Log::Stream($columnsSubscribeInfo, $pathmqtt_subscribe, $policylog_policy_subscribe)); Log::create_stream(MQTT::PUBLISH_LOG, Log::Stream($columnsPublishInfo, $pathmqtt_publish, $policylog_policy_publish)); Analyzer::register_for_ports(Analyzer::ANALYZER_MQTT, ports); }对应的三个Log::ID枚举在模块顶部通过redef enum Log::ID { CONNECT_LOG, SUBSCRIBE_LOG, PUBLISH_LOG }扩展main.zeek。三者与输出文件、列类型的对应关系如下Log::ID日志路径记录类型策略钩子MQTT::CONNECT_LOGmqtt_connect.logMQTT::ConnectInfoMQTT::log_policy_connectMQTT::SUBSCRIBE_LOGmqtt_subscribe.logMQTT::SubscribeInfoMQTT::log_policy_subscribeMQTT::PUBLISH_LOGmqtt_publish.logMQTT::PublishInfoMQTT::log_policy_publish同一连接在三个日志流中各产生一条或多条记录ts、uid、id三个字段是所有记录共有的基础字段用于关联同一会话。三、Redefinable Options可配置端口## Well-known ports for MQTT. const ports { 1883/tcp } redef;MQTT::ports类型为set[port]属性为redef默认值为{ 1883/tcp }。它在zeek_init中被用于Analyzer::register_for_ports(Analyzer::ANALYZER_MQTT, ports)即告知分析框架目的端口为 1883/tcp 的流量应交给 MQTT analyzer。若你的 MQTT 服务运行在非标准端口例如 8883/tcp 的 TLS 端口或内网自定义端口可以在local.zeek中重新定义redef MQTT::ports { 1883/tcp, 8883/tcp };注意由于签名 DPD 机制的存在见上文dpd.sig即使不配置端口Zeek 也可能通过载荷特征识别 MQTT 流量但显式注册端口是更可靠、开销更低的方式。四、核心类型三条日志记录与连接状态4.1 ConnectInfoCONNECT/CONNACK 会话信息MQTT::ConnectInfo记录main.zeek保存 MQTT 连接建立阶段的信息全部字段带log属性ts/uid/id之外均为可选字段类型语义tstime事件发生时间戳uidstring连接唯一 IDidconn_id连接的四元组两端地址/端口proto_namestring协议名称如MQTTproto_versionstring协议版本如3.1.1client_idstring客户端唯一标识connect_statusstring服务器对 CONNECT 请求的响应状态来自 CONNACK 返回码will_topicstring遗愿遗嘱Last Will and Testament消息要发布的主题will_payloadstring遗愿遗嘱消息的载荷字段填充逻辑在mqtt_connect与mqtt_connack事件中完成main.zeekCONNECT 报文解析后填充协议名、版本、client_id、遗嘱主题与载荷CONNACK 返回码通过return_codes表翻译成可读字符串并写入connect_status随后Log::write(CONNECT_LOG, info)落盘。版本与返回码的翻译表定义在 consts.zeek 中const versions { [3] 3.1, [4] 3.1.1, [5] 5.0, } default function(n: count): string { return fmt(unknown-version-%d, n); }; const return_codes { [0] Connection Accepted, [1] Refused: unacceptable protocol version, [2] Refused: identifier rejected, [3] Refused: server unavailable, [4] Refused: bad user name or password, [5] Refused: not authorized, } default function(n: count): string { return fmt(unknown-return-code-%d, n); };两张表都带default兜底函数未识别的取值会被格式化为unknown-version-N/unknown-return-code-N避免脚本报错。4.2 PublishInfoPUBLISH 报文详情MQTT::PublishInfomain.zeek记录一次 PUBLISH 消息的完整信息是三条日志中字段最丰富的一条字段类型语义tstimePUBLISH 消息开始的时间戳uid/idstring/conn_id连接标识from_clientbool消息由本连接客户端发布T还是服务器推送给客户端Fretainbool消息是否要求服务器保留retained messageqosstringQoS 等级的文本描述at most once/at least once/exactly oncestatusstring发布状态默认incomplete_qosQoS 完整交互完成则为oktopicstring发布主题payloadstring消息载荷可能按MQTT::max_payload_size截断payload_lencount载荷实际长度用于在payload被截断时还原真实长度ackbool消息是否被 ACK默认 Frecbool服务器是否发送了 QoS2 的 RECEIVED 报文PUBREC默认 Frelbool客户端是否发送了 QoS2 的 RELEASE 报文PUBREL默认 Fcompbool服务器是否发送了 QoS2 的 COMPLETE 报文PUBCOMP默认 Fqos_levelcount内部用于比较的数值型 QoS 等级默认 0其中ack、rec、rel、comp、qos_level五个字段不带log属性属于内部状态跟踪字段不会输出到日志文件——它们专门服务于下面要讲的 QoS 完成度判定。QoS 文本翻译表consts.zeekconst qos_levels { [0] at most once, [1] at least once, [2] exactly once, } default function(n: count): string { return fmt(unknown-qos-level-%d, n); };4.3 SubscribeInfo订阅/取消订阅MQTT::SubscribeInfomain.zeek同时服务于 SUBSCRIBE 与 UNSUBSCRIBE 两类操作通过action字段区分字段类型语义ts/uid/idtime/string/conn_id基础连接字段actionMQTT::SubUnsub是订阅MQTT::SUBSCRIBE还是取消订阅MQTT::UNSUBSCRIBEtopicsstring_vec被订阅的主题或主题通配符模式列表qos_levelsindex_vec各主题请求的 QoS 等级列表granted_qos_levelcount服务器最终授予的 QoS 等级ackbool服务器是否 ACK 了该请求默认 FMQTT::SubUnsub是一个可重定义的枚举redeftype MQTT::SubUnsub: enum { MQTT::SUBSCRIBE, MQTT::UNSUBSCRIBE, } redef;4.4 State连接级 pub/sub 状态跟踪MQTT::Statemain.zeek是挂载在connection记录上的内部数据结构用于跟踪单个连接的发布/订阅消息状态type State: record { publish: table[count] of PublishInfo optional write_expire5secs expire_funcpublish_expire; subscribe: table[count] of SubscribeInfo optional write_expire5secs expire_funcsubscribe_expire; };两个表均以msg_id报文标识符为主键publish尚未完成记录/落盘的已发布消息subscribe尚未被 ACK 或尚未记录落盘的订阅/取消订阅消息。两张表都带有write_expire5secs与expire_func属性。若某个msg_id在 5 秒内没有写入更新将触发过期函数——这构成了 QoS 交互超时的兜底机制。同时模块通过redef record connection 扩展了内置connection记录main.zeekredef record connection { mqtt: ConnectInfo optional; mqtt_state: State optional; };set_session()函数负责按需初始化这两个字段main.zeek首次调用时以network_time()填充ts、c$uid、c$id并创建空的publish/subscribe表。五、事件与策略钩子5.1 MQTT::log_mqttglobal MQTT::log_mqtt: event(rec: ConnectInfo);该事件在 CONNECT 日志记录被送往日志框架时触发仅针对ConnectInfo即mqtt_connect.log对应的流。通过在脚本中if/event MQTT::log_mqtt挂钩可以在记录落盘前访问并修改它。5.2 三个 PolicyHook钩子类型对应日志流MQTT::log_policy_connectLog::PolicyHookmqtt_connectMQTT::log_policy_publishLog::PolicyHookmqtt_publishMQTT::log_policy_subscribeLog::PolicyHookmqtt_subscribeLog::PolicyHook是 Zeek 日志框架的标准过滤机制典型用法是在local.zeek或策略脚本中redef Log::PolicyHook MQTT::log_policy_publish; # 不可直接赋值见下 # 正确方式扩展钩子处理函数 hook MQTT::log_policy_publish(rec: MQTT::PublishInfo, id: string) { # 例如过滤掉内部测试主题 if ( /^_test\// in rec$topic ) break; }通过break语句可以阻止记录写入通过修改rec字段可以改写输出内容。这是 Zeek 日志框架中所有Log::PolicyHook的统一语义。六、QoS 状态机status 字段如何被判定为 okPublishInfo$status字段的默认值是incomplete_qos只有完整观察到对应 QoS 级别的往返交互才会被置为ok。这一逻辑在mqtt_publish、mqtt_puback、mqtt_pubrec、mqtt_pubrel、mqtt_pubcomp等事件的成对priority5/priority-5处理器中实现main.zeek。各 QoS 级别的判定条件QoS 0at most oncemqtt_publish优先级 5 的处理器中直接pi$statusok优先级 -5 的处理器随即Log::write(PUBLISH_LOG, pi)并删除表中条目。QoS 0 无 ACK 交互因此立即记录。QoS 1at least once需要mqtt_puback确认。优先级 5 的处理器收到 PUBACK 后将pi$ack T若qos_level 1则status ok优先级 -5 的处理器在status ok时落盘并删除条目。QoS 2exactly once需要完整的 PUBREC → PUBREL → PUBCOMP 三次握手。mqtt_pubrec置rec Tmqtt_pubrel置rel Tmqtt_pubcomp在pi$qos_level 2 pi$rec pi$rel pi$comp时置status ok随后优先级 -5 的处理器落盘。若某个 QoS 1/2 消息在 5 秒内未完成交互write_expire5secs会触发过期函数function publish_expire(tbl: table[count] of PublishInfo, idx: count): interval { Log::write(PUBLISH_LOG, tbl[idx]); return 0sec; }publish_expire与subscribe_expiremain.zeek的逻辑一致把尚未落盘的记录直接写入对应日志流返回0sec表示立即删除条目。这样即使客户端断开或丢包观察到的半程 QoS 交互也会以incomplete_qos状态出现在日志中不会静默丢失。subscribe_expire同理兜底写入SUBSCRIBE_LOG。订阅侧的状态转换则相对简单mqtt_subscribe将请求存入subscribe表mqtt_suback在收到服务器授予的 QoS 后写入granted_qos_level、置ack T并立即落盘删除mqtt_unsubscribe/mqtt_unsuback走同样的流程仅action为MQTT::UNSUBSCRIBEmain.zeek。七、底层解析binpac 解析器与事件生成协议解析层位于 src/analyzer/protocol/mqtt/mqtt.pac。它声明了analyzer MQTT withcontext连接由双向流组成每个方向以MQTT_PDU(is_orig)作为 datagram 解析单元并依次%include了connect.pac、connack.pac、publish.pac、puback.pac、pubrec.pac、pubrel.pac、pubcomp.pac、subscribe.pac、suback.pac、unsuback.pac、unsubscribe.pac、disconnect.pac、pingreq.pac、pingresp.pac等命令定义文件位于src/analyzer/protocol/mqtt/commands/目录。解析器在识别出各类型控制报文后向脚本层抛出mqtt_connect、mqtt_connack、mqtt_publish、mqtt_puback、mqtt_pubrec、mqtt_pubrel、mqtt_pubcomp、mqtt_subscribe、mqtt_suback、mqtt_unsubscribe、mqtt_unsuback等事件脚本层的事件处理器再驱动上文描述的日志逻辑。mqtt-protocol.pacsrc/analyzer/protocol/mqtt/mqtt-protocol.pac提供MQTT_PDU等核心解析结构定义负责解析 MQTT 固定报头、剩余长度编码以及各类型报文的可变头部与载荷。八、常量总览consts.zeekconsts.zeekconsts.zeek共定义四张常量表覆盖 MQTT 报文类型1~14 对应 connect 到 disconnect 的全部 14 种控制报文、协议版本3→3.1、4→3.1.1、5→5.0、QoS 等级0→at most once、1→at least once、2→exactly once以及 CONNACK 返回码0~5。所有表均带default匿名函数兜底保证解析到未知数值时仍能安全输出形如unknown-msg-type-N的占位文本。九、测试验证与实战演示仓库自带三个 btest 用例testing/btest/scripts/base/protocols/mqtt/mqtt.test核心功能测试。执行zeek -b -r $TRACES/mqtt.pcap %INPUT并对mqtt_connect.log、mqtt_subscribe.log、mqtt_publish.log三个输出做btest-diff基线比对mqtt-payload-cap.test验证MQTT::max_payload_size对载荷截断的行为对应payload_len字段的设计意图mqtt-payload-cap-dynamic.test动态调整载荷上限后的行为验证。本地复现 MQTT 分析的完整命令zeek -r mqtt.pcap base/protocols/mqtt运行后会在当前目录生成三个文件mqtt_connect.log、mqtt_subscribe.log、mqtt_publish.log。其中mqtt_publish.log是信息量最大的输出典型的记录行TSV 格式字段顺序对应PublishInfo定义包含ts、uid、id、from_client、retain、qos、status、topic、payload、payload_len等列。若观察到的 QoS 1/2 交互不完整status列会显示incomplete_qos若握手完整则为ok。十、扩展实践二次加工日志的策略示例结合log_mqtt事件与log_policy_*钩子可以在不修改核心脚本的前提下扩展 MQTT 分析。例如将client_id提取为独立字段或按主题过滤load base/protocols/mqtt # 1) 在 mqtt_connect 记录落盘前补充自定义字段的示例思路 # ConnectInfo 本身不带 log 的自定义字段时建议通过 Log::create_stream 新建流 # 或直接用 log_policy 钩子改写现有字段。 # 2) 丢弃针对内部测试主题的发布记录 hook MQTT::log_policy_publish(rec: MQTT::PublishInfo, id: string) { if ( /^internal\.test\./ in rec$topic ) break; } # 3) 关注高价值主题的 QoS 2 消息 event MQTT::log_mqtt(rec: MQTT::ConnectInfo) { if ( rec?$client_id ) NOTICE([$noteMQTT::... ]); # 示例示意实际需按 Notice 框架完整构造 }三个策略钩子的处理函数签名均为 Zeek 标准 PolicyHook 形式hook(rec: 对应记录类型, id: string)。所有扩展都应放置在local.zeek或自定义策略脚本中经load引入以保持base/目录原始脚本的纯净。十一、小结base/protocols/mqtt/main.zeek是 Zeek 内置 MQTT v3.1.1 分析能力的核心脚本层与 mqtt.pac 解析器、consts.zeek 常量定义和 dpd.sig 检测签名协同工作。其设计要点可归纳为三条日志流分别覆盖连接建立、发布、订阅三类 MQTT 活动且共享ts/uid/id连接标识便于关联分析QoS 状态机通过State记录中的write_expire5secs表与publish_expire/subscribe_expire兜底函数确保不完整交互也能被记录并明确标记incomplete_qos策略钩子log_policy_connect/log_policy_publish/log_policy_subscribe与事件log_mqtt为上层策略脚本提供了标准的日志改写与过滤入口可配置性集中在MQTT::portsredef端口集合与consts.zeek的常量表扩展非标准端口只需一行redef。以上结论均以当前仓库 scripts/base/protocols/mqtt/main.zeek、consts.zeek 及对应测试用例为直接依据读者可对照源码逐行验证。赞分享网络安全网络IDS【免费下载链接】zeekZeek is a powerful network analysis framework that is much different from the typical IDS you may know.项目地址https://gitcode.com/gh_mirrors/ze/zeek点击查看免费下载相关推荐Zeek MQTT 协议分析详解从 MQTT v3.1.1 解析到 mqtt 日志生成的完整实践指南Zeek MQTT 协议分析详解从 MQTT v3.1.1 解析到 mqtt 日志生成的完整实践指南 导读 MQTTMessage Queuing Tele网络安全网络IDSZeek 的 PostgreSQL 协议分析器postgresql.log 日志深度解析与源码级实现Zeek 的 PostgreSQL 协议分析器postgresql.log 日志深度解析与源码级实现 导读 Zeek 在 7.1 版本起内置了一个基于 Spi网络安全网络IDSZeek Kerberos 协议分析深入解析 base/protocols/krb 包与 kerberos.log 生成机制Zeek Kerberos 协议分析深入解析 base/protocols/krb 包与 kerberos.log 生成机制 导读 Zeek 内置的 base网络安全网络IDS上一篇Apache Paimon核心技术概念解析下一篇3种简单方法免费提升macOS鼠标体验的终极指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Zeek MQTT 协议分析模块深入解析：mqtt.log 日志生成与 QoS 状态机实现
网络安全网络IDS【免费下载链接】zeekZeek is a powerful network analysis framework that is much different from the typical IDS you may know.项目地址https://gitcode.com/gh_mirrors/ze/zeek点击查看免费下载导读本文以 Zeek 仓库中 base/protocols/mqtt/main.zeek 的官方文档doc/scripts/base/protocols/mqtt/main.zeek.rst为主体结合其源码实现、binpac 解析器与 btest 测试用例完整剖析 Zeek 对 MQTTv3.1.1协议的检测能力。读完本文你将掌握 MQTT 分析模块的日志流设计、ConnectInfo / PublishInfo / SubscribeInfo 三种日志记录的字段语义、QoS 0/1/2 的完成度判定状态机、端口与常量表的可配置方式以及如何通过策略钩子PolicyHook对日志做二次加工。一、模块概览功能定位与加载方式base/protocols/mqtt/main.zeek实现了 Zeek 对MQTTv3.1.1协议的基础分析能力核心产出是三类日志文件。模块归属命名空间MQTT其自身导入base/protocols/mqtt/consts.zeekconsts.zeek后者提供了协议常量定义。整个 MQTT 支持目录的结构如下scripts/base/protocols/mqtt/load.zeek加载入口依次加载consts、main并通过load-sigs加载 DPD 签名scripts/base/protocols/mqtt/main.zeek日志框架、类型定义、事件处理器scripts/base/protocols/mqtt/consts.zeekMQTT 控制报文类型、协议版本、QoS 等级、CONNACK 返回码的常量映射表scripts/base/protocols/mqtt/dpd.sig用于动态端口检测DPD的签名。加载方式为load base/protocols/mqtt__load__.zeek中实际展开为加载consts、main与dpd.sig三部分。DPD 签名非常简洁signature dpd_mqtt { payload /^.{4,7}MQ/ enable mqtt }该签名匹配报文前 4~7 字节之后紧跟MQ字样的载荷用于在没有注册端口的情况下识别 MQTT 流量mqtt为 mqtt.pac 中定义的 analyzer 名称。需要说明的是binpac 解析器src/analyzer/protocol/mqtt/mqtt.pac目前标注为支持v3.1.1不含 v5.0且流量以 datagram 方式送入MQTT_PDU解析单元。二、日志设计三个独立的日志流模块在zeek_init事件中优先级 5注册了三个日志流main.zeekevent zeek_init() priority5 { Log::create_stream(MQTT::CONNECT_LOG, Log::Stream($columnsConnectInfo, $evlog_mqtt, $pathmqtt_connect, $policylog_policy_connect)); Log::create_stream(MQTT::SUBSCRIBE_LOG, Log::Stream($columnsSubscribeInfo, $pathmqtt_subscribe, $policylog_policy_subscribe)); Log::create_stream(MQTT::PUBLISH_LOG, Log::Stream($columnsPublishInfo, $pathmqtt_publish, $policylog_policy_publish)); Analyzer::register_for_ports(Analyzer::ANALYZER_MQTT, ports); }对应的三个Log::ID枚举在模块顶部通过redef enum Log::ID { CONNECT_LOG, SUBSCRIBE_LOG, PUBLISH_LOG }扩展main.zeek。三者与输出文件、列类型的对应关系如下Log::ID日志路径记录类型策略钩子MQTT::CONNECT_LOGmqtt_connect.logMQTT::ConnectInfoMQTT::log_policy_connectMQTT::SUBSCRIBE_LOGmqtt_subscribe.logMQTT::SubscribeInfoMQTT::log_policy_subscribeMQTT::PUBLISH_LOGmqtt_publish.logMQTT::PublishInfoMQTT::log_policy_publish同一连接在三个日志流中各产生一条或多条记录ts、uid、id三个字段是所有记录共有的基础字段用于关联同一会话。三、Redefinable Options可配置端口## Well-known ports for MQTT. const ports { 1883/tcp } redef;MQTT::ports类型为set[port]属性为redef默认值为{ 1883/tcp }。它在zeek_init中被用于Analyzer::register_for_ports(Analyzer::ANALYZER_MQTT, ports)即告知分析框架目的端口为 1883/tcp 的流量应交给 MQTT analyzer。若你的 MQTT 服务运行在非标准端口例如 8883/tcp 的 TLS 端口或内网自定义端口可以在local.zeek中重新定义redef MQTT::ports { 1883/tcp, 8883/tcp };注意由于签名 DPD 机制的存在见上文dpd.sig即使不配置端口Zeek 也可能通过载荷特征识别 MQTT 流量但显式注册端口是更可靠、开销更低的方式。四、核心类型三条日志记录与连接状态4.1 ConnectInfoCONNECT/CONNACK 会话信息MQTT::ConnectInfo记录main.zeek保存 MQTT 连接建立阶段的信息全部字段带log属性ts/uid/id之外均为可选字段类型语义tstime事件发生时间戳uidstring连接唯一 IDidconn_id连接的四元组两端地址/端口proto_namestring协议名称如MQTTproto_versionstring协议版本如3.1.1client_idstring客户端唯一标识connect_statusstring服务器对 CONNECT 请求的响应状态来自 CONNACK 返回码will_topicstring遗愿遗嘱Last Will and Testament消息要发布的主题will_payloadstring遗愿遗嘱消息的载荷字段填充逻辑在mqtt_connect与mqtt_connack事件中完成main.zeekCONNECT 报文解析后填充协议名、版本、client_id、遗嘱主题与载荷CONNACK 返回码通过return_codes表翻译成可读字符串并写入connect_status随后Log::write(CONNECT_LOG, info)落盘。版本与返回码的翻译表定义在 consts.zeek 中const versions { [3] 3.1, [4] 3.1.1, [5] 5.0, } default function(n: count): string { return fmt(unknown-version-%d, n); }; const return_codes { [0] Connection Accepted, [1] Refused: unacceptable protocol version, [2] Refused: identifier rejected, [3] Refused: server unavailable, [4] Refused: bad user name or password, [5] Refused: not authorized, } default function(n: count): string { return fmt(unknown-return-code-%d, n); };两张表都带default兜底函数未识别的取值会被格式化为unknown-version-N/unknown-return-code-N避免脚本报错。4.2 PublishInfoPUBLISH 报文详情MQTT::PublishInfomain.zeek记录一次 PUBLISH 消息的完整信息是三条日志中字段最丰富的一条字段类型语义tstimePUBLISH 消息开始的时间戳uid/idstring/conn_id连接标识from_clientbool消息由本连接客户端发布T还是服务器推送给客户端Fretainbool消息是否要求服务器保留retained messageqosstringQoS 等级的文本描述at most once/at least once/exactly oncestatusstring发布状态默认incomplete_qosQoS 完整交互完成则为oktopicstring发布主题payloadstring消息载荷可能按MQTT::max_payload_size截断payload_lencount载荷实际长度用于在payload被截断时还原真实长度ackbool消息是否被 ACK默认 Frecbool服务器是否发送了 QoS2 的 RECEIVED 报文PUBREC默认 Frelbool客户端是否发送了 QoS2 的 RELEASE 报文PUBREL默认 Fcompbool服务器是否发送了 QoS2 的 COMPLETE 报文PUBCOMP默认 Fqos_levelcount内部用于比较的数值型 QoS 等级默认 0其中ack、rec、rel、comp、qos_level五个字段不带log属性属于内部状态跟踪字段不会输出到日志文件——它们专门服务于下面要讲的 QoS 完成度判定。QoS 文本翻译表consts.zeekconst qos_levels { [0] at most once, [1] at least once, [2] exactly once, } default function(n: count): string { return fmt(unknown-qos-level-%d, n); };4.3 SubscribeInfo订阅/取消订阅MQTT::SubscribeInfomain.zeek同时服务于 SUBSCRIBE 与 UNSUBSCRIBE 两类操作通过action字段区分字段类型语义ts/uid/idtime/string/conn_id基础连接字段actionMQTT::SubUnsub是订阅MQTT::SUBSCRIBE还是取消订阅MQTT::UNSUBSCRIBEtopicsstring_vec被订阅的主题或主题通配符模式列表qos_levelsindex_vec各主题请求的 QoS 等级列表granted_qos_levelcount服务器最终授予的 QoS 等级ackbool服务器是否 ACK 了该请求默认 FMQTT::SubUnsub是一个可重定义的枚举redeftype MQTT::SubUnsub: enum { MQTT::SUBSCRIBE, MQTT::UNSUBSCRIBE, } redef;4.4 State连接级 pub/sub 状态跟踪MQTT::Statemain.zeek是挂载在connection记录上的内部数据结构用于跟踪单个连接的发布/订阅消息状态type State: record { publish: table[count] of PublishInfo optional write_expire5secs expire_funcpublish_expire; subscribe: table[count] of SubscribeInfo optional write_expire5secs expire_funcsubscribe_expire; };两个表均以msg_id报文标识符为主键publish尚未完成记录/落盘的已发布消息subscribe尚未被 ACK 或尚未记录落盘的订阅/取消订阅消息。两张表都带有write_expire5secs与expire_func属性。若某个msg_id在 5 秒内没有写入更新将触发过期函数——这构成了 QoS 交互超时的兜底机制。同时模块通过redef record connection 扩展了内置connection记录main.zeekredef record connection { mqtt: ConnectInfo optional; mqtt_state: State optional; };set_session()函数负责按需初始化这两个字段main.zeek首次调用时以network_time()填充ts、c$uid、c$id并创建空的publish/subscribe表。五、事件与策略钩子5.1 MQTT::log_mqttglobal MQTT::log_mqtt: event(rec: ConnectInfo);该事件在 CONNECT 日志记录被送往日志框架时触发仅针对ConnectInfo即mqtt_connect.log对应的流。通过在脚本中if/event MQTT::log_mqtt挂钩可以在记录落盘前访问并修改它。5.2 三个 PolicyHook钩子类型对应日志流MQTT::log_policy_connectLog::PolicyHookmqtt_connectMQTT::log_policy_publishLog::PolicyHookmqtt_publishMQTT::log_policy_subscribeLog::PolicyHookmqtt_subscribeLog::PolicyHook是 Zeek 日志框架的标准过滤机制典型用法是在local.zeek或策略脚本中redef Log::PolicyHook MQTT::log_policy_publish; # 不可直接赋值见下 # 正确方式扩展钩子处理函数 hook MQTT::log_policy_publish(rec: MQTT::PublishInfo, id: string) { # 例如过滤掉内部测试主题 if ( /^_test\// in rec$topic ) break; }通过break语句可以阻止记录写入通过修改rec字段可以改写输出内容。这是 Zeek 日志框架中所有Log::PolicyHook的统一语义。六、QoS 状态机status 字段如何被判定为 okPublishInfo$status字段的默认值是incomplete_qos只有完整观察到对应 QoS 级别的往返交互才会被置为ok。这一逻辑在mqtt_publish、mqtt_puback、mqtt_pubrec、mqtt_pubrel、mqtt_pubcomp等事件的成对priority5/priority-5处理器中实现main.zeek。各 QoS 级别的判定条件QoS 0at most oncemqtt_publish优先级 5 的处理器中直接pi$statusok优先级 -5 的处理器随即Log::write(PUBLISH_LOG, pi)并删除表中条目。QoS 0 无 ACK 交互因此立即记录。QoS 1at least once需要mqtt_puback确认。优先级 5 的处理器收到 PUBACK 后将pi$ack T若qos_level 1则status ok优先级 -5 的处理器在status ok时落盘并删除条目。QoS 2exactly once需要完整的 PUBREC → PUBREL → PUBCOMP 三次握手。mqtt_pubrec置rec Tmqtt_pubrel置rel Tmqtt_pubcomp在pi$qos_level 2 pi$rec pi$rel pi$comp时置status ok随后优先级 -5 的处理器落盘。若某个 QoS 1/2 消息在 5 秒内未完成交互write_expire5secs会触发过期函数function publish_expire(tbl: table[count] of PublishInfo, idx: count): interval { Log::write(PUBLISH_LOG, tbl[idx]); return 0sec; }publish_expire与subscribe_expiremain.zeek的逻辑一致把尚未落盘的记录直接写入对应日志流返回0sec表示立即删除条目。这样即使客户端断开或丢包观察到的半程 QoS 交互也会以incomplete_qos状态出现在日志中不会静默丢失。subscribe_expire同理兜底写入SUBSCRIBE_LOG。订阅侧的状态转换则相对简单mqtt_subscribe将请求存入subscribe表mqtt_suback在收到服务器授予的 QoS 后写入granted_qos_level、置ack T并立即落盘删除mqtt_unsubscribe/mqtt_unsuback走同样的流程仅action为MQTT::UNSUBSCRIBEmain.zeek。七、底层解析binpac 解析器与事件生成协议解析层位于 src/analyzer/protocol/mqtt/mqtt.pac。它声明了analyzer MQTT withcontext连接由双向流组成每个方向以MQTT_PDU(is_orig)作为 datagram 解析单元并依次%include了connect.pac、connack.pac、publish.pac、puback.pac、pubrec.pac、pubrel.pac、pubcomp.pac、subscribe.pac、suback.pac、unsuback.pac、unsubscribe.pac、disconnect.pac、pingreq.pac、pingresp.pac等命令定义文件位于src/analyzer/protocol/mqtt/commands/目录。解析器在识别出各类型控制报文后向脚本层抛出mqtt_connect、mqtt_connack、mqtt_publish、mqtt_puback、mqtt_pubrec、mqtt_pubrel、mqtt_pubcomp、mqtt_subscribe、mqtt_suback、mqtt_unsubscribe、mqtt_unsuback等事件脚本层的事件处理器再驱动上文描述的日志逻辑。mqtt-protocol.pacsrc/analyzer/protocol/mqtt/mqtt-protocol.pac提供MQTT_PDU等核心解析结构定义负责解析 MQTT 固定报头、剩余长度编码以及各类型报文的可变头部与载荷。八、常量总览consts.zeekconsts.zeekconsts.zeek共定义四张常量表覆盖 MQTT 报文类型1~14 对应 connect 到 disconnect 的全部 14 种控制报文、协议版本3→3.1、4→3.1.1、5→5.0、QoS 等级0→at most once、1→at least once、2→exactly once以及 CONNACK 返回码0~5。所有表均带default匿名函数兜底保证解析到未知数值时仍能安全输出形如unknown-msg-type-N的占位文本。九、测试验证与实战演示仓库自带三个 btest 用例testing/btest/scripts/base/protocols/mqtt/mqtt.test核心功能测试。执行zeek -b -r $TRACES/mqtt.pcap %INPUT并对mqtt_connect.log、mqtt_subscribe.log、mqtt_publish.log三个输出做btest-diff基线比对mqtt-payload-cap.test验证MQTT::max_payload_size对载荷截断的行为对应payload_len字段的设计意图mqtt-payload-cap-dynamic.test动态调整载荷上限后的行为验证。本地复现 MQTT 分析的完整命令zeek -r mqtt.pcap base/protocols/mqtt运行后会在当前目录生成三个文件mqtt_connect.log、mqtt_subscribe.log、mqtt_publish.log。其中mqtt_publish.log是信息量最大的输出典型的记录行TSV 格式字段顺序对应PublishInfo定义包含ts、uid、id、from_client、retain、qos、status、topic、payload、payload_len等列。若观察到的 QoS 1/2 交互不完整status列会显示incomplete_qos若握手完整则为ok。十、扩展实践二次加工日志的策略示例结合log_mqtt事件与log_policy_*钩子可以在不修改核心脚本的前提下扩展 MQTT 分析。例如将client_id提取为独立字段或按主题过滤load base/protocols/mqtt # 1) 在 mqtt_connect 记录落盘前补充自定义字段的示例思路 # ConnectInfo 本身不带 log 的自定义字段时建议通过 Log::create_stream 新建流 # 或直接用 log_policy 钩子改写现有字段。 # 2) 丢弃针对内部测试主题的发布记录 hook MQTT::log_policy_publish(rec: MQTT::PublishInfo, id: string) { if ( /^internal\.test\./ in rec$topic ) break; } # 3) 关注高价值主题的 QoS 2 消息 event MQTT::log_mqtt(rec: MQTT::ConnectInfo) { if ( rec?$client_id ) NOTICE([$noteMQTT::... ]); # 示例示意实际需按 Notice 框架完整构造 }三个策略钩子的处理函数签名均为 Zeek 标准 PolicyHook 形式hook(rec: 对应记录类型, id: string)。所有扩展都应放置在local.zeek或自定义策略脚本中经load引入以保持base/目录原始脚本的纯净。十一、小结base/protocols/mqtt/main.zeek是 Zeek 内置 MQTT v3.1.1 分析能力的核心脚本层与 mqtt.pac 解析器、consts.zeek 常量定义和 dpd.sig 检测签名协同工作。其设计要点可归纳为三条日志流分别覆盖连接建立、发布、订阅三类 MQTT 活动且共享ts/uid/id连接标识便于关联分析QoS 状态机通过State记录中的write_expire5secs表与publish_expire/subscribe_expire兜底函数确保不完整交互也能被记录并明确标记incomplete_qos策略钩子log_policy_connect/log_policy_publish/log_policy_subscribe与事件log_mqtt为上层策略脚本提供了标准的日志改写与过滤入口可配置性集中在MQTT::portsredef端口集合与consts.zeek的常量表扩展非标准端口只需一行redef。以上结论均以当前仓库 scripts/base/protocols/mqtt/main.zeek、consts.zeek 及对应测试用例为直接依据读者可对照源码逐行验证。赞分享网络安全网络IDS【免费下载链接】zeekZeek is a powerful network analysis framework that is much different from the typical IDS you may know.项目地址https://gitcode.com/gh_mirrors/ze/zeek点击查看免费下载相关推荐Zeek MQTT 协议分析详解从 MQTT v3.1.1 解析到 mqtt 日志生成的完整实践指南Zeek MQTT 协议分析详解从 MQTT v3.1.1 解析到 mqtt 日志生成的完整实践指南 导读 MQTTMessage Queuing Tele网络安全网络IDSZeek 的 PostgreSQL 协议分析器postgresql.log 日志深度解析与源码级实现Zeek 的 PostgreSQL 协议分析器postgresql.log 日志深度解析与源码级实现 导读 Zeek 在 7.1 版本起内置了一个基于 Spi网络安全网络IDSZeek Kerberos 协议分析深入解析 base/protocols/krb 包与 kerberos.log 生成机制Zeek Kerberos 协议分析深入解析 base/protocols/krb 包与 kerberos.log 生成机制 导读 Zeek 内置的 base网络安全网络IDS上一篇Apache Paimon核心技术概念解析下一篇3种简单方法免费提升macOS鼠标体验的终极指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考