ZoreMQ Transport Protocol是一個傳輸層協(xié)議，用于ZMQ的連接的信息交互，本文檔描述的是3.0協(xié)議，主要分析基于NULL Security Mechanism

---

協(xié)議語法

ZMTP由三部分組成，分別是 greeting、handshake、traffic

部分	描述	構(gòu)成
greeting	描述ZMQ版本、安全機(jī)制等	signature + version + mechanism + as-server + filler
handshake	描述端類型，如 PUB/SUB，REQ/REP	一個或多個command
traffic	命令或者消息	command

ZMTP Wireshark 抓包

WireShark 默認(rèn)不提供ZMTP解析插件，需要自己配置，步驟如下：

插件倉庫：https://github.com/whitequark/zmtp-wireshark

下載插件：https://github.com/whitequark/zmtp-wireshark/blob/master/zmtp-dissector.lua

將插件zmtp-dissector.lua放到WireShark安裝目錄，比如我的是：C:\Program Files\Wireshark

修改C:\Program Files\Wireshark\init.lua，在文件末尾添加

dofile(DATA_DIR.."zmtp-dissector.lua")

對基于TCP端口通訊ZMQ進(jìn)行抓包，例如端口為7380，將該端口Decode As ZMTP

解析接如下

greeting

greeting 固定64個字節(jié)大小，下面將依次介紹每個部分。

signature

固定10字節(jié)大小，固定值為ff 00 00 00 00 00 00 00 01 7f;

signature可以用來校驗(yàn)鏈接是否為ZMQ鏈接，連續(xù)讀取10個字節(jié)，判斷開頭是否為0xff，結(jié)尾是否為0x7f。

version

固定2字節(jié)大小，格式為{major_version, minor_version}3.0 協(xié)議則為03 00，實(shí)際編碼過程中只會校驗(yàn)major_version;

mechanism

固定20字節(jié)大小，這里只介紹NULL Security Mechanism，也就是不校驗(yàn)，其值為NULL,剩余以內(nèi)容填充0；

as-server

固定一個字節(jié)大小，0x00 或者 0x01 ，當(dāng)mechanism為NULL時候，as-server必須為0。

filler

填充greeting至64個字節(jié)。

抓包示意

由Wireshark解析過后的協(xié)議。

Frame

在greeting之后的所有數(shù)據(jù)格式都為Frame，包含command和message。

frame的格式如下:

Frame = Flag + Payload Length + Payload

抓包示意如下

• Flag
  Flag 為1字節(jié)大小，每位代表不同的意思，參考抓包解釋
  
  低1位：表示是否有更多Frame，這里用于ZMQ中sendmore屬性
  低2位：表示長度是否為8字節(jié)長度，否則為1字節(jié)長度
  低3位：表示當(dāng)前frame是否為Command
  其他：保留，為0

• Payload Length
  數(shù)據(jù)長度，可以為1字節(jié)或者8字節(jié)大小，根據(jù)Flag中的標(biāo)志位決定

• Payload
  實(shí)際的數(shù)據(jù)，大小為Payload Length。

handshake

此階段用來交換對端的READY命令以及metadata，主要包含對端的類型。handshake本質(zhì)是Command，為Frame的一種。
在NULL Security Mechanism機(jī)制中，以PUB/SUB模式為例，handshake的數(shù)據(jù)如下：

Payload內(nèi)容如下：

[1 byte] Command size + [n bytes]Command Name + [1 bytes]Metadata Key size + [n bytes]Metadata Key + [4 bytes]Metadata Value size + [n bytes]Metadata Value

使用Socket實(shí)現(xiàn)ZMQ SUB方法

代碼如下：

//
//  ZMTP 3.0 debugging subscriber
#define WIN32_LEAN_AND_MEAN
#include <Windows.h>
#include <WinSock2.h>
#include <ws2tcpip.h>#include <stdio.h>
#include <stdlib.h>
#include <stdbool.h>
#include <string.h>
#include <assert.h>
#include <fcntl.h>
#include <cstdint>
#include <iostream>#pragma comment(lib, "ws2_32.lib")typedef struct
{uint8_t flags;     //  Must be zerouint8_t size;      //  Size, 0 to 255 byteuint8_t data[255]; //  Message data
} zmtp_msg_t;static void derp(char *s)
{perror(s);exit(1);
}static void tcp_send(int handle, void *buffer, size_t len)
{if (send(handle, (char *) buffer, len, 0) == -1)derp((char *) "send");
}static void tcp_recv(int handle, void *buffer, size_t len)
{printf(" - reading %d bytes: ", (int) len);fflush(stdout);size_t len_recd = 0;while (len_recd < len){size_t bytes = recv(handle, (char *) buffer + len_recd, len - len_recd, 0);if (bytes == 0)break; //  Peer has shutdownprintf(" [%d]", (int) bytes);fflush(stdout);if (bytes == -1)derp((char *) "recv");len_recd += bytes;}printf("\n");fflush(stdout);
}static void zmtp_recv(int handle, zmtp_msg_t *msg)
{tcp_recv(handle, (uint8_t *) msg, 2);tcp_recv(handle, msg->data, msg->size);
}static void zmtp_send(int handle, zmtp_msg_t *msg)
{tcp_send(handle, (uint8_t *) msg, msg->size + 2);
}//  This is the 3.0 greeting (64 bytes)
typedef struct
{uint8_t signature[10];uint8_t version[2];uint8_t mechanism[20];uint8_t as_server[1];uint8_t filler[31];
} zmtp_greeting_t;int main(void)
{puts("I: starting subscriber");WSADATA wsData;if (WSAStartup(MAKEWORD(2, 2), &wsData) != 0){std::cerr << "無法初始化Winsock" << std::endl;return 1;}//  Create TCP socketint peer;if ((peer = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) == -1)derp((char *) "socket");const char *serverIP   = "127.0.0.1";const int   serverPort = 5559;sockaddr_in serverAddress {};serverAddress.sin_family = AF_INET;serverAddress.sin_port   = htons(serverPort);if (inet_pton(AF_INET, serverIP, &(serverAddress.sin_addr)) <= 0){std::cerr << "無效的服務(wù)器IP地址" << std::endl;closesocket(peer);WSACleanup();return 1;}//  Keep trying to connect until we succeedputs("I: waiting for connection");while (connect(peer, reinterpret_cast<sockaddr *>(&serverAddress), sizeof(serverAddress)) == -1)Sleep(1);puts("I: connected OK");//  This is our greeting (64 octets)zmtp_greeting_t outgoing = {{0xFF, 0, 0, 0, 0, 0, 0, 0, 1, 0x7F},{3, 0},{'N', 'U', 'L', 'L', 0},{0},{0}};//  Do full backwards version detection following RFC23//  Send first ten bytes of greeting to peertcp_send(peer, &outgoing, 10);//  Read first byte from peerzmtp_greeting_t incoming;tcp_recv(peer, &incoming, 1);uint8_t length = incoming.signature[0];if (length != 0xFF){puts("E: signature not valid (1)");closesocket(peer);exit(0);}//  Looks like 2.0+, read 9 more bytes to be suretcp_recv(peer, (uint8_t *) &incoming + 1, 9);if ((incoming.signature[9] & 1) != 1){puts("E: signature not valid (2)");closesocket(peer);exit(0);}//  Exchange major version numbersputs("I: signature valid, exchanging major versions");tcp_send(peer, (uint8_t *) &outgoing + 10, 1);tcp_recv(peer, (uint8_t *) &incoming + 10, 1);if (incoming.version[0] >= 3){//  If version >= 3, the peer is using ZMTP 3.0, so send//  rest of the greeting and continue with ZMTP 3.0.puts("I: peer is talking ZMTP 3.0");puts("I: sending rest of greeting...");tcp_send(peer, (uint8_t *) &outgoing + 11, 53);//  Get remainder of greeting from peerputs("I: waiting for greeting from peer...");tcp_recv(peer, (uint8_t *) &incoming + 11, 53);//  Do NULL handshake - send READY command//  For now, empty dictionaryputs("I: have full greeting from peer");zmtp_msg_t  ready = {0x04, 0x19};std::string data;data.push_back(0x05);data.append("READY");data.push_back(0x0b);data.append("Socket-Type");int         netByteOrderSize = htonl(3);const char *valueBytes       = reinterpret_cast<const char *>(&netByteOrderSize);data.append(valueBytes, sizeof(netByteOrderSize));data.append("SUB");memcpy(ready.data, data.c_str(), data.size());puts("I: sending READY");zmtp_send(peer, &ready);//  Now wait for peer's READY commandputs("I: expecting READY from peer");zmtp_recv(peer, &ready);//assert(memcmp(ready.data, "READY   ", 8) == 0);puts("I: OK! NULL security handshake completed");puts("I: send sub command");zmtp_msg_t subCmd {0x00, 0x01};subCmd.data[0] = 0x01;zmtp_send(peer, &subCmd);}else{puts("E: major version not valid");closesocket(peer);exit(0);}puts("I: READY, printing messages");while (true){zmtp_msg_t msg;zmtp_recv(peer, &msg);msg.data[msg.size] = 0;puts((char *) msg.data);}closesocket(peer);return 0;
}