企业官网建设流程全解析

EMQX 畸形首包处理改进：无效 CONNECT 分类与协议提示日志深度解析

【免费下载链接】emqxThe most scalable and reliable MQTT broker for AI, IoT, IIoT and connected vehicles项目地址: https://gitcode.com/gh_mirrors/em/emqx

导读

MQTT 服务端口上经常会出现非 MQTT 流量——浏览器误连、健康检查探测、运维工具扫描，甚至是把 HTTP、SSH、Redis 等协议的客户端直接指到了 1883 端口。EMQX 在 fix-16779 变更中系统性地改进了对这些畸形首包（malformed first packets）的处理：把 CONNECT 之前到达的任何非 MQTT 数据包统一分类为"无效 CONNECT 数据包"（invalid CONNECT packets），并在关闭连接时输出带有协议猜测提示（protocol hints）的日志，帮助运维人员一眼判断"这到底是谁在连接、发来了什么"。本文基于 changes/ee/fix-16779.en.md 展开，结合 emqx_frame.erl、emqx_channel.erl、emqx_connection.erl 的源码实现与测试用例，完整还原这一改进的前因后果与底层机制。

一、问题背景：为什么首包校验如此关键

MQTT 连接建立的第一个数据包必须是 CONNECT 报文。EMQX 的连接解析流程中，第一个到达的字节流会进入 emqx_frame.erl 的parse/2进行帧解析，而validate_connect_first/2承担着"首包必须是 CONNECT"的守门职责：

%% Reject any packet received before CONNECT, while only the fixed header is %% read, so that no body byte is buffered for an unauthenticated connection. validate_connect_first(?CONNECT, _Options) -> ok; validate_connect_first(_Type, #options{expect_connect = false}) -> ok; validate_connect_first(Type, _Options) -> ?PARSE_ERR(#{ cause => unexpected_packet_before_connect, header_type => emqx_packet:type_name(Type) }).

源码位置：apps/emqx/src/emqx_frame.erl#L263-L273

从这段代码可以看到三个关键设计：

1. 只要固定头（fixed header）就能判定：解析器只读取 1 字节固定头即可判断首包类型，不需要缓冲任何 body 字节，因此未经认证的连接不会占用内存缓冲区；
2. 错误信息结构化：错误以 map 形式携带cause => unexpected_packet_before_connect和header_type（如'SUBSCRIBE'、'PUBLISH'），为上层分类统计提供了依据；
3. 严格模式开关：expect_connect选项控制 CONNECT 围栏（fence）是否生效，emqx_frame:update_opts/2在协商完成后会清除该围栏但保留已协商的版本。

二、核心改进一：畸形首包统一分类为无效 CONNECT

在 fix-16779 之前，不同类型的首包错误走的是各自零散的处理路径。本次变更在 emqx_channel.erl 的handle_frame_error/2中为conn_state = idle（即 CONNECT 尚未解析）阶段增加了专门的分支：

%% Frame error before CONNECT is parsed (conn_state still idle). %% This happens when the first packet is not a valid MQTT CONNECT, %% e.g. an HTTP request or other non-MQTT protocol sent to the MQTT port. %% No CONNACK is sent because no MQTT version has been negotiated yet. handle_frame_error( Reason, Channel = #channel{conn_state = idle} ) -> shutdown( shutdown_count(frame_error_kind(Reason, invalid_connect_packet), Reason, Channel), Channel );

源码位置：apps/emqx/src/emqx_channel.erl#L1397-L1408

这段实现体现了几个核心语义：

• 分类统一：只要是在 CONNECT 被成功解析之前发生的帧错误（无论原因是unexpected_packet_before_connect、bad_frame_header还是其他解析错误），都通过frame_error_kind/2归入invalid_connect_packet这一计数类别，而不是散落在frame_error大类中；
• 不发送 CONNACK：由于此时尚未协商出任何 MQTT 版本，协议双方没有共同语言，因此直接关闭连接，绝不回应 CONNACK；
• 连接进程退出原因保持为原子：shutdown/2的关闭原因最终是原子形式的关闭计数，这让连接监督者（supervisor）能维护稳定的 shutdown 计数器，而不是报告为无法识别的 error 级关闭——畸形客户端输入属于客户端问题，不应污染 broker 的错误日志。

对照来看，同样是帧错误，emqx_channel.erl中针对不同连接阶段有不同的处置策略（apps/emqx/src/emqx_channel.erl#L1384-L1448）：

计数器命名的设计意图

frame_error_kind/2决定关闭计数器使用哪个名字（apps/emqx/src/emqx_channel.erl#L1450-L1469）：

frame_error_kind(Reason, _Default) when is_atom(Reason) -> Reason; frame_error_kind(#{cause := frame_too_large}, _Default) -> frame_too_large; frame_error_kind(#{cause := connect_packet_too_large}, _Default) -> connect_packet_too_large; frame_error_kind(#{cause := too_many_user_properties}, _Default) -> too_many_user_properties; frame_error_kind(_Reason, Default) -> Default.

设计要点：

• 计数器名必须来自有界集合（bounded set），否则指标会失控；
• 原子形式的错误直接用原子命名计数器；map 形式的错误默认共享Default，具体原因保留在关闭原因和 trace 中；
• 唯一例外是frame_too_large、connect_packet_too_large、too_many_user_properties——这三者表示"客户端超过了运维配置的某个限额"，是独立的运维信号，必须与"客户端在发垃圾数据"（走invalid_connect_packet等默认分类）区分开。运维人员看计数器时能明确分辨"客户端撞到了你设的限流"与"客户端在发乱码"。

三、核心改进二：日志中的协议提示（protocol hints）

3.1 首包协议猜测器

本次变更最直观的成果体现在日志上。新的guess_first_packet_protocol/1函数（apps/emqx/src/emqx_frame.erl#L275-L285）会对收到的畸形首包做两层信息提取：

guess_first_packet_protocol(<<Type:4, _Flags:4, _/binary>> = Data) -> {Preview, PreviewEncoding} = format_data_prefix(Data, 32), #{ packet_type => emqx_packet:type_name(Type), resemble_protocol => guess_plaintext_protocol(Data), received_prefix => Preview, received_prefix_encoding => PreviewEncoding }; guess_first_packet_protocol(_Data) -> #{}.

它产出的字段包括：

• packet_type：首字节高 4 位解析出的 MQTT 报文类型名（如CONNECT、PUBLISH），用于确认"这确实不是合法 CONNECT"；
• resemble_protocol：最关键的协议猜测结果，见下文；
• received_prefix：收到的数据前缀（最多 32 字节），供人工核对；
• received_prefix_encoding：前缀的编码方式，printable表示可打印 ASCII，hex表示十六进制转储，避免二进制垃圾污染日志可读性。

3.2 协议指纹识别表

guess_plaintext_protocol/1（apps/emqx/src/emqx_frame.erl#L287-L330）取首包前 16 字节做前缀匹配，目前能识别以下协议指纹：

测试 apps/emqx/test/emqx_frame_SUITE.erl#L320-L406 的t_guess_first_packet_protocol/1覆盖了上表中的全部识别分支，包括 8 种 HTTP 方法逐一验证，以及二进制数据（<<0,1,2,3,4>>）应输出resemble_protocol := unknown且前缀以hex编码展示的用例。

3.3 传输层错误转换与日志

连接进程侧，emqx_connection.erl 的handle_sock_error/2在收到 socket 错误时会先调用maybe_log_first_packet_non_mqtt/2（apps/emqx/src/emqx_connection.erl#L1349-L1358）：

maybe_log_first_packet_non_mqtt(emsgsize, #state{channel = Channel}) -> case emqx_channel:info(conn_state, Channel) of idle -> ?SLOG(info, #{ msg => "first_packet_probably_not_mqtt", reason => emsgsize }); _ -> ok end; maybe_log_first_packet_non_mqtt(_Reason, _State) -> ok.

源码位置：apps/emqx/src/emqx_connection.erl#L1397-L1408

当连接仍处于idle（CONNECT 未解析）状态时，emsgsize这类传输层错误会被记录为first_packet_probably_not_mqtt的 info 级日志——这一日志文案本身就是本次变更的一部分：它明确告知运维人员"首个数据包很可能不是 MQTT"，而不是含糊的 socket 错误。

配套的connect_too_large_error/2（apps/emqx/src/emqx_connection.erl#L1373-L1388）处理另一种常见情况：TCP 传输层用packet_size（等于max_connect_packet_size）在解析器看到任何字节之前就拒绝了超大的 CONNECT，gen_tcp上报emsgsize，ssl上报{invalid_packet, Data}。该函数把这两种传输错误统一转换为与流式解析器一致的connect_packet_too_large帧错误，保证 shutdown 计数器不随传输层不同而分裂，同时避免把客户端原始字节带进进程退出原因。

四、配套防线：CONNECT 大小与剩余长度校验

畸形首包不仅包括"非 MQTT 协议"，还包括"伪装成 CONNECT 的超大包"。validate_frame_len/3（apps/emqx/src/emqx_frame.erl#L375-L390）在读取任何 body 字节之前就从固定头校验帧长度：

• CONNECT 报文同时受max_connect_size与max_size双重限制——这是刻意设计：未认证客户端不能让 broker 缓冲一整个max_size的 body，因此 CONNECT 单独收紧到max_connect_packet_size；
• 超限时报connect_packet_too_large（携带limit与received字段），与frame_too_large区分，对应上文的独立计数器；
• 剩余长度（remaining length）解析同样有保护：可变字节整数超过 4 字节上限时报malformed_variable_byte_integer（apps/emqx/src/emqx_frame.erl#L358-L361）。

五、测试验证：行为被逐条固化

本次变更的行为在测试中被系统性地固化，是理解预期语义的最佳教材。

5.1 连接层测试

apps/emqx/test/emqx_connection_SUITE.erl#L248-L318 的t_parse_incoming/1覆盖了五种典型畸形首包：

1. 垃圾字节：严格模式下<<"for_testing">>报bad_frame_header（宽松解析器则会当作部分字节缓冲）；
2. SUBSCRIBE 作为首包：<<16#82, 16#00>>在 idle 状态报unexpected_packet_before_connect，并携带header_type := 'SUBSCRIBE'与resemble_protocol提示——这正是文档所述"日志中增加协议提示"的直接验证；
3. CONNECT 剩余长度为 0：<<16#10, 16#00>>报zero_remaining_len；
4. v3.1.1 CONNECT 密码标志位异常：按 MQTT 规范 [MQTT-3.1.2-22] 严格拒绝，报invalid_password_flag，同时携带proto_ver := 4、proto_name := <<"MQTT">>等字段，便于运维定位具体违规点；
5. 已连接状态下的错误不做提示增强：bad_subqos在 connected 状态下保持原子形式，不做协议猜测等富化处理，避免在正常连接上浪费开销。

测试还通过meck强制emqx_frame:parse抛错，验证异常路径同样被转换为frame_error而不是崩溃。

5.2 协议猜测测试

apps/emqx/test/emqx_frame_SUITE.erl#L320-L406 的t_guess_first_packet_protocol/1逐条验证了 HTTP/HTTP2/PROXY v1/v2/SSH/SMTP/Redis RESP/plain_text/unknown 共 9 类协议的识别指纹。其中对<<?CONNECT:4, 0:4, 0>>的断言（packet_type := 'CONNECT'）说明：即使首包确实是 CONNECT 类型，也能被正确识别，为上层区分"合法 CONNECT 但解析失败"与"完全非 MQTT"提供依据。

5.3 传输层与 WebSocket 测试

• apps/emqx/test/emqx_socket_connection_SUITE.erl#L151-L154：TCP 传输场景验证首包非 CONNECT 时报unexpected_packet_before_connect并携带resemble_protocol；
• apps/emqx/test/emqx_ws_connection_SUITE.erl#L685：WebSocket 场景下 PUBLISH 作为首包同样被拒。

六、运维视角：如何利用这些改进排查问题

fix-16779 的最终价值体现在排障效率上。升级后遇到"1883 端口有异常连接"时，可以从以下路径定位：

1. 看日志：搜索first_packet_probably_not_mqtt，配合resemble_protocol、received_prefix、received_prefix_encoding字段，即可判断是 HTTP 健康检查、Redis 客户端、PROXY 协议配置错误，还是纯粹的二进制垃圾扫描；
2. 看连接关闭计数器：invalid_connect_packet计数器统计所有 CONNECT 前的帧错误，connect_packet_too_large与frame_too_large单独统计限额违规，通过三者的相对数量可以区分"被误连攻击"与"客户端配置了过大的报文"；
3. 看 trace：frame_error的完整 map（含 cause 与具体字段）会通过emqx_trace输出，需要完整报文细节时可在此获取，而连接进程的退出原因保持为原子形式，不会因日志字段缺失而丢失监控语义。

结语

fix-16779 虽然是一个单点修复，却完整覆盖了"识别—分类—记录—统计"四个环节：emqx_frame在解析层识别畸形首包并猜测协议指纹，emqx_channel按连接阶段将错误分类到有界的计数器集合，emqx_connection在传输层补充first_packet_probably_not_mqtt日志并统一超大 CONNECT 的报错路径，三套测试套件把每种预期行为固化成了可回归的断言。对于运维一个暴露在公网、面向海量异构客户端的 MQTT broker 而言，这种"把模糊的 socket 错误变成可读的协议诊断信息"的改进，正是可观测性的基础积累。

【免费下载链接】emqxThe most scalable and reliable MQTT broker for AI, IoT, IIoT and connected vehicles项目地址: https://gitcode.com/gh_mirrors/em/emqx