diff --git a/scapy/contrib/zenoh.py b/scapy/contrib/zenoh.py new file mode 100644 index 00000000000..0363215a107 --- /dev/null +++ b/scapy/contrib/zenoh.py @@ -0,0 +1,1858 @@ +# SPDX-License-Identifier: GPL-2.0-only +# This file is part of Scapy +# See https://scapy.net/ for more information + +# scapy.contrib.description = Zenoh Protocol +# scapy.contrib.status = loads + +""" +Zenoh protocol for Scapy. + +Implements the Zenoh 1.x wire format used for publish/subscribe/query +communication in IoT and edge computing environments. The three protocol +layers defined by Zenoh are supported: + +- *scouting* messages (Scout, Hello) used for peer discovery, usually on + UDP port 7446, +- *transport* messages (Init, Open, Close, KeepAlive, Frame, Fragment, Join, + OAM) which establish and maintain a session, usually on TCP or UDP + port 7447, +- *network* messages (Push, Request, Response, ResponseFinal, Interest, + Declare, OAM) carried inside transport Frames, together with the + *zenoh* messages (Put, Del, Query, Reply, Err) that hold the user payload. + +Transport messages are grouped in batches. On datagram links a batch is the +datagram itself (:class:`ZenohBatch`); on stream links such as TCP every batch +is prefixed with its length as a 16-bit little endian integer +(:class:`ZenohStreamBatch`). + +Example:: + + >>> pkt = IP()/UDP(dport=7446)/ZenohScout(what="Router+Peer") + >>> msg = ZenohPush(key_scope=1, key_suffix="temp")/ZenohPut(data=b"42") + >>> batch = IP()/TCP(dport=7447)/ZenohStreamBatch( + ... messages=[ZenohFrame(sn=1, messages=[msg])]) + +References: + +- https://zenoh.io/ +- https://github.com/eclipse-zenoh/zenoh (``commons/zenoh-protocol`` and + ``commons/zenoh-codec`` hold the normative wire format description) +""" + +from scapy.compat import chb, orb +from scapy.config import conf +from scapy.fields import ( + BitEnumField, + BitField, + BitFieldLenField, + ByteEnumField, + ByteField, + ConditionalField, + Field, + FieldLenField, + FlagsField, + LEShortField, + PacketListField, + XStrLenField, +) +from scapy.layers.inet import TCP, UDP +from scapy.packet import Packet, bind_bottom_up, bind_layers +from scapy.volatile import RandBin, RandNum + +# ============================================================================ +# Constants +# ============================================================================ + +ZENOH_VERSION = 0x09 + +ZENOH_PORT_SCOUTING = 7446 +ZENOH_PORT_TRANSPORT = 7447 + +# Scouting message IDs, bits [4:0] of the header byte +ZENOH_MID_SCOUT = 0x01 +ZENOH_MID_HELLO = 0x02 + +ZENOH_SCOUTING_MID = { + ZENOH_MID_SCOUT: "Scout", + ZENOH_MID_HELLO: "Hello", +} + +# Transport message IDs, bits [4:0] of the header byte. They never collide +# with the network message IDs, which is what allows a Frame to be followed +# by another transport message inside the same batch. +ZENOH_MID_T_OAM = 0x00 +ZENOH_MID_INIT = 0x01 +ZENOH_MID_OPEN = 0x02 +ZENOH_MID_CLOSE = 0x03 +ZENOH_MID_KEEPALIVE = 0x04 +ZENOH_MID_FRAME = 0x05 +ZENOH_MID_FRAGMENT = 0x06 +ZENOH_MID_JOIN = 0x07 + +ZENOH_TRANSPORT_MID = { + ZENOH_MID_T_OAM: "OAM", + ZENOH_MID_INIT: "Init", + ZENOH_MID_OPEN: "Open", + ZENOH_MID_CLOSE: "Close", + ZENOH_MID_KEEPALIVE: "KeepAlive", + ZENOH_MID_FRAME: "Frame", + ZENOH_MID_FRAGMENT: "Fragment", + ZENOH_MID_JOIN: "Join", +} + +# Network message IDs, bits [4:0] of the header byte +ZENOH_MID_INTEREST = 0x19 +ZENOH_MID_RESPONSE_FINAL = 0x1a +ZENOH_MID_RESPONSE = 0x1b +ZENOH_MID_REQUEST = 0x1c +ZENOH_MID_PUSH = 0x1d +ZENOH_MID_DECLARE = 0x1e +ZENOH_MID_N_OAM = 0x1f + +ZENOH_NETWORK_MID = { + ZENOH_MID_INTEREST: "Interest", + ZENOH_MID_RESPONSE_FINAL: "ResponseFinal", + ZENOH_MID_RESPONSE: "Response", + ZENOH_MID_REQUEST: "Request", + ZENOH_MID_PUSH: "Push", + ZENOH_MID_DECLARE: "Declare", + ZENOH_MID_N_OAM: "OAM", +} + +# Zenoh (payload) message IDs, bits [4:0] of the header byte +ZENOH_MID_PUT = 0x01 +ZENOH_MID_DEL = 0x02 +ZENOH_MID_QUERY = 0x03 +ZENOH_MID_REPLY = 0x04 +ZENOH_MID_ERR = 0x05 + +ZENOH_ZENOH_MID = { + ZENOH_MID_PUT: "Put", + ZENOH_MID_DEL: "Del", + ZENOH_MID_QUERY: "Query", + ZENOH_MID_REPLY: "Reply", + ZENOH_MID_ERR: "Err", +} + +# Declaration IDs, bits [4:0] of the declaration header byte +ZENOH_DECL_KEYEXPR = 0x00 +ZENOH_DECL_U_KEYEXPR = 0x01 +ZENOH_DECL_SUBSCRIBER = 0x02 +ZENOH_DECL_U_SUBSCRIBER = 0x03 +ZENOH_DECL_QUERYABLE = 0x04 +ZENOH_DECL_U_QUERYABLE = 0x05 +ZENOH_DECL_TOKEN = 0x06 +ZENOH_DECL_U_TOKEN = 0x07 +ZENOH_DECL_FINAL = 0x1a + +ZENOH_DECLARATION_ID = { + ZENOH_DECL_KEYEXPR: "DeclareKeyExpr", + ZENOH_DECL_U_KEYEXPR: "UndeclareKeyExpr", + ZENOH_DECL_SUBSCRIBER: "DeclareSubscriber", + ZENOH_DECL_U_SUBSCRIBER: "UndeclareSubscriber", + ZENOH_DECL_QUERYABLE: "DeclareQueryable", + ZENOH_DECL_U_QUERYABLE: "UndeclareQueryable", + ZENOH_DECL_TOKEN: "DeclareToken", + ZENOH_DECL_U_TOKEN: "UndeclareToken", + ZENOH_DECL_FINAL: "DeclareFinal", +} + +# WhatAmI, as a 2-bit value in Hello/Init/Join +ZENOH_WHATAMI = { + 0x00: "Router", + 0x01: "Peer", + 0x02: "Client", +} + +# WhatAmI, as a 3-bit interest bitmap in Scout +ZENOH_WHATAMI_FLAGS = ["Router", "Peer", "Client"] + +# SN/ID resolution, 2 bits per field +ZENOH_RESOLUTION = { + 0x00: "8bit", + 0x01: "16bit", + 0x02: "32bit", + 0x03: "64bit", +} + +# Close reasons +ZENOH_CLOSE_REASON = { + 0x00: "Generic", + 0x01: "Unsupported", + 0x02: "Invalid", + 0x03: "MaxSessions", + 0x04: "MaxLinks", + 0x05: "Expired", + 0x06: "Unresponsive", + 0x07: "ConnectionToSelf", +} + +# Interest declaration modes, bits [6:5] of the header byte +ZENOH_INTEREST_MODE = { + 0x00: "Final", + 0x01: "Current", + 0x02: "Future", + 0x03: "CurrentFuture", +} + +# Interest options, one bit each +ZENOH_INTEREST_OPTIONS = [ + "keyexprs", + "subscribers", + "queryables", + "tokens", + "restricted", + "named", + "mapping", + "aggregate", +] + +ZENOH_CONSOLIDATION = { + 0x00: "Auto", + 0x01: "None", + 0x02: "Monotonic", + 0x03: "Latest", +} + +# Extension body encodings, bits [6:5] of the extension header byte +ZENOH_EXT_ENCODING = { + 0x00: "Unit", + 0x01: "Z64", + 0x02: "ZBuf", + 0x03: "Reserved", +} + +# Well-known encoding IDs. Zenoh does not enforce this mapping, it is only +# a convention of the Zenoh API and used here to render a readable name. +ZENOH_ENCODING_ID = { + 0: "zenoh/bytes", + 1: "zenoh/string", + 2: "zenoh/serialized", + 3: "application/octet-stream", + 4: "text/plain", + 5: "application/json", + 6: "text/json", + 7: "application/cdr", + 8: "application/cbor", + 9: "application/yaml", + 10: "text/yaml", + 11: "text/json5", + 12: "application/python-serialized-object", + 13: "application/protobuf", + 14: "application/java-serialized-object", + 15: "application/openmetrics-text", + 16: "image/png", + 17: "image/jpeg", + 18: "image/gif", + 19: "image/bmp", + 20: "image/webp", + 21: "application/xml", + 22: "application/x-www-form-urlencoded", + 23: "text/html", + 24: "text/xml", + 25: "text/css", + 26: "text/javascript", + 27: "text/markdown", + 28: "text/csv", + 29: "application/sql", + 30: "application/coap-payload", + 31: "application/json-patch+json", + 32: "application/json-seq", + 33: "application/jsonpath", + 34: "application/jwt", + 35: "application/mp4", + 36: "application/soap+xml", + 37: "application/yang", + 38: "audio/aac", + 39: "audio/flac", + 40: "audio/mp4", + 41: "audio/ogg", + 42: "audio/vorbis", + 43: "video/h261", + 44: "video/h263", + 45: "video/h264", + 46: "video/h265", + 47: "video/h266", + 48: "video/mp4", + 49: "video/ogg", + 50: "video/raw", + 51: "video/vp8", + 52: "video/vp9", +} + + +# ============================================================================ +# VLE (variable length encoding) codec +# ============================================================================ + +# A zint never spans more than 9 bytes: the 9th byte carries 8 payload bits +# instead of 7, which is enough to cover the remaining bits of an u64. +ZENOH_ZINT_MAX_LEN = 9 + + +def zenoh_zint_encode(val): + # type: (int) -> bytes + """Encode an unsigned integer using the zenoh VLE (``zint``) format.""" + if val < 0: + raise ValueError("zint values must not be negative") + if val > 0xFFFFFFFFFFFFFFFF: + raise ValueError("zint values must fit in 64 bits") + data = bytearray() + while val > 0x7F and len(data) < ZENOH_ZINT_MAX_LEN - 1: + data.append((val & 0x7F) | 0x80) + val >>= 7 + data.append(val & 0xFF) + return bytes(data) + + +def zenoh_zint_decode(s): + # type: (bytes) -> tuple + """Decode a zenoh VLE integer. + + Returns a ``(value, consumed)`` tuple. ``consumed`` is 0 when the buffer + holds a truncated integer, in which case the partially decoded value is + still returned. + """ + value = 0 + for i in range(min(len(s), ZENOH_ZINT_MAX_LEN)): + b = orb(s[i]) + if i == ZENOH_ZINT_MAX_LEN - 1: + return value | (b << (7 * i)), i + 1 + value |= (b & 0x7F) << (7 * i) + if not b & 0x80: + return value, i + 1 + return value, 0 + + +class ZenohZIntField(Field): + """Unsigned integer with the zenoh VLE encoding. + + Each byte carries 7 payload bits, least significant group first, with + bit 7 set on every byte but the last one. For example 300 (0x12C) is + encoded as ``ac 02``. + """ + + def __init__(self, name, default): + Field.__init__(self, name, default, "B") + + def addfield(self, pkt, s, val): + return s + zenoh_zint_encode(self.i2m(pkt, val)) + + def getfield(self, pkt, s): + value, consumed = zenoh_zint_decode(s) + if not consumed: + return b"", value + return s[consumed:], value + + def i2len(self, pkt, val): + return len(zenoh_zint_encode(self.i2m(pkt, val))) + + def randval(self): + return RandNum(0, 0xFFFFFFFF) + + +class ZenohZIntLenField(FieldLenField): + """:class:`ZenohZIntField` computed from the length or count of a field.""" + + def __init__(self, name, default, **kwargs): + kwargs.setdefault("fmt", "B") + FieldLenField.__init__(self, name, default, **kwargs) + + def addfield(self, pkt, s, val): + return s + zenoh_zint_encode(self.i2m(pkt, val)) + + def getfield(self, pkt, s): + value, consumed = zenoh_zint_decode(s) + if not consumed: + return b"", value + return s[consumed:], value + + def i2len(self, pkt, val): + return len(zenoh_zint_encode(self.i2m(pkt, val))) + + +class ZenohZBufField(Field): + """Byte buffer prefixed by its length as a zint (spec ````).""" + + def __init__(self, name, default): + Field.__init__(self, name, default, "B") + + def i2m(self, pkt, x): + if x is None: + return b"" + if isinstance(x, str): + return x.encode() + return bytes(x) + + def addfield(self, pkt, s, val): + val = self.i2m(pkt, val) + return s + zenoh_zint_encode(len(val)) + val + + def getfield(self, pkt, s): + length, consumed = zenoh_zint_decode(s) + if not consumed: + return b"", b"" + return s[consumed + length:], s[consumed:consumed + length] + + def i2len(self, pkt, val): + val = self.i2m(pkt, val) + return len(zenoh_zint_encode(len(val))) + len(val) + + def i2repr(self, pkt, val): + if isinstance(val, bytes): + return val.hex() + return repr(val) + + def randval(self): + return RandBin(RandNum(0, 16)) + + +class ZenohZStrField(ZenohZBufField): + """UTF-8 string prefixed by its length as a zint. + + Buffers that are not valid UTF-8 are kept as bytes so that a dissected + packet still rebuilds to the original bytes. + """ + + def getfield(self, pkt, s): + remain, val = ZenohZBufField.getfield(self, pkt, s) + try: + return remain, val.decode("utf-8") + except UnicodeDecodeError: + return remain, val + + def i2repr(self, pkt, val): + return repr(val) + + def randval(self): + return RandBin(RandNum(0, 16)) + + +class ZenohEncodingField(Field): + """Zenoh ``encoding``: a zint holding ``(id << 1) | S`` and a schema. + + The field value is the encoding ID as an integer, or an + ``(id, schema)`` tuple when the S flag announces a schema. + """ + + def __init__(self, name, default): + Field.__init__(self, name, default, "B") + + def i2m(self, pkt, x): + if x is None: + return (0, None) + if isinstance(x, tuple): + eid, schema = x + if isinstance(schema, str): + schema = schema.encode() + return (eid, schema) + return (x, None) + + def addfield(self, pkt, s, val): + eid, schema = self.i2m(pkt, val) + raw = (eid << 1) | (0x01 if schema is not None else 0x00) + s += zenoh_zint_encode(raw) + if schema is not None: + s += zenoh_zint_encode(len(schema)) + schema + return s + + def getfield(self, pkt, s): + raw, consumed = zenoh_zint_decode(s) + if not consumed: + return b"", 0 + s = s[consumed:] + eid = raw >> 1 + if not raw & 0x01: + return s, eid + length, consumed = zenoh_zint_decode(s) + if not consumed: + return b"", (eid, b"") + return s[consumed + length:], (eid, s[consumed:consumed + length]) + + def i2repr(self, pkt, val): + eid, schema = self.i2m(pkt, val) + name = ZENOH_ENCODING_ID.get(eid, str(eid)) + if schema: + try: + return "%s;%s" % (name, schema.decode("utf-8")) + except UnicodeDecodeError: + return "%s;%s" % (name, schema.hex()) + return name + + def randval(self): + return RandNum(0, 52) + + +class ZenohConsolidationField(ZenohZIntField): + """Query consolidation mode, a zint with well known values.""" + + def i2repr(self, pkt, val): + return ZENOH_CONSOLIDATION.get(val, str(val)) + + +class ZenohTimestampField(Field): + """Zenoh timestamp: an NTP64 counter followed by the ID of its source. + + The field value is a ``(ntp64, source_id)`` tuple; a plain integer is + understood as a timestamp without source ID. NTP64 counts seconds in its + upper 32 bits and fractions of a second in its lower 32 bits. + """ + + def __init__(self, name, default): + Field.__init__(self, name, default, "B") + + def i2m(self, pkt, x): + if x is None: + return (0, b"") + if isinstance(x, tuple): + ntp64, source_id = x + if isinstance(source_id, str): + source_id = source_id.encode() + return (int(ntp64), bytes(source_id)) + return (int(x), b"") + + def addfield(self, pkt, s, val): + ntp64, source_id = self.i2m(pkt, val) + return (s + zenoh_zint_encode(ntp64) + + zenoh_zint_encode(len(source_id)) + source_id) + + def getfield(self, pkt, s): + ntp64, consumed = zenoh_zint_decode(s) + if not consumed: + return b"", (ntp64, b"") + s = s[consumed:] + length, consumed = zenoh_zint_decode(s) + if not consumed: + return b"", (ntp64, b"") + return s[consumed + length:], (ntp64, s[consumed:consumed + length]) + + def i2len(self, pkt, val): + return len(self.addfield(pkt, b"", val)) + + def i2repr(self, pkt, val): + ntp64, source_id = self.i2m(pkt, val) + return "%d.%09d/%s" % ( + ntp64 >> 32, + ((ntp64 & 0xFFFFFFFF) * 10 ** 9) >> 32, + source_id.hex(), + ) + + def randval(self): + return RandNum(0, 0xFFFFFFFFFFFFFFFF) + + +# ============================================================================ +# Extensions +# ============================================================================ + +class ZenohExtension(Packet): + """Zenoh extension, encoded as a type-length-value triple. + + Every zenoh message may be followed by extensions when the Z flag is set + in its header. The extension header byte is:: + + 7 6 5 4 3 2 1 0 + +-+-+-+-+-+-+-+-+ + |Z|ENC|M| ID | + +-+---+-+-------+ + % length % -- if ENC == ZBuf + +---------------+ + ~ [u8] ~ -- if ENC == ZBuf + +---------------+ + + ``Z`` announces a further extension, ``M`` marks the extension as + mandatory (a receiver that does not understand it must drop the message) + and ``ENC`` selects the body encoding: no body, a zint value, or a + length-prefixed buffer. + + The meaning of ``eid`` depends on the enclosing message, e.g. 0x1 is QoS + and 0x2 is a timestamp for most network messages. + + Extensions are the last field of a message unless the diagram of that + message shows them elsewhere, which is the case whenever a payload or a + nested message follows. + """ + name = "ZenohExt" + fields_desc = [ + BitField("more", 0, 1), + BitEnumField("enc", 0, 2, ZENOH_EXT_ENCODING), + BitField("mandatory", 0, 1), + BitField("eid", 0, 4), + ConditionalField(ZenohZIntField("value", 0), + lambda pkt: pkt.enc == 0x01), + ConditionalField(ZenohZBufField("body", b""), + lambda pkt: pkt.enc == 0x02), + ] + + def extract_padding(self, s): + return b"", s + + def default_payload_class(self, payload): + return conf.padding_layer + + +class ZenohExtensionsField(PacketListField): + """List of :class:`ZenohExtension`, chained through their ``more`` bit. + + The ``more`` bit is a structural flag and is therefore recomputed while + building, just like a length field. + """ + + def addfield(self, pkt, s, val): + parts = [self.i2m(pkt, v) for v in val] + parts = [p for p in parts if p] + for i, part in enumerate(parts): + first = orb(part[0]) + if i < len(parts) - 1: + first |= 0x80 + else: + first &= 0x7F + s += chb(first) + part[1:] + return s + + +def _next_extension(pkt, lst, cur, remain): + if not remain: + return None + if cur is None: + return ZenohExtension if pkt.getfieldval("flag_z") else None + return ZenohExtension if cur.more else None + + +def _extensions_field(): + return ZenohExtensionsField("extensions", [], next_cls_cb=_next_extension) + + +# ============================================================================ +# Message bases +# ============================================================================ + +def _is_set(value): + """True when a field holds a value that is meant to go on the wire.""" + if value is None: + return False + if isinstance(value, (bytes, str, list)) and not len(value): + return False + return True + + +def _flag_or_value(flag, name): + """Condition of an optional field announced by a header flag. + + While dissecting, the flag decides. While building, giving the field a + value is enough: :meth:`_ZenohMsg.post_build` then sets the flag, unless + it was set explicitly. + """ + def _cond(pkt): + if pkt.getfieldval(flag): + return True + return _is_set(pkt.fields.get(name)) + return _cond + + +class _ZenohMsg(Packet): + """Base class of all zenoh messages. + + Zenoh messages are siblings inside a batch or a frame, not enclosing + layers of each other, so trailing bytes are handed back to the enclosing + list instead of being dissected as a payload. + """ + + # Flags that announce an optional field, as + # (flag field, announced field, header byte index, bit mask) tuples. + auto_flags = [] # type: list + + def post_build(self, p, pay): + for flag, name, index, mask in self.auto_flags: + if flag in self.fields or len(p) <= index: + continue + if _is_set(self.fields.get(name)): + p = p[:index] + chb(orb(p[index]) | mask) + p[index + 1:] + # The Z flag is bit 7 of the header byte of every zenoh message. + if p and self.extensions and "flag_z" not in self.fields: + p = chb(orb(p[0]) | 0x80) + p[1:] + return p + pay + + def extract_padding(self, s): + return b"", s + + def mysummary(self): + return self.name + + +class _ZenohMsgWithBody(_ZenohMsg): + """Zenoh message that carries another zenoh message as its payload.""" + + def extract_padding(self, s): + return s, b"" + + +def _guess_msg_class(payload, table): + if not payload: + return conf.padding_layer + return table.get(orb(payload[0]) & 0x1F, conf.raw_layer) + + +def _interest_option(pkt, mask): + """True when an Interest carries options and the given one is set.""" + if pkt.mode == 0x00: + return False + options = pkt.options + if options is None: + return False + return bool(int(options) & mask) + + +# ============================================================================ +# Scouting messages +# ============================================================================ + +class ZenohScouting(Packet): + """Dispatcher for zenoh scouting messages (UDP port 7446).""" + name = "ZenohScouting" + fields_desc = [] + + @classmethod + def dispatch_hook(cls, _pkt=None, *args, **kargs): + if _pkt: + return _SCOUTING_MSG_CLASSES.get(orb(_pkt[0]) & 0x1F, + conf.raw_layer) + return cls + + +class ZenohScout(_ZenohMsg): + """Scout message, multicast to discover the zenoh nodes of a network. + + :: + + 7 6 5 4 3 2 1 0 + +-+-+-+-+-+-+-+-+ + |Z|X|X| SCOUT | + +-+-+-+---------+ + | version | + +---------------+ + |zid_len|I| what| + +-+-+-+-+-+-+-+-+ + ~ [u8] ~ if I==1 -- Zenoh ID + +---------------+ + + ``what`` is a bitmap of the node kinds the sender is interested in and + ``zid_len`` holds the Zenoh ID length minus one. + """ + name = "ZenohScout" + fields_desc = [ + BitField("flag_z", 0, 1), + BitField("res1", 0, 1), + BitField("res2", 0, 1), + BitEnumField("mid", ZENOH_MID_SCOUT, 5, ZENOH_SCOUTING_MID), + ByteField("version", ZENOH_VERSION), + BitFieldLenField("zid_len", None, 4, length_of="zid", + adjust=lambda pkt, x: max(x, 1) - 1), + BitField("flag_i", 0, 1), + FlagsField("what", 0x07, 3, ZENOH_WHATAMI_FLAGS), + ConditionalField( + XStrLenField("zid", b"", + length_from=lambda pkt: (pkt.zid_len or 0) + 1), + _flag_or_value("flag_i", "zid"), + ), + _extensions_field(), + ] + auto_flags = [("flag_i", "zid", 2, 0x08)] + + +class ZenohLocator(Packet): + """A single locator of a Hello message, e.g. ``tcp/192.168.1.1:7447``.""" + name = "ZenohLocator" + fields_desc = [ZenohZStrField("locator", "")] + + def extract_padding(self, s): + return b"", s + + def default_payload_class(self, payload): + return conf.padding_layer + + +class ZenohHello(_ZenohMsg): + """Hello message, sent in reply to a Scout or to advertise a node. + + :: + + 7 6 5 4 3 2 1 0 + +-+-+-+-+-+-+-+-+ + |Z|X|L| HELLO | + +-+-+-+---------+ + | version | + +---------------+ + |zid_len|X|X|wai| + +-+-+-+-+-+-+-+-+ + ~ [u8] ~ -- Zenoh ID + +---------------+ + ~ ~ if L==1 -- List of locators + +---------------+ + """ + name = "ZenohHello" + fields_desc = [ + BitField("flag_z", 0, 1), + BitField("res1", 0, 1), + BitEnumField("flag_l", 0, 1, {0: "NoLocators", 1: "Locators"}), + BitEnumField("mid", ZENOH_MID_HELLO, 5, ZENOH_SCOUTING_MID), + ByteField("version", ZENOH_VERSION), + BitFieldLenField("zid_len", None, 4, length_of="zid", + adjust=lambda pkt, x: max(x, 1) - 1), + BitField("res2", 0, 2), + BitEnumField("whatami", 0x01, 2, ZENOH_WHATAMI), + XStrLenField("zid", b"\x01", + length_from=lambda pkt: (pkt.zid_len or 0) + 1), + ConditionalField( + ZenohZIntLenField("num_locators", None, count_of="locators"), + _flag_or_value("flag_l", "locators"), + ), + ConditionalField( + PacketListField("locators", [], ZenohLocator, + count_from=lambda pkt: pkt.num_locators or 0), + _flag_or_value("flag_l", "locators"), + ), + _extensions_field(), + ] + auto_flags = [("flag_l", "locators", 0, 0x20)] + + +# ============================================================================ +# Transport messages +# ============================================================================ + +def _resolution_fields(): + """SN/ID resolution byte and batch size, present when the S flag is set.""" + return [ + ConditionalField(BitField("res_resolution", 0, 4), + lambda pkt: pkt.flag_s), + ConditionalField(BitEnumField("rid_resolution", 0x02, 2, + ZENOH_RESOLUTION), + lambda pkt: pkt.flag_s), + ConditionalField(BitEnumField("fsn_resolution", 0x02, 2, + ZENOH_RESOLUTION), + lambda pkt: pkt.flag_s), + ConditionalField(LEShortField("batch_size", 65535), + lambda pkt: pkt.flag_s), + ] + + +class ZenohInit(_ZenohMsg): + """Init message, the first half of the session establishment handshake. + + ``flag_a`` tells an InitSyn (0, sent by the initiator) from an InitAck + (1, sent by the responder). Only the InitAck carries a cookie, which the + initiator has to echo in its OpenSyn. + + :: + + 7 6 5 4 3 2 1 0 + +-+-+-+-+-+-+-+-+ + |Z|S|A| INIT | + +-+-+-+---------+ + | version | + +---------------+ + |zid_len|x|x|wai| + +-------+-+-+---+ + ~ [u8] ~ -- Zenoh ID of the sender + +---------------+ + |x|x|x|x|rid|fsn| \\ -- SN/ID resolution + +---------------+ | if S==1 + | u16 | / -- Batch size + +---------------+ + ~ ~ if A==1 -- Cookie + +---------------+ + """ + name = "ZenohInit" + fields_desc = [ + BitField("flag_z", 0, 1), + BitField("flag_s", 0, 1), + BitEnumField("flag_a", 0, 1, {0: "Syn", 1: "Ack"}), + BitEnumField("mid", ZENOH_MID_INIT, 5, ZENOH_TRANSPORT_MID), + ByteField("version", ZENOH_VERSION), + BitFieldLenField("zid_len", None, 4, length_of="zid", + adjust=lambda pkt, x: max(x, 1) - 1), + BitField("res1", 0, 2), + BitEnumField("whatami", 0x01, 2, ZENOH_WHATAMI), + XStrLenField("zid", b"\x01", + length_from=lambda pkt: (pkt.zid_len or 0) + 1), + ] + _resolution_fields() + [ + ConditionalField(ZenohZBufField("cookie", b""), + lambda pkt: pkt.flag_a), + _extensions_field(), + ] + + +class ZenohOpen(_ZenohMsg): + """Open message, the second half of the session establishment handshake. + + ``flag_a`` tells an OpenSyn (0) from an OpenAck (1); only the OpenSyn + echoes the cookie received in the InitAck. ``flag_t`` selects the unit of + the lease period: seconds when set, milliseconds otherwise. + + :: + + 7 6 5 4 3 2 1 0 + +-+-+-+-+-+-+-+-+ + |Z|T|A| OPEN | + +-+-+-+---------+ + % lease % + +---------------+ + % initial_sn % + +---------------+ + ~ ~ if A==0 -- Cookie + +---------------+ + """ + name = "ZenohOpen" + fields_desc = [ + BitField("flag_z", 0, 1), + BitEnumField("flag_t", 0, 1, {0: "Milliseconds", 1: "Seconds"}), + BitEnumField("flag_a", 0, 1, {0: "Syn", 1: "Ack"}), + BitEnumField("mid", ZENOH_MID_OPEN, 5, ZENOH_TRANSPORT_MID), + ZenohZIntField("lease", 10000), + ZenohZIntField("initial_sn", 0), + ConditionalField(ZenohZBufField("cookie", b""), + lambda pkt: pkt.flag_a == 0), + _extensions_field(), + ] + + +class ZenohClose(_ZenohMsg): + """Close message, terminating either a single link or the whole session. + + :: + + 7 6 5 4 3 2 1 0 + +-+-+-+-+-+-+-+-+ + |Z|X|S| CLOSE | + +-+-+-+---------+ + | reason | + +---------------+ + """ + name = "ZenohClose" + fields_desc = [ + BitField("flag_z", 0, 1), + BitField("res1", 0, 1), + BitEnumField("flag_s", 0, 1, {0: "Link", 1: "Session"}), + BitEnumField("mid", ZENOH_MID_CLOSE, 5, ZENOH_TRANSPORT_MID), + ByteEnumField("reason", 0, ZENOH_CLOSE_REASON), + _extensions_field(), + ] + + +class ZenohKeepAlive(_ZenohMsg): + """KeepAlive message, refreshing the lease period of a link. + + :: + + 7 6 5 4 3 2 1 0 + +-+-+-+-+-+-+-+-+ + |Z|X|X| KALIVE | + +-+-+-+---------+ + """ + name = "ZenohKeepAlive" + fields_desc = [ + BitField("flag_z", 0, 1), + BitField("res1", 0, 1), + BitField("res2", 0, 1), + BitEnumField("mid", ZENOH_MID_KEEPALIVE, 5, ZENOH_TRANSPORT_MID), + _extensions_field(), + ] + + +class ZenohFrame(_ZenohMsg): + """Frame message, carrying one or more complete network messages. + + :: + + 7 6 5 4 3 2 1 0 + +-+-+-+-+-+-+-+-+ + |Z|X|R| FRAME | + +-+-+-+---------+ + % seq num % + +---------------+ + ~ [FrameExts] ~ if Z==1 + +---------------+ + ~ [NetworkMsg] ~ + +---------------+ + + Network messages are collected in the ``messages`` list. Since network + and transport message IDs cannot collide, dissection stops as soon as a + byte that is not a network message ID is found, which leaves room for a + further transport message in the same batch. + """ + name = "ZenohFrame" + fields_desc = [ + BitField("flag_z", 0, 1), + BitField("res1", 0, 1), + BitEnumField("flag_r", 0, 1, {0: "BestEffort", 1: "Reliable"}), + BitEnumField("mid", ZENOH_MID_FRAME, 5, ZENOH_TRANSPORT_MID), + ZenohZIntField("sn", 0), + _extensions_field(), + PacketListField("messages", [], + next_cls_cb=lambda pkt, lst, cur, remain: + _next_network_msg(remain)), + ] + + def mysummary(self): + return "ZenohFrame sn=%s" % self.sn + + +class ZenohFragment(_ZenohMsg): + """Fragment message, carrying a piece of an oversized network message. + + :: + + 7 6 5 4 3 2 1 0 + +-+-+-+-+-+-+-+-+ + |Z|M|R| FRAGMENT| + +-+-+-+---------+ + % seq num % + +---------------+ + ~ [FragExts] ~ if Z==1 + +---------------+ + ~ [u8] ~ + +---------------+ + + The fragment payload runs to the end of the batch, so a Fragment is + always the last message of its batch. + """ + name = "ZenohFragment" + fields_desc = [ + BitField("flag_z", 0, 1), + BitEnumField("flag_m", 0, 1, {0: "Last", 1: "More"}), + BitEnumField("flag_r", 0, 1, {0: "BestEffort", 1: "Reliable"}), + BitEnumField("mid", ZENOH_MID_FRAGMENT, 5, ZENOH_TRANSPORT_MID), + ZenohZIntField("sn", 0), + _extensions_field(), + ] + + def extract_padding(self, s): + return s, b"" + + def guess_payload_class(self, payload): + return conf.raw_layer if payload else conf.padding_layer + + +class ZenohJoin(_ZenohMsg): + """Join message, advertising the transport parameters on a multicast link. + + :: + + 7 6 5 4 3 2 1 0 + +-+-+-+-+-+-+-+-+ + |Z|S|T| JOIN | + +-+-+-+---------+ + | version | + +---------------+ + |zid_len|x|x|wai| + +-------+-+-+---+ + ~ [u8] ~ -- Zenoh ID of the sender + +---------------+ + |x|x|x|x|rid|fsn| \\ -- SN/ID resolution + +---------------+ | if S==1 + | u16 | / -- Batch size + +---------------+ + % lease % + +---------------+ + % next_sn (x2) % -- Reliable and best effort next sequence numbers + +---------------+ + """ + name = "ZenohJoin" + fields_desc = [ + BitField("flag_z", 0, 1), + BitField("flag_s", 0, 1), + BitEnumField("flag_t", 0, 1, {0: "Milliseconds", 1: "Seconds"}), + BitEnumField("mid", ZENOH_MID_JOIN, 5, ZENOH_TRANSPORT_MID), + ByteField("version", ZENOH_VERSION), + BitFieldLenField("zid_len", None, 4, length_of="zid", + adjust=lambda pkt, x: max(x, 1) - 1), + BitField("res1", 0, 2), + BitEnumField("whatami", 0x01, 2, ZENOH_WHATAMI), + XStrLenField("zid", b"\x01", + length_from=lambda pkt: (pkt.zid_len or 0) + 1), + ] + _resolution_fields() + [ + ZenohZIntField("lease", 10000), + ZenohZIntField("next_sn_reliable", 0), + ZenohZIntField("next_sn_best_effort", 0), + _extensions_field(), + ] + + +class ZenohTransportOAM(_ZenohMsg): + """Transport level operation, administration and maintenance message. + + :: + + 7 6 5 4 3 2 1 0 + +-+-+-+-+-+-+-+-+ + |Z|ENC| OAM | + +-+-+-+---------+ + ~ id:z16 ~ + +---------------+ + ~ [OamExts] ~ if Z==1 + +---------------+ + % length % \\ if ENC == Z64 or ZBuf + ~ [u8] ~ / if ENC == ZBuf + +---------------+ + """ + name = "ZenohTransportOAM" + fields_desc = [ + BitField("flag_z", 0, 1), + BitEnumField("enc", 0, 2, ZENOH_EXT_ENCODING), + BitEnumField("mid", ZENOH_MID_T_OAM, 5, ZENOH_TRANSPORT_MID), + ZenohZIntField("oam_id", 0), + _extensions_field(), + ConditionalField(ZenohZIntField("value", 0), + lambda pkt: pkt.enc == 0x01), + ConditionalField(ZenohZBufField("body", b""), + lambda pkt: pkt.enc == 0x02), + ] + + +# ============================================================================ +# Network messages +# ============================================================================ + +def _wire_expr_fields(): + """Key expression: a scope ID plus an optional suffix, announced by N.""" + return [ + ZenohZIntField("key_scope", 0), + ConditionalField(ZenohZStrField("key_suffix", ""), + _flag_or_value("flag_n", "key_suffix")), + ] + + +# The N flag of the key expression sits in bit 5 of the header byte. +_AUTO_FLAG_N = [("flag_n", "key_suffix", 0, 0x20)] + + +class ZenohPush(_ZenohMsgWithBody): + """Push message, publishing data towards the subscribers of a key. + + :: + + 7 6 5 4 3 2 1 0 + +-+-+-+-+-+-+-+-+ + |Z|M|N| PUSH | + +-+-+-+---------+ + ~ key_scope:z16 ~ + +---------------+ + ~ key_suffix ~ if N==1 + +---------------+ + ~ [PushExts] ~ if Z==1 + +---------------+ + ~ PushBody ~ -- Put or Del + +---------------+ + """ + name = "ZenohPush" + fields_desc = [ + BitField("flag_z", 0, 1), + BitEnumField("flag_m", 0, 1, {0: "Receiver", 1: "Sender"}), + BitField("flag_n", 0, 1), + BitEnumField("mid", ZENOH_MID_PUSH, 5, ZENOH_NETWORK_MID), + ] + _wire_expr_fields() + [ + _extensions_field(), + ] + auto_flags = _AUTO_FLAG_N + + def guess_payload_class(self, payload): + return _guess_msg_class(payload, _PUSH_BODY_CLASSES) + + def mysummary(self): + return "ZenohPush %s" % _key_expr_repr(self) + + +class ZenohRequest(_ZenohMsgWithBody): + """Request message, sending a query to the queryables of a key. + + :: + + 7 6 5 4 3 2 1 0 + +-+-+-+-+-+-+-+-+ + |Z|M|N| REQUEST | + +-+-+-+---------+ + ~ request_id:z32~ + +---------------+ + ~ key_scope:z16 ~ + +---------------+ + ~ key_suffix ~ if N==1 + +---------------+ + ~ [ReqExts] ~ if Z==1 + +---------------+ + ~ RequestBody ~ -- Query + +---------------+ + """ + name = "ZenohRequest" + fields_desc = [ + BitField("flag_z", 0, 1), + BitEnumField("flag_m", 0, 1, {0: "Receiver", 1: "Sender"}), + BitField("flag_n", 0, 1), + BitEnumField("mid", ZENOH_MID_REQUEST, 5, ZENOH_NETWORK_MID), + ZenohZIntField("request_id", 0), + ] + _wire_expr_fields() + [ + _extensions_field(), + ] + + auto_flags = _AUTO_FLAG_N + + def guess_payload_class(self, payload): + return _guess_msg_class(payload, _REQUEST_BODY_CLASSES) + + def mysummary(self): + return "ZenohRequest id=%s %s" % (self.request_id, + _key_expr_repr(self)) + + +class ZenohResponse(_ZenohMsgWithBody): + """Response message, answering a Request. + + :: + + 7 6 5 4 3 2 1 0 + +-+-+-+-+-+-+-+-+ + |Z|M|N| RESPONSE| + +-+-+-+---------+ + ~ request_id:z32~ + +---------------+ + ~ key_scope:z16 ~ + +---------------+ + ~ key_suffix ~ if N==1 + +---------------+ + ~ [RespExts] ~ if Z==1 + +---------------+ + ~ ResponseBody ~ -- Reply or Err + +---------------+ + """ + name = "ZenohResponse" + fields_desc = [ + BitField("flag_z", 0, 1), + BitEnumField("flag_m", 0, 1, {0: "Receiver", 1: "Sender"}), + BitField("flag_n", 0, 1), + BitEnumField("mid", ZENOH_MID_RESPONSE, 5, ZENOH_NETWORK_MID), + ZenohZIntField("request_id", 0), + ] + _wire_expr_fields() + [ + _extensions_field(), + ] + + auto_flags = _AUTO_FLAG_N + + def guess_payload_class(self, payload): + return _guess_msg_class(payload, _RESPONSE_BODY_CLASSES) + + def mysummary(self): + return "ZenohResponse id=%s %s" % (self.request_id, + _key_expr_repr(self)) + + +class ZenohResponseFinal(_ZenohMsg): + """ResponseFinal message, closing the response stream of a Request. + + :: + + 7 6 5 4 3 2 1 0 + +-+-+-+-+-+-+-+-+ + |Z|X|X| ResFinal| + +-+-+-+---------+ + ~ request_id:z32~ + +---------------+ + """ + name = "ZenohResponseFinal" + fields_desc = [ + BitField("flag_z", 0, 1), + BitField("res1", 0, 1), + BitField("res2", 0, 1), + BitEnumField("mid", ZENOH_MID_RESPONSE_FINAL, 5, ZENOH_NETWORK_MID), + ZenohZIntField("request_id", 0), + _extensions_field(), + ] + + +class ZenohInterest(_ZenohMsg): + """Interest message, requesting the transmission of declarations. + + :: + + 7 6 5 4 3 2 1 0 + +-+-+-+-+-+-+-+-+ + |Z|Mod|INTEREST | + +-+-+-+---------+ + ~ id:z32 ~ + +---------------+ + |A|M|N|R|T|Q|S|K| if Mod!=Final + +---------------+ + ~ key_scope:z16 ~ if Mod!=Final and R==1 + +---------------+ + ~ key_suffix ~ if Mod!=Final and R==1 and N==1 + +---------------+ + """ + name = "ZenohInterest" + fields_desc = [ + BitField("flag_z", 0, 1), + BitEnumField("mode", 0x01, 2, ZENOH_INTEREST_MODE), + BitEnumField("mid", ZENOH_MID_INTEREST, 5, ZENOH_NETWORK_MID), + ZenohZIntField("interest_id", 0), + ConditionalField(FlagsField("options", 0, 8, ZENOH_INTEREST_OPTIONS), + lambda pkt: pkt.mode != 0x00), + ConditionalField(ZenohZIntField("key_scope", 0), + lambda pkt: _interest_option(pkt, 0x10)), + ConditionalField(ZenohZStrField("key_suffix", ""), + lambda pkt: _interest_option(pkt, 0x10) and + _interest_option(pkt, 0x20)), + _extensions_field(), + ] + + +class ZenohDeclare(_ZenohMsgWithBody): + """Declare message, carrying a single declaration. + + :: + + 7 6 5 4 3 2 1 0 + +-+-+-+-+-+-+-+-+ + |Z|X|I| DECLARE | + +-+-+-+---------+ + ~interest_id:z32~ if I==1 + +---------------+ + ~ [DeclExts] ~ if Z==1 + +---------------+ + ~ declaration ~ + +---------------+ + """ + name = "ZenohDeclare" + fields_desc = [ + BitField("flag_z", 0, 1), + BitField("res1", 0, 1), + BitField("flag_i", 0, 1), + BitEnumField("mid", ZENOH_MID_DECLARE, 5, ZENOH_NETWORK_MID), + ConditionalField(ZenohZIntField("interest_id", 0), + _flag_or_value("flag_i", "interest_id")), + _extensions_field(), + ] + auto_flags = [("flag_i", "interest_id", 0, 0x20)] + + def guess_payload_class(self, payload): + return _guess_msg_class(payload, _DECLARATION_CLASSES) + + +class ZenohNetworkOAM(_ZenohMsg): + """Network level operation, administration and maintenance message. + + :: + + 7 6 5 4 3 2 1 0 + +-+-+-+-+-+-+-+-+ + |Z|ENC| OAM | + +-+-+-+---------+ + ~ id:z16 ~ + +---------------+ + ~ [OamExts] ~ if Z==1 + +---------------+ + % length % \\ if ENC == Z64 or ZBuf + ~ [u8] ~ / if ENC == ZBuf + +---------------+ + """ + name = "ZenohNetworkOAM" + fields_desc = [ + BitField("flag_z", 0, 1), + BitEnumField("enc", 0, 2, ZENOH_EXT_ENCODING), + BitEnumField("mid", ZENOH_MID_N_OAM, 5, ZENOH_NETWORK_MID), + ZenohZIntField("oam_id", 0), + _extensions_field(), + ConditionalField(ZenohZIntField("value", 0), + lambda pkt: pkt.enc == 0x01), + ConditionalField(ZenohZBufField("body", b""), + lambda pkt: pkt.enc == 0x02), + ] + + +# ============================================================================ +# Declarations, carried by a Declare message +# ============================================================================ + +class ZenohDeclareKeyExpr(_ZenohMsg): + """Bind a numerical expression ID to a key expression. + + :: + + 7 6 5 4 3 2 1 0 + +-+-+-+-+-+-+-+-+ + |Z|X|N| D_KEXPR | + +---------------+ + ~ expr_id:z16 ~ + +---------------+ + ~ key_scope:z16 ~ + +---------------+ + ~ key_suffix ~ if N==1 + +---------------+ + """ + name = "ZenohDeclareKeyExpr" + fields_desc = [ + BitField("flag_z", 0, 1), + BitField("res1", 0, 1), + BitField("flag_n", 0, 1), + BitEnumField("did", ZENOH_DECL_KEYEXPR, 5, ZENOH_DECLARATION_ID), + ZenohZIntField("expr_id", 0), + ] + _wire_expr_fields() + [ + _extensions_field(), + ] + auto_flags = _AUTO_FLAG_N + + +class ZenohUndeclareKeyExpr(_ZenohMsg): + """Release an expression ID previously bound by a DeclareKeyExpr.""" + name = "ZenohUndeclareKeyExpr" + fields_desc = [ + BitField("flag_z", 0, 1), + BitField("res1", 0, 1), + BitField("res2", 0, 1), + BitEnumField("did", ZENOH_DECL_U_KEYEXPR, 5, ZENOH_DECLARATION_ID), + ZenohZIntField("expr_id", 0), + _extensions_field(), + ] + + +class _ZenohDeclareEntity(_ZenohMsg): + """Common layout of the subscriber, queryable and token declarations.""" + auto_flags = _AUTO_FLAG_N + + +class ZenohDeclareSubscriber(_ZenohDeclareEntity): + """Announce a subscriber on a key expression. + + :: + + 7 6 5 4 3 2 1 0 + +-+-+-+-+-+-+-+-+ + |Z|M|N| D_SUB | + +---------------+ + ~ subs_id:z32 ~ + +---------------+ + ~ key_scope:z16 ~ + +---------------+ + ~ key_suffix ~ if N==1 + +---------------+ + """ + name = "ZenohDeclareSubscriber" + fields_desc = [ + BitField("flag_z", 0, 1), + BitEnumField("flag_m", 0, 1, {0: "Receiver", 1: "Sender"}), + BitField("flag_n", 0, 1), + BitEnumField("did", ZENOH_DECL_SUBSCRIBER, 5, ZENOH_DECLARATION_ID), + ZenohZIntField("subscriber_id", 0), + ] + _wire_expr_fields() + [ + _extensions_field(), + ] + + +class ZenohUndeclareSubscriber(_ZenohMsg): + """Withdraw a subscriber. The key expression travels in extension 0x0f.""" + name = "ZenohUndeclareSubscriber" + fields_desc = [ + BitField("flag_z", 0, 1), + BitField("res1", 0, 1), + BitField("res2", 0, 1), + BitEnumField("did", ZENOH_DECL_U_SUBSCRIBER, 5, ZENOH_DECLARATION_ID), + ZenohZIntField("subscriber_id", 0), + _extensions_field(), + ] + + +class ZenohDeclareQueryable(_ZenohDeclareEntity): + """Announce a queryable on a key expression.""" + name = "ZenohDeclareQueryable" + fields_desc = [ + BitField("flag_z", 0, 1), + BitEnumField("flag_m", 0, 1, {0: "Receiver", 1: "Sender"}), + BitField("flag_n", 0, 1), + BitEnumField("did", ZENOH_DECL_QUERYABLE, 5, ZENOH_DECLARATION_ID), + ZenohZIntField("queryable_id", 0), + ] + _wire_expr_fields() + [ + _extensions_field(), + ] + + +class ZenohUndeclareQueryable(_ZenohMsg): + """Withdraw a queryable. The key expression travels in extension 0x0f.""" + name = "ZenohUndeclareQueryable" + fields_desc = [ + BitField("flag_z", 0, 1), + BitField("res1", 0, 1), + BitField("res2", 0, 1), + BitEnumField("did", ZENOH_DECL_U_QUERYABLE, 5, ZENOH_DECLARATION_ID), + ZenohZIntField("queryable_id", 0), + _extensions_field(), + ] + + +class ZenohDeclareToken(_ZenohDeclareEntity): + """Announce a liveliness token on a key expression.""" + name = "ZenohDeclareToken" + fields_desc = [ + BitField("flag_z", 0, 1), + BitEnumField("flag_m", 0, 1, {0: "Receiver", 1: "Sender"}), + BitField("flag_n", 0, 1), + BitEnumField("did", ZENOH_DECL_TOKEN, 5, ZENOH_DECLARATION_ID), + ZenohZIntField("token_id", 0), + ] + _wire_expr_fields() + [ + _extensions_field(), + ] + + +class ZenohUndeclareToken(_ZenohMsg): + """Withdraw a token. The key expression travels in extension 0x0f.""" + name = "ZenohUndeclareToken" + fields_desc = [ + BitField("flag_z", 0, 1), + BitField("res1", 0, 1), + BitField("res2", 0, 1), + BitEnumField("did", ZENOH_DECL_U_TOKEN, 5, ZENOH_DECLARATION_ID), + ZenohZIntField("token_id", 0), + _extensions_field(), + ] + + +class ZenohDeclareFinal(_ZenohMsg): + """Mark the end of the declarations sent in response to an Interest.""" + name = "ZenohDeclareFinal" + fields_desc = [ + BitField("flag_z", 0, 1), + BitField("res1", 0, 1), + BitField("res2", 0, 1), + BitEnumField("did", ZENOH_DECL_FINAL, 5, ZENOH_DECLARATION_ID), + _extensions_field(), + ] + + +# ============================================================================ +# Zenoh messages, the user facing payload of the network messages +# ============================================================================ + +class ZenohPut(_ZenohMsg): + """Put message, holding the published payload. + + :: + + 7 6 5 4 3 2 1 0 + +-+-+-+-+-+-+-+-+ + |Z|E|T| PUT | + +-+-+-+---------+ + ~ ts: ~ if T==1 + +---------------+ + ~ encoding ~ if E==1 + +---------------+ + ~ [PutExts] ~ if Z==1 + +---------------+ + ~ pl: ~ -- Payload + +---------------+ + + The payload is exposed as ``data``, since ``payload`` is reserved by + Scapy for the next layer. + """ + name = "ZenohPut" + fields_desc = [ + BitField("flag_z", 0, 1), + BitField("flag_e", 0, 1), + BitField("flag_t", 0, 1), + BitEnumField("mid", ZENOH_MID_PUT, 5, ZENOH_ZENOH_MID), + ConditionalField(ZenohTimestampField("timestamp", None), + _flag_or_value("flag_t", "timestamp")), + ConditionalField(ZenohEncodingField("encoding", 0), + _flag_or_value("flag_e", "encoding")), + _extensions_field(), + ZenohZBufField("data", b""), + ] + auto_flags = [("flag_t", "timestamp", 0, 0x20), + ("flag_e", "encoding", 0, 0x40)] + + def mysummary(self): + # The key expression of the enclosing message tells what was written, + # so ask for its summary to be kept. + return "ZenohPut %d bytes" % len(self.data or b""), _summary_parents() + + +class ZenohDel(_ZenohMsg): + """Del message, deleting the data associated with a key expression. + + :: + + 7 6 5 4 3 2 1 0 + +-+-+-+-+-+-+-+-+ + |Z|X|T| DEL | + +-+-+-+---------+ + ~ ts: ~ if T==1 + +---------------+ + """ + name = "ZenohDel" + fields_desc = [ + BitField("flag_z", 0, 1), + BitField("res1", 0, 1), + BitField("flag_t", 0, 1), + BitEnumField("mid", ZENOH_MID_DEL, 5, ZENOH_ZENOH_MID), + ConditionalField(ZenohTimestampField("timestamp", None), + _flag_or_value("flag_t", "timestamp")), + _extensions_field(), + ] + auto_flags = [("flag_t", "timestamp", 0, 0x20)] + + def mysummary(self): + return self.name, _summary_parents() + + +class ZenohQuery(_ZenohMsg): + """Query message, the body of a Request. + + :: + + 7 6 5 4 3 2 1 0 + +-+-+-+-+-+-+-+-+ + |Z|P|C| QUERY | + +-+-+-+---------+ + % consolidation % if C==1 + +---------------+ + ~ ps: ~ if P==1 -- Selector parameters + +---------------+ + """ + name = "ZenohQuery" + fields_desc = [ + BitField("flag_z", 0, 1), + BitField("flag_p", 0, 1), + BitField("flag_c", 0, 1), + BitEnumField("mid", ZENOH_MID_QUERY, 5, ZENOH_ZENOH_MID), + ConditionalField(ZenohConsolidationField("consolidation", 0), + _flag_or_value("flag_c", "consolidation")), + ConditionalField(ZenohZStrField("parameters", ""), + _flag_or_value("flag_p", "parameters")), + _extensions_field(), + ] + auto_flags = [("flag_c", "consolidation", 0, 0x20), + ("flag_p", "parameters", 0, 0x40)] + + def mysummary(self): + if self.parameters: + return "ZenohQuery ?%s" % self.parameters, _summary_parents() + return self.name, _summary_parents() + + +class ZenohReply(_ZenohMsgWithBody): + """Reply message, the body of a successful Response. + + :: + + 7 6 5 4 3 2 1 0 + +-+-+-+-+-+-+-+-+ + |Z|X|C| REPLY | + +-+-+-+---------+ + % consolidation % if C==1 + +---------------+ + ~ [ReplyExts] ~ if Z==1 + +---------------+ + ~ ReplyBody ~ -- Put or Del + +---------------+ + """ + name = "ZenohReply" + fields_desc = [ + BitField("flag_z", 0, 1), + BitField("res1", 0, 1), + BitField("flag_c", 0, 1), + BitEnumField("mid", ZENOH_MID_REPLY, 5, ZENOH_ZENOH_MID), + ConditionalField(ZenohConsolidationField("consolidation", 0), + _flag_or_value("flag_c", "consolidation")), + _extensions_field(), + ] + auto_flags = [("flag_c", "consolidation", 0, 0x20)] + + def guess_payload_class(self, payload): + return _guess_msg_class(payload, _REPLY_BODY_CLASSES) + + +class ZenohErr(_ZenohMsg): + """Err message, the body of a failed Response. + + :: + + 7 6 5 4 3 2 1 0 + +-+-+-+-+-+-+-+-+ + |Z|E|X| ERR | + +-+-+-+---------+ + ~ encoding ~ if E==1 + +---------------+ + ~ [ErrExts] ~ if Z==1 + +---------------+ + ~ pl: ~ -- Payload + +---------------+ + """ + name = "ZenohErr" + fields_desc = [ + BitField("flag_z", 0, 1), + BitField("flag_e", 0, 1), + BitField("res1", 0, 1), + BitEnumField("mid", ZENOH_MID_ERR, 5, ZENOH_ZENOH_MID), + ConditionalField(ZenohEncodingField("encoding", 0), + _flag_or_value("flag_e", "encoding")), + _extensions_field(), + ZenohZBufField("data", b""), + ] + auto_flags = [("flag_e", "encoding", 0, 0x40)] + + def mysummary(self): + return "ZenohErr %d bytes" % len(self.data or b""), _summary_parents() + + +# ============================================================================ +# Batches +# ============================================================================ + +class ZenohBatch(Packet): + """A batch of zenoh transport messages, as carried by one datagram. + + Datagram links keep the message boundaries themselves, so no length + prefix is present. Stream links use :class:`ZenohStreamBatch` instead. + """ + name = "ZenohBatch" + fields_desc = [ + PacketListField("messages", [], + next_cls_cb=lambda pkt, lst, cur, remain: + _next_transport_msg(remain)), + ] + + def mysummary(self): + return "ZenohBatch %s" % " / ".join(m.name for m in self.messages) + + +class ZenohStreamBatch(Packet): + """A batch of zenoh transport messages on a stream link, e.g. TCP. + + Streams do not preserve message boundaries, so every batch is prefixed + with its total length as a 16-bit little endian integer. A single TCP + segment may hold several batches. + """ + name = "ZenohStreamBatch" + fields_desc = [ + FieldLenField("len", None, fmt=" 2: + length = orb(payload[0]) | (orb(payload[1]) << 8) + if 0 < length <= len(payload) - 2: + return ZenohStreamBatch + return Packet.guess_payload_class(self, payload) + + def mysummary(self): + return "ZenohStreamBatch %s" % " / ".join(m.name + for m in self.messages) + + +# ============================================================================ +# Dispatch tables +# ============================================================================ + +_SCOUTING_MSG_CLASSES = { + ZENOH_MID_SCOUT: ZenohScout, + ZENOH_MID_HELLO: ZenohHello, +} + +_TRANSPORT_MSG_CLASSES = { + ZENOH_MID_T_OAM: ZenohTransportOAM, + ZENOH_MID_INIT: ZenohInit, + ZENOH_MID_OPEN: ZenohOpen, + ZENOH_MID_CLOSE: ZenohClose, + ZENOH_MID_KEEPALIVE: ZenohKeepAlive, + ZENOH_MID_FRAME: ZenohFrame, + ZENOH_MID_FRAGMENT: ZenohFragment, + ZENOH_MID_JOIN: ZenohJoin, +} + +_NETWORK_MSG_CLASSES = { + ZENOH_MID_INTEREST: ZenohInterest, + ZENOH_MID_RESPONSE_FINAL: ZenohResponseFinal, + ZENOH_MID_RESPONSE: ZenohResponse, + ZENOH_MID_REQUEST: ZenohRequest, + ZENOH_MID_PUSH: ZenohPush, + ZENOH_MID_DECLARE: ZenohDeclare, + ZENOH_MID_N_OAM: ZenohNetworkOAM, +} + +_PUSH_BODY_CLASSES = { + ZENOH_MID_PUT: ZenohPut, + ZENOH_MID_DEL: ZenohDel, +} + +_REQUEST_BODY_CLASSES = { + ZENOH_MID_QUERY: ZenohQuery, +} + +_RESPONSE_BODY_CLASSES = { + ZENOH_MID_REPLY: ZenohReply, + ZENOH_MID_ERR: ZenohErr, +} + +_REPLY_BODY_CLASSES = dict(_PUSH_BODY_CLASSES) + +_DECLARATION_CLASSES = { + ZENOH_DECL_KEYEXPR: ZenohDeclareKeyExpr, + ZENOH_DECL_U_KEYEXPR: ZenohUndeclareKeyExpr, + ZENOH_DECL_SUBSCRIBER: ZenohDeclareSubscriber, + ZENOH_DECL_U_SUBSCRIBER: ZenohUndeclareSubscriber, + ZENOH_DECL_QUERYABLE: ZenohDeclareQueryable, + ZENOH_DECL_U_QUERYABLE: ZenohUndeclareQueryable, + ZENOH_DECL_TOKEN: ZenohDeclareToken, + ZENOH_DECL_U_TOKEN: ZenohUndeclareToken, + ZENOH_DECL_FINAL: ZenohDeclareFinal, +} + + +def _next_transport_msg(remain): + """Class of the next transport message of a batch, if any.""" + if not remain: + return None + return _TRANSPORT_MSG_CLASSES.get(orb(remain[0]) & 0x1F, conf.raw_layer) + + +def _next_network_msg(remain): + """Class of the next network message of a frame, if any. + + Returning None ends the frame, which lets the enclosing batch look for a + further transport message. + """ + if not remain: + return None + return _NETWORK_MSG_CLASSES.get(orb(remain[0]) & 0x1F) + + +def _key_expr_repr(pkt): + suffix = pkt.key_suffix + if isinstance(suffix, bytes): + suffix = suffix.decode("utf-8", "replace") + if not suffix: + return str(pkt.key_scope or 0) + return "%d/%s" % (pkt.key_scope or 0, suffix) + + +def _summary_parents(): + """Messages whose summary is worth keeping in front of a body message.""" + return [ZenohPush, ZenohRequest, ZenohResponse, ZenohReply] + + +# ============================================================================ +# Layer bindings +# ============================================================================ + +# Only the destination port is bound top down, so that building a packet does +# not set the source port as well. +bind_layers(UDP, ZenohScouting, dport=ZENOH_PORT_SCOUTING) +bind_bottom_up(UDP, ZenohScouting, sport=ZENOH_PORT_SCOUTING) + +bind_layers(UDP, ZenohBatch, dport=ZENOH_PORT_TRANSPORT) +bind_bottom_up(UDP, ZenohBatch, sport=ZENOH_PORT_TRANSPORT) + +bind_layers(TCP, ZenohStreamBatch, dport=ZENOH_PORT_TRANSPORT) +bind_bottom_up(TCP, ZenohStreamBatch, sport=ZENOH_PORT_TRANSPORT) diff --git a/test/contrib/zenoh.uts b/test/contrib/zenoh.uts new file mode 100644 index 00000000000..308fef2dbb6 --- /dev/null +++ b/test/contrib/zenoh.uts @@ -0,0 +1,849 @@ +% Zenoh Protocol tests + +# Type the following command to launch the tests: +# $ test/run_tests -P "load_contrib('zenoh')" -t test/contrib/zenoh.uts + ++ Syntax check + += Import the Zenoh layer +from scapy.contrib.zenoh import * +from scapy.compat import raw +from scapy.config import conf +from scapy.layers.inet import IP, TCP, UDP +from scapy.packet import Raw + + +############ +############ ++ Variable length integers (zint) + += zint: single byte values +assert zenoh_zint_encode(0) == b'\x00' +assert zenoh_zint_encode(1) == b'\x01' +assert zenoh_zint_encode(127) == b'\x7f' + += zint: multi byte values, 7 bits per byte, least significant first +assert zenoh_zint_encode(128) == b'\x80\x01' +assert zenoh_zint_encode(300) == b'\xac\x02' +assert zenoh_zint_encode(16383) == b'\xff\x7f' +assert zenoh_zint_encode(16384) == b'\x80\x80\x01' + += zint: the ninth byte carries eight payload bits +assert zenoh_zint_encode(2 ** 64 - 1) == b'\xff' * 9 +assert len(zenoh_zint_encode(2 ** 64 - 1)) == ZENOH_ZINT_MAX_LEN + += zint: values outside the u64 range are rejected +for bad in [-1, 2 ** 64]: + try: + zenoh_zint_encode(bad) + raise AssertionError("invalid zint accepted: %d" % bad) + except ValueError: + pass + += zint: decode returns the value and the number of bytes consumed +assert zenoh_zint_decode(b'\x00') == (0, 1) +assert zenoh_zint_decode(b'\x7f') == (127, 1) +assert zenoh_zint_decode(b'\x80\x01') == (128, 2) +assert zenoh_zint_decode(b'\xac\x02') == (300, 2) +assert zenoh_zint_decode(b'\xff' * 9) == (2 ** 64 - 1, 9) + += zint: trailing bytes are left untouched +assert zenoh_zint_decode(b'\x2a\xff\xff') == (42, 1) + += zint: a truncated integer reports zero consumed bytes +assert zenoh_zint_decode(b'\x80\x80\x80') == (0, 0) +assert zenoh_zint_decode(b'') == (0, 0) + += zint: roundtrip +for val in [0, 1, 42, 127, 128, 255, 300, 16383, 16384, 2 ** 32 - 1, 2 ** 64 - 1]: + data = zenoh_zint_encode(val) + assert zenoh_zint_decode(data) == (val, len(data)) + + +############ +############ ++ Zenoh field types + += ZenohZIntField: build and dissect +f = ZenohZIntField('test', 0) +assert f.addfield(None, b'', 300) == b'\xac\x02' +assert f.getfield(None, b'\xac\x02\xff') == (b'\xff', 300) +assert f.i2len(None, 300) == 2 + += ZenohZIntField: a truncated integer consumes the buffer +f = ZenohZIntField('test', 0) +assert f.getfield(None, b'\x80\x80') == (b'', 0) + += ZenohZBufField: length prefixed buffer +f = ZenohZBufField('cookie', b'') +assert f.addfield(None, b'', b'secret') == b'\x06secret' +assert f.addfield(None, b'', None) == b'\x00' +assert f.addfield(None, b'', 'hi') == b'\x02hi' +assert f.getfield(None, b'\x06secret!') == (b'!', b'secret') +assert f.i2len(None, b'secret') == 7 + += ZenohZBufField: a truncated length consumes the buffer +f = ZenohZBufField('cookie', b'') +assert f.getfield(None, b'\x80\x80') == (b'', b'') + += ZenohZBufField: repr is hexadecimal +f = ZenohZBufField('cookie', b'') +assert f.i2repr(None, b'\xab\xcd') == 'abcd' + += ZenohZStrField: utf-8 strings +f = ZenohZStrField('locator', '') +assert f.addfield(None, b'', 'tcp/127.0.0.1:7447') == b'\x12tcp/127.0.0.1:7447' +assert f.getfield(None, b'\x03a/b') == (b'', 'a/b') + += ZenohZStrField: invalid utf-8 stays bytes so that the packet rebuilds +f = ZenohZStrField('locator', '') +assert f.getfield(None, b'\x02\xff\xfe') == (b'', b'\xff\xfe') +assert f.addfield(None, b'', b'\xff\xfe') == b'\x02\xff\xfe' + += ZenohEncodingField: an encoding without schema is (id << 1) +f = ZenohEncodingField('encoding', 0) +assert f.addfield(None, b'', 0) == b'\x00' +assert f.addfield(None, b'', 4) == b'\x08' +assert f.getfield(None, b'\x08') == (b'', 4) + += ZenohEncodingField: the S flag announces a schema +f = ZenohEncodingField('encoding', 0) +assert f.addfield(None, b'', (5, 'utf8')) == b'\x0b\x04utf8' +assert f.getfield(None, b'\x0b\x04utf8') == (b'', (5, b'utf8')) + += ZenohEncodingField: repr uses the well known encoding names +f = ZenohEncodingField('encoding', 0) +assert f.i2repr(None, 0) == 'zenoh/bytes' +assert f.i2repr(None, 4) == 'text/plain' +assert f.i2repr(None, (5, b'utf8')) == 'application/json;utf8' +assert f.i2repr(None, 999) == '999' + += ZenohTimestampField: a zint counter followed by the source ID +f = ZenohTimestampField('timestamp', None) +assert f.addfield(None, b'', (7, b'\x01\x02')) == b'\x07\x02\x01\x02' +assert f.getfield(None, b'\x07\x02\x01\x02') == (b'', (7, b'\x01\x02')) +assert f.addfield(None, b'', None) == b'\x00\x00' + += ZenohTimestampField: repr splits the NTP64 counter +f = ZenohTimestampField('timestamp', None) +assert f.i2repr(None, (2 ** 32, b'\xaa')) == '1.000000000/aa' + += ZenohConsolidationField: repr uses the consolidation names +f = ZenohConsolidationField('consolidation', 0) +assert f.i2repr(None, 0) == 'Auto' +assert f.i2repr(None, 3) == 'Latest' +assert ZenohQuery(consolidation=2).sprintf('%ZenohQuery.consolidation%') == 'Monotonic' + + +############ +############ ++ Extensions + += ZenohExtension: unit extension has no body +ext = ZenohExtension(eid=1, enc='Unit') +assert raw(ext) == b'\x01' + += ZenohExtension: Z64 extension carries a zint +ext = ZenohExtension(eid=1, enc='Z64', mandatory=1, value=300) +assert raw(ext) == b'\x31\xac\x02' +assert ZenohExtension(b'\x31\xac\x02').value == 300 + += ZenohExtension: ZBuf extension carries a length prefixed buffer +ext = ZenohExtension(eid=2, enc='ZBuf', body=b'\xde\xad') +assert raw(ext) == b'\x42\x02\xde\xad' +assert ZenohExtension(b'\x42\x02\xde\xad').body == b'\xde\xad' + += Extensions: the Z flag of the message and the more bit are computed +pkt = ZenohKeepAlive(extensions=[ZenohExtension(eid=1, enc='Z64', value=1), ZenohExtension(eid=2)]) +data = raw(pkt) +assert data == b'\x84\xa1\x01\x02' +parsed = ZenohKeepAlive(data) +assert parsed.flag_z == 1 +assert len(parsed.extensions) == 2 +assert parsed.extensions[0].more == 1 +assert parsed.extensions[1].more == 0 +assert raw(parsed) == data + += Extensions: no extension leaves the Z flag clear +assert raw(ZenohKeepAlive()) == b'\x04' +assert ZenohKeepAlive(b'\x04').extensions == [] + += Extensions: an explicit Z flag is not overwritten +assert raw(ZenohKeepAlive(flag_z=1)) == b'\x84' + + +############ +############ ++ Scouting messages + += ZenohScout: default scout asks every kind of node +scout = ZenohScout() +assert raw(scout) == b'\x01\x09\x07' +assert scout.what == 0x07 + += ZenohScout: the I flag and zid_len follow the Zenoh ID +scout = ZenohScout(zid=b'\x01\x02\x03', what='Router') +data = raw(scout) +assert data == b'\x01\x09\x29\x01\x02\x03' +parsed = ZenohScout(data) +assert parsed.flag_i == 1 +assert parsed.zid_len == 2 +assert parsed.zid == b'\x01\x02\x03' +assert parsed.what == 1 +assert raw(parsed) == data + += ZenohScout: the what field is a bitmap of node kinds +assert ZenohScout(what='Router+Client').what == 0x05 +assert 'Peer' in ZenohScout(b'\x01\x09\x02').what + += ZenohHello: version, whatami and Zenoh ID +hello = ZenohHello(zid=b'\xaa\xbb', whatami='Router') +data = raw(hello) +assert data == b'\x02\x09\x10\xaa\xbb' +parsed = ZenohHello(data) +assert parsed.whatami == 0 +assert parsed.zid == b'\xaa\xbb' +assert parsed.flag_l == 0 +assert not parsed.locators + += ZenohHello: the L flag announces the locator list +hello = ZenohHello(zid=b'\x01', locators=[ZenohLocator(locator='tcp/1.2.3.4:7447')]) +data = raw(hello) +assert data == b'\x22\x09\x01\x01\x01\x10tcp/1.2.3.4:7447' +parsed = ZenohHello(data) +assert parsed.flag_l == 1 +assert parsed.num_locators == 1 +assert parsed.locators[0].locator == 'tcp/1.2.3.4:7447' +assert raw(parsed) == data + += ZenohHello: several locators +hello = ZenohHello(zid=b'\x01', locators=[ZenohLocator(locator='tcp/a:1'), ZenohLocator(locator='udp/b:2')]) +parsed = ZenohHello(raw(hello)) +assert parsed.num_locators == 2 +assert [x.locator for x in parsed.locators] == ['tcp/a:1', 'udp/b:2'] + += Scouting: UDP port 7446 dispatches on the message ID +pkt = IP()/UDP(sport=45678, dport=ZENOH_PORT_SCOUTING)/ZenohScout() +parsed = IP(raw(pkt)) +assert ZenohScout in parsed +assert not parsed.haslayer(Raw) + += Scouting: a Hello answer is dispatched as well +pkt = IP()/UDP(sport=ZENOH_PORT_SCOUTING, dport=45678)/ZenohHello(zid=b'\x01') +parsed = IP(raw(pkt)) +assert ZenohHello in parsed + += Scouting: an unknown scouting message ID falls back to Raw +pkt = ZenohScouting(b'\x1f\x09') +assert isinstance(pkt, Raw) + + +############ +############ ++ Transport messages: session establishment + += ZenohInit: InitSyn +init = ZenohInit(whatami='Router', zid=b'\x01\x02') +data = raw(init) +assert data == b'\x01\x09\x10\x01\x02' +parsed = ZenohInit(data) +assert parsed.flag_a == 0 +assert parsed.flag_s == 0 +assert parsed.batch_size is None + += ZenohInit: the S flag announces the resolutions and the batch size +init = ZenohInit(flag_s=1, zid=b'\x01', batch_size=65535) +data = raw(init) +assert data == b'\x41\x09\x01\x01\x0a\xff\xff' +parsed = ZenohInit(data) +assert parsed.fsn_resolution == 0x02 +assert parsed.rid_resolution == 0x02 +assert parsed.batch_size == 65535 + += ZenohInit: only the InitAck carries a cookie +init = ZenohInit(flag_a=1, zid=b'\x01', cookie=b'\x11\x22') +data = raw(init) +assert data == b'\x21\x09\x01\x01\x02\x11\x22' +assert ZenohInit(data).cookie == b'\x11\x22' +assert ZenohInit(raw(ZenohInit(zid=b'\x01'))).cookie is None + += ZenohInit: zid_len holds the ID length minus one +init = ZenohInit(zid=bytes(range(16))) +parsed = ZenohInit(raw(init)) +assert parsed.zid_len == 15 +assert parsed.zid == bytes(range(16)) + += ZenohOpen: the OpenSyn echoes the cookie of the InitAck +op = ZenohOpen(lease=10, initial_sn=7, cookie=b'\xaa') +data = raw(op) +assert data == b'\x02\x0a\x07\x01\xaa' +parsed = ZenohOpen(data) +assert parsed.lease == 10 +assert parsed.initial_sn == 7 +assert parsed.cookie == b'\xaa' + += ZenohOpen: the OpenAck has no cookie +op = ZenohOpen(flag_a=1, flag_t=1, lease=10, initial_sn=1) +data = raw(op) +assert data == b'\x62\x0a\x01' +assert ZenohOpen(data).cookie is None + += ZenohClose: reason codes are named +close = ZenohClose(flag_s=1, reason=3) +data = raw(close) +assert data == b'\x23\x03' +parsed = ZenohClose(data) +assert parsed.reason == 3 +assert parsed.sprintf('%ZenohClose.reason%') == 'MaxSessions' + += ZenohKeepAlive: header only +assert raw(ZenohKeepAlive()) == b'\x04' + + +############ +############ ++ Transport messages: data transfer + += ZenohFrame: sequence number and reliability +frame = ZenohFrame(flag_r=1, sn=42) +assert raw(frame) == b'\x25\x2a' +assert ZenohFrame(b'\x25\x2a').sn == 42 +assert ZenohFrame(b'\x05\x2a').flag_r == 0 + += ZenohFrame: carries a list of network messages +frame = ZenohFrame(sn=1, messages=[ZenohPush(key_scope=1)/ZenohPut(data=b'a'), ZenohPush(key_scope=2)/ZenohDel()]) +data = raw(frame) +parsed = ZenohFrame(data) +assert len(parsed.messages) == 2 +assert parsed.messages[0][ZenohPut].data == b'a' +assert ZenohDel in parsed.messages[1] +assert raw(parsed) == data + += ZenohFragment: the payload runs to the end of the batch +frag = ZenohFragment(flag_m=1, flag_r=1, sn=2)/Raw(b'\x01\x02\x03') +data = raw(frag) +assert data == b'\x66\x02\x01\x02\x03' +parsed = ZenohFragment(data) +assert parsed.flag_m == 1 +assert bytes(parsed.payload) == b'\x01\x02\x03' + += ZenohJoin: multicast transport parameters +join = ZenohJoin(flag_s=1, flag_t=1, whatami='Router', zid=b'\x01', batch_size=8192, lease=10, next_sn_reliable=1, next_sn_best_effort=2) +data = raw(join) +assert data == b'\x67\x09\x00\x01\x0a\x00\x20\x0a\x01\x02' +parsed = ZenohJoin(data) +assert parsed.batch_size == 8192 +assert parsed.next_sn_reliable == 1 +assert parsed.next_sn_best_effort == 2 +assert raw(parsed) == data + += ZenohTransportOAM: the ENC bits select the body +oam = ZenohTransportOAM(enc='Z64', oam_id=3, value=7) +data = raw(oam) +assert data == b'\x20\x03\x07' +assert ZenohTransportOAM(data).value == 7 +oam = ZenohTransportOAM(enc='ZBuf', oam_id=3, body=b'\xff') +assert raw(oam) == b'\x40\x03\x01\xff' + + +############ +############ ++ Network messages + += ZenohPush: the N flag announces the key expression suffix +push = ZenohPush(key_scope=1) +assert raw(push) == b'\x1d\x01' +push = ZenohPush(key_scope=1, key_suffix='demo/example') +data = raw(push) +assert data == b'\x3d\x01\x0cdemo/example' +parsed = ZenohPush(data) +assert parsed.flag_n == 1 +assert parsed.key_suffix == 'demo/example' + += ZenohPush: carries a Put or a Del +parsed = ZenohPush(raw(ZenohPush(key_scope=1)/ZenohPut(data=b'42'))) +assert ZenohPut in parsed +assert parsed[ZenohPut].data == b'42' +assert ZenohDel in ZenohPush(raw(ZenohPush(key_scope=1)/ZenohDel())) + += ZenohRequest: request ID precedes the key expression +req = ZenohRequest(request_id=5, key_scope=1, key_suffix='a/b') +data = raw(req) +assert data == b'\x3c\x05\x01\x03a/b' +parsed = ZenohRequest(data) +assert parsed.request_id == 5 +assert parsed.key_suffix == 'a/b' + += ZenohRequest: carries a Query +parsed = ZenohRequest(raw(ZenohRequest(request_id=1, key_scope=1)/ZenohQuery(parameters='k=v'))) +assert ZenohQuery in parsed +assert parsed[ZenohQuery].parameters == 'k=v' + += ZenohResponse: carries a Reply or an Err +parsed = ZenohResponse(raw(ZenohResponse(request_id=1, key_scope=1)/ZenohReply()/ZenohPut(data=b'x'))) +assert ZenohReply in parsed +assert parsed[ZenohPut].data == b'x' +parsed = ZenohResponse(raw(ZenohResponse(request_id=1, key_scope=1)/ZenohErr(data=b'bad'))) +assert parsed[ZenohErr].data == b'bad' + += ZenohResponseFinal: closes a response stream +assert raw(ZenohResponseFinal(request_id=9)) == b'\x1a\x09' +assert ZenohResponseFinal(b'\x1a\x09').request_id == 9 + += ZenohInterest: the mode is encoded in the header +interest = ZenohInterest(mode='Current', interest_id=1, options=0x0f) +data = raw(interest) +assert data == b'\x39\x01\x0f' +parsed = ZenohInterest(data) +assert parsed.mode == 0x01 +assert 'tokens' in parsed.options + += ZenohInterest: a Final interest has no options +interest = ZenohInterest(mode='Final', interest_id=2) +data = raw(interest) +assert data == b'\x19\x02' +assert ZenohInterest(data).options is None + += ZenohInterest: the R and N options announce a key expression +interest = ZenohInterest(interest_id=1, options=0x30, key_scope=1, key_suffix='a/*') +data = raw(interest) +assert data == b'\x39\x01\x30\x01\x03a/*' +parsed = ZenohInterest(data) +assert parsed.key_scope == 1 +assert parsed.key_suffix == 'a/*' + += ZenohDeclare: the I flag announces the interest ID +decl = ZenohDeclare()/ZenohDeclareFinal() +assert raw(decl) == b'\x1e\x1a' +decl = ZenohDeclare(interest_id=4)/ZenohDeclareFinal() +data = raw(decl) +assert data == b'\x3e\x04\x1a' +parsed = ZenohDeclare(data) +assert parsed.flag_i == 1 +assert parsed.interest_id == 4 +assert ZenohDeclareFinal in parsed + += ZenohNetworkOAM: body follows the extensions +oam = ZenohNetworkOAM(enc='Z64', oam_id=1, value=2) +assert raw(oam) == b'\x3f\x01\x02' +assert ZenohNetworkOAM(b'\x3f\x01\x02').value == 2 + + +############ +############ ++ Declarations + += ZenohDeclareKeyExpr: binds an expression ID to a key expression +d = ZenohDeclareKeyExpr(expr_id=1, key_scope=0, key_suffix='demo/example/test') +data = raw(d) +assert data == b'\x20\x01\x00\x11demo/example/test' +parsed = ZenohDeclareKeyExpr(data) +assert parsed.expr_id == 1 +assert parsed.key_suffix == 'demo/example/test' + += ZenohUndeclareKeyExpr: releases an expression ID +assert raw(ZenohUndeclareKeyExpr(expr_id=1)) == b'\x01\x01' + += ZenohDeclareSubscriber: entity ID and key expression +d = ZenohDeclareSubscriber(flag_m=1, subscriber_id=1, key_scope=1) +data = raw(d) +assert data == b'\x42\x01\x01' +parsed = ZenohDeclareSubscriber(data) +assert parsed.subscriber_id == 1 +assert parsed.key_scope == 1 + += ZenohDeclareQueryable and ZenohDeclareToken +assert raw(ZenohDeclareQueryable(flag_m=1, queryable_id=2, key_scope=1)) == b'\x44\x02\x01' +assert raw(ZenohDeclareToken(flag_m=1, token_id=3, key_scope=2)) == b'\x46\x03\x02' + += Undeclarations carry their entity ID only +assert raw(ZenohUndeclareSubscriber(subscriber_id=1)) == b'\x03\x01' +assert raw(ZenohUndeclareQueryable(queryable_id=2)) == b'\x05\x02' +assert raw(ZenohUndeclareToken(token_id=3)) == b'\x07\x03' + += ZenohDeclareFinal: header only +assert raw(ZenohDeclareFinal()) == b'\x1a' + += Declarations are dispatched by their declaration ID +for cls in [ZenohDeclareKeyExpr, ZenohUndeclareKeyExpr, ZenohDeclareSubscriber, + ZenohUndeclareSubscriber, ZenohDeclareQueryable, ZenohUndeclareQueryable, + ZenohDeclareToken, ZenohUndeclareToken, ZenohDeclareFinal]: + parsed = ZenohDeclare(raw(ZenohDeclare()/cls())) + assert cls in parsed, cls.__name__ + + +############ +############ ++ Zenoh messages + += ZenohPut: payload only +put = ZenohPut(data=b'hello') +data = raw(put) +assert data == b'\x01\x05hello' +assert ZenohPut(data).data == b'hello' + += ZenohPut: the E flag announces the encoding +put = ZenohPut(encoding=4, data=b'hi') +data = raw(put) +assert data == b'\x41\x08\x02hi' +parsed = ZenohPut(data) +assert parsed.flag_e == 1 +assert parsed.encoding == 4 + += ZenohPut: the T flag announces the timestamp +put = ZenohPut(timestamp=(1, b'\x01'), data=b'') +data = raw(put) +assert data == b'\x21\x01\x01\x01\x00' +parsed = ZenohPut(data) +assert parsed.flag_t == 1 +assert parsed.timestamp == (1, b'\x01') + += ZenohPut: timestamp, encoding, extensions and payload keep their order +put = ZenohPut(timestamp=(1, b'\x01'), encoding=4, extensions=[ZenohExtension(eid=1)], data=b'x') +data = raw(put) +assert data == b'\xe1\x01\x01\x01\x08\x01\x01x' +assert raw(ZenohPut(data)) == data + += ZenohDel: optional timestamp +assert raw(ZenohDel()) == b'\x02' +assert raw(ZenohDel(timestamp=(1, b'\x01'))) == b'\x22\x01\x01\x01' + += ZenohQuery: consolidation and selector parameters +q = ZenohQuery(consolidation=3, parameters='arg=42') +data = raw(q) +assert data == b'\x63\x03\x06arg=42' +parsed = ZenohQuery(data) +assert parsed.consolidation == 3 +assert parsed.parameters == 'arg=42' +assert raw(ZenohQuery()) == b'\x03' + += ZenohReply: carries a Put or a Del +parsed = ZenohReply(raw(ZenohReply()/ZenohPut(data=b'x'))) +assert ZenohPut in parsed +assert raw(ZenohReply(consolidation=1)) == b'\x24\x01' + += ZenohErr: encoding and payload +err = ZenohErr(encoding=4, data=b'oops') +data = raw(err) +assert data == b'\x45\x08\x04oops' +parsed = ZenohErr(data) +assert parsed.encoding == 4 +assert parsed.data == b'oops' + + +############ +############ ++ Batches and framing + += ZenohBatch: a datagram holds a sequence of transport messages +batch = ZenohBatch(messages=[ZenohKeepAlive(), ZenohClose(reason=1)]) +data = raw(batch) +assert data == b'\x04\x03\x01' +parsed = ZenohBatch(data) +assert len(parsed.messages) == 2 +assert isinstance(parsed.messages[0], ZenohKeepAlive) +assert parsed.messages[1].reason == 1 + += ZenohBatch: a Frame may be followed by another transport message +batch = ZenohBatch(messages=[ZenohFrame(sn=1, messages=[ZenohPush(key_scope=1)/ZenohPut(data=b'x')]), ZenohKeepAlive()]) +data = raw(batch) +parsed = ZenohBatch(data) +assert len(parsed.messages) == 2 +assert isinstance(parsed.messages[1], ZenohKeepAlive) +assert raw(parsed) == data + += ZenohStreamBatch: stream links prefix the batch with its length +batch = ZenohStreamBatch(messages=[ZenohKeepAlive()]) +data = raw(batch) +assert data == b'\x01\x00\x04' +parsed = ZenohStreamBatch(data) +assert parsed.len == 1 +assert isinstance(parsed.messages[0], ZenohKeepAlive) + += ZenohStreamBatch: the length is computed from the messages +batch = ZenohStreamBatch(messages=[ZenohFrame(sn=1, messages=[ZenohPush(key_scope=1)/ZenohPut(data=b'hello')])]) +data = raw(batch) +assert data[:2] == b'\x0b\x00' +assert len(data) == 13 +assert raw(ZenohStreamBatch(data)) == data + += ZenohStreamBatch: several batches may share one segment +data = raw(ZenohStreamBatch(messages=[ZenohKeepAlive()])) * 3 +parsed = ZenohStreamBatch(data) +assert list(parsed.layers()).count(ZenohStreamBatch) == 3 +assert raw(parsed) == data + += ZenohStreamBatch: a truncated trailing batch is left as payload +data = raw(ZenohStreamBatch(messages=[ZenohKeepAlive()])) + b'\x20\x00\x04' +parsed = ZenohStreamBatch(data) +assert list(parsed.layers()).count(ZenohStreamBatch) == 1 +assert raw(parsed) == data + += Transport messages: TCP port 7447 is a stream batch +pkt = IP()/TCP(sport=45678, dport=ZENOH_PORT_TRANSPORT)/ZenohStreamBatch(messages=[ZenohKeepAlive()]) +parsed = IP(raw(pkt)) +assert ZenohStreamBatch in parsed +assert ZenohKeepAlive in parsed + += Transport messages: UDP port 7447 is a plain batch +pkt = IP()/UDP(sport=ZENOH_PORT_TRANSPORT, dport=45678)/ZenohBatch(messages=[ZenohJoin(zid=b'\x01')]) +parsed = IP(raw(pkt)) +assert ZenohBatch in parsed +assert ZenohJoin in parsed + += Batches: an unknown transport message ID stops the dissection as Raw +parsed = ZenohBatch(b'\x0f\x00') +assert Raw in parsed + += Bindings: building sets the destination port only +pkt = UDP()/ZenohScouting() +assert pkt.dport == ZENOH_PORT_SCOUTING +assert pkt.sport != ZENOH_PORT_SCOUTING +pkt = TCP()/ZenohStreamBatch() +assert pkt.dport == ZENOH_PORT_TRANSPORT + + +############ +############ ++ Regression vectors captured from zenohd 1.9.0 + += Scout sent by a peer looking for a router +data = bytes.fromhex('010907') +pkt = ZenohScouting(data) +assert isinstance(pkt, ZenohScout) +assert pkt.version == ZENOH_VERSION +assert pkt.what == 0x07 +assert pkt.flag_i == 0 +assert raw(pkt) == data + += InitSyn of a client, with the QoS, shared memory and patch extensions +data = bytes.fromhex('2000c109f2ecf54bd31a96e15921118ed998afc8250ac8ff81c205dde2f49e0b2701') +pkt = ZenohStreamBatch(data) +init = pkt.messages[0] +assert isinstance(init, ZenohInit) +assert init.flag_a == 0 +assert init.whatami == 2 +assert init.zid == bytes.fromhex('ecf54bd31a96e15921118ed998afc825') +assert init.batch_size == 65480 +assert init.fsn_resolution == 0x02 +assert [e.eid for e in init.extensions] == [1, 2, 7] +assert raw(pkt) == data + += InitAck of a router, carrying the cookie +data = bytes.fromhex('4b00e109f02e2d91488b1b5b18f9cea99572734ad50a00c03130710eae9fa5071655875b6b42cd07eadbd6e21442b744d84d4e783d3502baf3e2e742256be17ef241e5bf5e3c063592e7812701') +pkt = ZenohStreamBatch(data) +init = pkt.messages[0] +assert isinstance(init, ZenohInit) +assert init.flag_a == 1 +assert init.whatami == 0 +assert init.batch_size == 49152 +assert len(init.cookie) == 49 +assert raw(pkt) == data + += OpenSyn echoing the cookie of the InitAck +data = bytes.fromhex('3800420aa39b904e3130710eae9fa5071655875b6b42cd07eadbd6e21442b744d84d4e783d3502baf3e2e742256be17ef241e5bf5e3c063592e7') +pkt = ZenohStreamBatch(data) +op = pkt.messages[0] +assert isinstance(op, ZenohOpen) +assert op.flag_a == 0 +assert op.flag_t == 1 +assert op.lease == 10 +assert op.initial_sn == 163843491 +assert len(op.cookie) == 49 +assert raw(pkt) == data + += OpenAck without cookie +data = bytes.fromhex('0600620aa28d8b70') +pkt = ZenohStreamBatch(data) +op = pkt.messages[0] +assert isinstance(op, ZenohOpen) +assert op.flag_a == 1 +assert op.cookie is None +assert raw(pkt) == data + += Frame with a Declare of a key expression +data = bytes.fromhex('1f00a5a39b904e31009e21082001001164656d6f2f6578616d706c652f74657374') +pkt = ZenohStreamBatch(data) +frame = pkt.messages[0] +assert isinstance(frame, ZenohFrame) +assert frame.flag_r == 1 +assert frame.sn == 163843491 +assert frame.extensions[0].eid == 1 +assert frame.extensions[0].mandatory == 1 +kexpr = frame.messages[0][ZenohDeclareKeyExpr] +assert kexpr.expr_id == 1 +assert kexpr.key_scope == 0 +assert kexpr.key_suffix == 'demo/example/test' +assert raw(pkt) == data + += Frame with a Declare of a subscriber +data = bytes.fromhex('0d00a5a49b904e31009e2108420101') +pkt = ZenohStreamBatch(data) +sub = pkt[ZenohDeclareSubscriber] +assert sub.subscriber_id == 1 +assert sub.key_scope == 1 +assert sub.flag_m == 1 +assert raw(pkt) == data + += Frame with a Declare of a queryable +data = bytes.fromhex('0d00a5a59b904e31009e2108440201') +pkt = ZenohStreamBatch(data) +assert pkt[ZenohDeclareQueryable].queryable_id == 2 +assert raw(pkt) == data + += Frame with two Declare messages, a token and its withdrawal +data = bytes.fromhex('1600a5a79b904e31009e21084603029e210887035f020000') +pkt = ZenohStreamBatch(data) +frame = pkt.messages[0] +assert len(frame.messages) == 2 +assert frame.messages[0][ZenohDeclareToken].token_id == 3 +undecl = frame.messages[1][ZenohUndeclareToken] +assert undecl.token_id == 3 +assert undecl.extensions[0].eid == 0x0f +assert undecl.extensions[0].mandatory == 1 +assert raw(pkt) == data + += Frame with an Interest restricted to a key expression +data = bytes.fromhex('0d00a5a89b904e3100f90453012108') +pkt = ZenohStreamBatch(data) +interest = pkt[ZenohInterest] +assert interest.interest_id == 4 +assert interest.mode == 0x03 +assert 'restricted' in interest.options +assert 'named' not in interest.options +assert interest.key_scope == 1 +assert raw(pkt) == data + += Frame with a Push of a Put payload +data = bytes.fromhex('100025a39b904e5d01010768656c6c6f2d30') +pkt = ZenohStreamBatch(data) +push = pkt[ZenohPush] +assert push.key_scope == 1 +assert push.flag_m == 1 +assert pkt[ZenohPut].data == b'hello-0' +assert raw(pkt) == data + += Frame with a Declare closing an Interest +data = bytes.fromhex('0c00a5a28d8b703100be0421081a') +pkt = ZenohStreamBatch(data) +decl = pkt[ZenohDeclare] +assert decl.flag_i == 1 +assert decl.interest_id == 4 +assert ZenohDeclareFinal in pkt +assert raw(pkt) == data + += Frame with a Request holding a Query +data = bytes.fromhex('160025a49b904edc0101a10d26904e6303066172673d3432') +pkt = ZenohStreamBatch(data) +req = pkt[ZenohRequest] +assert req.request_id == 1 +assert req.key_scope == 1 +assert [e.eid for e in req.extensions] == [1, 6] +assert req.extensions[1].value == 10000 +query = pkt[ZenohQuery] +assert query.consolidation == 3 +assert query.parameters == 'arg=42' +assert raw(pkt) == data + += Frame with a ResponseFinal +data = bytes.fromhex('090025a28d8b709a01210d') +pkt = ZenohStreamBatch(data) +assert pkt[ZenohResponseFinal].request_id == 1 +assert raw(pkt) == data + += Close of a link +data = bytes.fromhex('02000300') +pkt = ZenohStreamBatch(data) +close = pkt.messages[0] +assert isinstance(close, ZenohClose) +assert close.flag_s == 0 +assert close.reason == 0 +assert raw(pkt) == data + += Every captured payload dissects without leaving undecoded bytes +CAPTURED = [ + '2000c109f2ecf54bd31a96e15921118ed998afc8250ac8ff81c205dde2f49e0b2701', + '4b00e109f02e2d91488b1b5b18f9cea99572734ad50a00c03130710eae9fa5071655875b6b42cd07eadbd6e21442b744d84d4e783d3502baf3e2e742256be17ef241e5bf5e3c063592e7812701', + '3800420aa39b904e3130710eae9fa5071655875b6b42cd07eadbd6e21442b744d84d4e783d3502baf3e2e742256be17ef241e5bf5e3c063592e7', + '0600620aa28d8b70', + '1f00a5a39b904e31009e21082001001164656d6f2f6578616d706c652f74657374', + '0d00a5a49b904e31009e2108420101', + '0d00a5a59b904e31009e2108440201', + '2300a5a69b904e31009e21082002001564656d6f2f6c6976656c696e6573732f7363617079', + '1600a5a79b904e31009e21084603029e210887035f020000', + '0d00a5a89b904e3100f90453012108', + '100025a39b904e5d01010768656c6c6f2d30', + '0c00a5a28d8b703100be0421081a', + '160025a49b904edc0101a10d26904e6303066172673d3432', + '090025a28d8b709a01210d', + '02000300', +] +for hexdata in CAPTURED: + data = bytes.fromhex(hexdata) + pkt = ZenohStreamBatch(data) + assert not pkt.haslayer(Raw), hexdata + assert not pkt.haslayer(Padding), hexdata + assert raw(pkt) == data, hexdata + pkt.show(dump=True) + assert pkt.summary() + + +############ +############ ++ Robustness + += Every message class rebuilds after a dissection of its own bytes +CLASSES = [ZenohScout, ZenohHello, ZenohInit, ZenohOpen, ZenohClose, ZenohKeepAlive, + ZenohFrame, ZenohFragment, ZenohJoin, ZenohTransportOAM, ZenohPush, + ZenohRequest, ZenohResponse, ZenohResponseFinal, ZenohInterest, + ZenohDeclare, ZenohNetworkOAM, ZenohPut, ZenohDel, ZenohQuery, + ZenohReply, ZenohErr, ZenohDeclareKeyExpr, ZenohUndeclareKeyExpr, + ZenohDeclareSubscriber, ZenohUndeclareSubscriber, ZenohDeclareQueryable, + ZenohUndeclareQueryable, ZenohDeclareToken, ZenohUndeclareToken, + ZenohDeclareFinal] +for cls in CLASSES: + data = raw(cls()) + assert raw(cls(data)) == data, cls.__name__ + += Fuzzed messages rebuild to the same bytes +import random +from scapy.packet import fuzz +random.seed(0x5eed) +for cls in CLASSES: + for _ in range(20): + data = raw(fuzz(cls())) + assert raw(cls(data)) == data, cls.__name__ + += Random bytes never raise an unexpected exception +import struct +random.seed(0xf00d) +for _ in range(500): + data = bytes(random.randrange(256) for _ in range(random.randint(0, 48))) + for cls in [ZenohScouting, ZenohBatch, ZenohStreamBatch]: + try: + pkt = cls(data) + raw(pkt) + pkt.show(dump=True) + except struct.error: + pass + += Truncated messages do not loop forever +data = raw(ZenohStreamBatch(messages=[ZenohFrame(sn=1, messages=[ZenohPush(key_scope=1, key_suffix='a/b')/ZenohPut(data=b'x' * 10)])])) +for i in range(len(data) + 1): + try: + ZenohStreamBatch(data[:i]).show(dump=True) + except struct.error: + pass + + +############ +############ ++ Summaries + += Messages have a readable summary +assert 'ZenohScout' in ZenohScout().summary() +assert 'demo' in (ZenohPush(key_scope=1, key_suffix='demo/a')/ZenohPut(data=b'x')).summary() +assert 'sn=7' in ZenohFrame(sn=7).summary() +assert 'id=3' in ZenohRequest(request_id=3, key_scope=1).summary() +pkt = ZenohStreamBatch(messages=[ZenohFrame(sn=1, messages=[ZenohPush(key_scope=1)/ZenohPut(data=b'x')])]) +assert 'ZenohFrame' in ZenohStreamBatch(raw(pkt)).summary()