diff --git a/scapy/contrib/ieee1722.py b/scapy/contrib/ieee1722.py new file mode 100644 index 00000000000..729483173f5 --- /dev/null +++ b/scapy/contrib/ieee1722.py @@ -0,0 +1,1705 @@ +# SPDX-License-Identifier: GPL-2.0-only +# This file is part of Scapy +# See https://scapy.net/ for more information +# Copyright (C) Naresh Nayak + +# scapy.contrib.description = IEEE 1722 serialization formats +# scapy.contrib.status = loads + +""" +IEEE 1722 serialization formats for Scapy +""" + +import struct +from enum import Enum + +from scapy.all import UDP, Ether, bind_layers # pylint: disable=no-name-in-module +from scapy.fields import ( + BitEnumField, + BitField, + BitFieldLenField, + ByteField, + ConditionalField, + FieldLenField, + IntField, + LenField, + LongField, + PacketListField, + ShortField, + StrFixedLenField, + XByteEnumField, + XByteField, + XIntField, + XLongField, + XShortField, +) +from scapy.packet import Packet + + +class AvtpStreamType(Enum): + """ + AVTP Stream Subtypes - Table 7 - IEEE 1722-2025 + """ + + IIDC_61883 = 0x0 + MMA_STREAM = 0x1 + AAF = 0x2 + CVF = 0x3 + CRF = 0x4 + TSCF = 0x5 + SVF = 0x6 + RVF = 0x7 + AEF_CONTINUOUS = 0x6E + VSF_STREAM = 0x6F + EF_STREAM = 0x7F + NTSCF = 0x82 + IEEE_8021_MLAA = 0xEB + ESCF = 0xEC + EECF = 0xED + AEF_DISCRETE = 0xEE + ADP = 0xFA + AECP = 0xFB + ACMP = 0xFC + MAAP = 0xFE + EF_CONTROL = 0xFF + + +class AvtpAcfType(Enum): + """ + AVTP ACF Message Types - Table 23 - IEEE 1722-2025 + """ + + ACF_FLEXRAY = 0x0 + ACF_CAN = 0x1 + ACF_CAN_BRIEF = 0x2 + ACF_LIN = 0x3 + ACF_MOST = 0x4 + ACF_GPC = 0x5 + ACF_SERIAL = 0x6 + ACF_PARALLEL = 0x7 + ACF_SENSOR = 0x8 + ACF_SENSOR_BRIEF = 0x9 + ACF_AECP = 0xA + ACF_ANCILLARY = 0xB + ACF_GISF = 0xC + ACF_GBB = 0xD + ACF_ABB = 0xE + ACF_I2C = 0xF + ACF_I2C_BRIEF = 0x10 + ACF_CAN_XL = 0x11 + ACF_CAN_XL_BRIEF = 0x12 + ACF_CAN_V2 = 0x21 + ACF_CAN_BRIEF_V2 = 0x22 + ACF_LIN_V2 = 0x23 + ACF_CHECKSUM = 0x76 + ACF_CRC = 0x77 + + +class AvtpHeaderVersion(Enum): + """ + AVTP Header Versions - Table 8 - IEEE 1722-2025 + """ + + V0 = 0 + V1 = 1 + AVTP_MAX_HEADER_VERSION = 1 + + +class AvtpCommonHeader(Packet): + """ + Common Header (Clause 4.7.3) for the AVTP-2025 + """ + + name = "AVTP Common Header" + fields_desc = [ + ConditionalField( + IntField(name="encapsulation_sequence_num", default=0), + lambda pkt: isinstance(pkt.underlayer, UDP), + ), + XByteEnumField( + name="subtype", + default=None, + enum={i.name: i.value for i in AvtpStreamType}, + ), + BitField(name="h", size=1, default=0), + BitField(name="version", size=3, default=0), + ] + common_header_variants = {} + + @classmethod + def register_variant(cls): + """ + Register a variant of the common header based on subtype and version. + We use a tuple of (subtype, version) as the key to store the variant + class in a dictionary. For registering a variant, the subclass should + set the default values for subtype and version fields. + """ + + if cls.subtype.default is not None: + key = (cls.subtype.default, cls.version.default) + cls.common_header_variants[key] = cls + + @classmethod + def dispatch_hook(cls, pkt=None, **kargs): # pylint: disable=unused-argument + """ + Dispatch the appropriate class based on the parsed subtype and version. + """ + if pkt is not None: + if isinstance(kargs["_underlayer"], UDP): + parsed_type = ord(pkt[4:5]) + parsed_version = (ord(pkt[5:6]) & 0x70) >> 4 + else: + parsed_type = ord(pkt[0:1]) + parsed_version = (ord(pkt[1:2]) & 0x70) >> 4 + + key = (parsed_type, parsed_version) + return cls.common_header_variants.get(key, cls) + return cls + + +class _CommonStreamHeaderV0(AvtpCommonHeader): + name = "AVTP Common Stream Header v0" + fields_desc = [ + ConditionalField( + IntField(name="encapsulation_sequence_num", default=0), + lambda pkt: isinstance(pkt.underlayer, UDP), + ), + XByteEnumField( + name="subtype", + default=None, + enum={i.name: i.value for i in AvtpStreamType}, + ), + BitField(name="h", size=1, default=0), + BitField(name="version", size=3, default=0), + XByteField(name="sequence_num", default=0), + BitField(name="format_specific_data_1", size=7, default=0), + BitField(name="tu", size=1, default=0), + XLongField(name="stream_id", default=0), + XIntField(name="avtp_timestamp", default=0), + XIntField(name="format_specific_data_2", default=0), + LenField(name="stream_data_length", default=None), + XShortField(name="format_specific_data_3", default=0), + ] + + +class _CommonStreamHeaderV1(AvtpCommonHeader): + name = "AVTP Common Stream Header v1" + fields_desc = [ + ConditionalField( + IntField(name="encapsulation_sequence_num", default=0), + lambda pkt: isinstance(pkt.underlayer, UDP), + ), + XByteEnumField( + name="subtype", + default=None, + enum={i.name: i.value for i in AvtpStreamType}, + ), + BitField(name="h", size=1, default=0), + BitField(name="version", size=3, default=1), + BitField(name="mr", size=1, default=0), + BitField(name="f_s_d", size=2, default=0), + BitField(name="tv", size=1, default=0), + XByteField(name="format_specific_data_0", default=0), + BitField(name="format_specific_data_1", size=7, default=0), + BitField(name="tu", size=1, default=0), + XLongField(name="stream_id", default=0), + XIntField(name="sequence_num", default=0), + XLongField(name="avtp_timestamp", default=0), + XLongField(name="ptp_grandmaster_identity", default=0), + XIntField(name="format_specific_data_2", default=0), + LenField(name="stream_data_length", default=None), + XShortField(name="format_specific_data_3", default=0), + ] + + +class AvtpCommonStreamHeader(AvtpCommonHeader): + """ + Common Stream Header (Clause 4.7.4) for the AVTP-2025 + """ + + name = "AVTP Common Stream Header" + + @classmethod + def dispatch_hook(cls, pkt=None, **kargs): + version = 0 + if "version" in kargs: + version = kargs.get("version", 0) == 1 + else: + if pkt is not None: + parsed_version = (ord(pkt[1:2]) & 0x70) >> 4 + version = parsed_version == 1 + return _CommonStreamHeaderV1 if version else _CommonStreamHeaderV0 + + +class AvtpCommonControlHeader(AvtpCommonHeader): + """ + Common Control Header (Clause 4.7.5) for the AVTP-2025 + """ + + name = "AVTP Common Control Header" + fields_desc = [ + ConditionalField( + IntField(name="encapsulation_sequence_num", default=0), + lambda pkt: isinstance(pkt.underlayer, UDP), + ), + XByteEnumField( + name="subtype", + default=None, + enum={i.name: i.value for i in AvtpStreamType}, + ), + BitField(name="sv", size=1, default=0), + BitField(name="version", size=3, default=0), + BitField(name="format_specific_data", size=9, default=0), + BitField(name="control_data_length", size=11, default=None), + XLongField(name="stream_id", default=0), + ] + + def post_build(self, pkt: bytes, pay: bytes) -> bytes: + # Update the length fields on packet building + pay_length = len(pay) if len(pay) < 2**11 else 0 + current_length = ( + int.from_bytes(pkt[-10:-8], byteorder="big") & 0xF800 + ) | pay_length + pkt = pkt[:-10] + struct.pack("!H", current_length) + pkt[-8:] + pkt += pay + return pkt + + +class _AlternativeHeaderV0(AvtpCommonHeader): + name = "AVTP Alternative Header v0" + fields_desc = [ + ConditionalField( + IntField(name="encapsulation_sequence_num", default=0), + lambda pkt: isinstance(pkt.underlayer, UDP), + ), + XByteEnumField( + name="subtype", + default=None, + enum={i.name: i.value for i in AvtpStreamType}, + ), + BitField(name="h", size=1, default=0), + BitField(name="version", size=3, default=0), + ] + + +class _AlternativeHeaderV1(AvtpCommonHeader): + name = "AVTP Alternative Header v1" + fields_desc = [ + ConditionalField( + IntField(name="encapsulation_sequence_num", default=0), + lambda pkt: isinstance(pkt.underlayer, UDP), + ), + XByteEnumField( + name="subtype", + default=None, + enum={i.name: i.value for i in AvtpStreamType}, + ), + BitField(name="h", size=1, default=0), + BitField(name="version", size=3, default=0), + BitField(name="reserved_1", size=20, default=0), + XIntField(name="sequence_num", default=0), + XLongField(name="gptp_grandmaster_identity", default=0), + BitField(name="reserved_2", size=12, default=0), + ] + + +class AvtpAlternativeHeader(AvtpCommonHeader): + """ + Alternative Header (Clause 4.7.6) for the AVTP-2025 + """ + + name = "AVTP Alternative Header" + + @classmethod + def dispatch_hook(cls, pkt=None, **kargs): + version = 0 + if "version" in kargs: + version = kargs.get("version", 0) == 1 + else: + if pkt is not None: + parsed_version = (ord(pkt[1:2]) & 0x70) >> 4 + version = parsed_version == 1 + return _AlternativeHeaderV1 if version else _AlternativeHeaderV0 + + +class AvtpAcfHeader(Packet): + """ + Header for ACF Messages - Clause 9.4 - IEEE 1722 - 2025 + """ + + name = "ACF Header" + fields_desc = [ + BitEnumField( + name="acf_msg_type", + default=None, + size=7, + enum={i.name: i.value for i in AvtpAcfType}, + ), + BitField(name="acf_msg_length", size=9, default=None), + ] + + acf_variants = {} + + @classmethod + def register_variant(cls): + """ + Register a variant of the ACF header based on acf_msg_type. + """ + if cls.acf_msg_type.default is not None: + cls.acf_variants[cls.acf_msg_type.default] = cls + + @classmethod + def dispatch_hook(cls, pkt=None, **kargs): # pylint: disable=unused-argument + """ + Dispatch to the appropriate ACF header variant based on the acf_msg_type field. + """ + if pkt is not None: + tmp_type = (ord(pkt[:1]) & 0xFE) >> 1 + return cls.acf_variants.get(tmp_type, cls) + return cls + + +class AvtpAcfFlexrayHeader(AvtpAcfHeader): + """ + Header for FlexRay Messages - Clause 9.4.2 - IEEE 1722 - 2025 + """ + + name = "ACF Flexray Header" + fields_desc = [ + BitEnumField( + name="acf_msg_type", + default=AvtpAcfType.ACF_FLEXRAY, + size=7, + enum={i.name: i.value for i in AvtpAcfType}, + ), + BitField(name="acf_msg_length", size=9, default=None), + BitField(name="padlength", size=2, default=0), + BitField(name="mtv", size=1, default=0), + BitField(name="fr_bus_id", size=5, default=0), + BitField(name="rsv", size=2, default=0), + BitField(name="chan", size=2, default=0), + BitField(name="str", size=1, default=0), + BitField(name="syn", size=1, default=0), + BitField(name="pre", size=1, default=0), + BitField(name="nfi", size=1, default=0), + LongField(name="message_timestamp", default=0), + BitField(name="fr_frame_id", size=11, default=0), + BitField(name="reserved", size=15, default=0), + BitField(name="cycle", size=6, default=0), + ] + + def do_dissect_payload(self, s): + return super().do_dissect_payload(s[: len(s) - self.padlength]) + + def extract_padding(self, s): + """ + Extract padding from the payload. + """ + payload_length = self.acf_msg_length * 4 - len(self.self_build()) + return s[0:payload_length], s[payload_length:] + + def post_build(self, pkt: bytes, pay: bytes) -> bytes: + # Do padding + if len(pay) % 4: + pad = [0] * (4 - len(pay) % 4) + pad_byte = pkt[2] | ((4 - len(pay) % 4) << 6) + pkt = pkt[0:2] + struct.pack("!B", pad_byte) + pkt[3:] + pay += bytes(pad) + + # Take care for length updation if the acf_msg_length field is not set + if self.acf_msg_length is None: + acf_length = int((len(pkt) + len(pay)) / 4) & 0x1FF + first_byte = (pkt[0] & 0xFE) | ((acf_length >> 8) & 0x01) + pkt = ( + struct.pack("!B", first_byte) + + struct.pack("!B", acf_length & 0xFF) + + pkt[2:] + ) + + return pkt + pay + + +class AvtpAcfCanHeader(AvtpAcfHeader): + """ + Header for CAN/CAN-FD Messages - Clause 9.4.3 - IEEE 1722 - 2025 + """ + + name = "ACF CAN Header" + fields_desc = [ + BitEnumField( + name="acf_msg_type", + default=AvtpAcfType.ACF_CAN, + size=7, + enum={i.name: i.value for i in AvtpAcfType}, + ), + BitField(name="acf_msg_length", size=9, default=None), + BitField(name="padlength", size=2, default=0), + BitField(name="mtv", size=1, default=0), + BitField(name="rtr", size=1, default=0), + BitField(name="eff", size=1, default=0), + BitField(name="brs", size=1, default=0), + BitField(name="fdf", size=1, default=0), + BitField(name="esi", size=1, default=0), + BitField(name="rsv1", size=3, default=0), + BitField(name="can_bus_id", size=5, default=0), + LongField(name="message_timestamp", default=0), + BitField(name="rsv2", size=3, default=0), + BitField(name="can_identifier", size=29, default=0), + ] + + def do_dissect_payload(self, s): + return super().do_dissect_payload(s[: len(s) - self.padlength]) + + def extract_padding(self, s): + payload_length = self.acf_msg_length * 4 - len(self.self_build()) + return s[0:payload_length], s[payload_length:] + + def post_build(self, pkt: bytes, pay: bytes) -> bytes: + # Do padding + if len(pay) % 4: + pad = [0] * (4 - len(pay) % 4) + pad_byte = pkt[2] | ((4 - len(pay) % 4) << 6) + pkt = pkt[0:2] + struct.pack("!B", pad_byte) + pkt[3:] + pay += bytes(pad) + + # Take care for length updation if the acf_msg_length field is not set + if self.acf_msg_length is None: + acf_length = int((len(pkt) + len(pay)) / 4) & 0x1FF + first_byte = (pkt[0] & 0xFE) | ((acf_length >> 8) & 0x01) + pkt = ( + struct.pack("!B", first_byte) + + struct.pack("!B", acf_length & 0xFF) + + pkt[2:] + ) + + return pkt + pay + + +class AvtpAcfCanV2Header(AvtpAcfHeader): + """ + Header for CAN/CAN-FD version 2 Messages - Clause 9.4.3 - IEEE 1722 - 2025 + """ + + name = "ACF CAN V2 Header" + + fields_desc = [ + BitEnumField( + name="acf_msg_type", + default=AvtpAcfType.ACF_CAN_V2, + size=7, + enum={i.name: i.value for i in AvtpAcfType}, + ), + BitField(name="acf_msg_length", size=9, default=None), + BitField(name="padlength", size=2, default=0), + BitField(name="mtv", size=1, default=0), + BitField(name="rtr", size=1, default=0), + BitField(name="eff", size=1, default=0), + BitField(name="can_bus_id", size=11, default=0), + LongField(name="message_timestamp", default=0), + BitField(name="brs", size=1, default=0), + BitField(name="fdf", size=1, default=0), + BitField(name="esi", size=1, default=0), + BitField(name="can_identifier", size=29, default=0), + ] + + def do_dissect_payload(self, s): + return super().do_dissect_payload(s[: len(s) - self.padlength]) + + def extract_padding(self, s): + payload_length = self.acf_msg_length * 4 - len(self.self_build()) + return s[0:payload_length], s[payload_length:] + + def post_build(self, pkt: bytes, pay: bytes) -> bytes: + # Do padding + if len(pay) % 4: + pad = [0] * (4 - len(pay) % 4) + pad_byte = pkt[2] | ((4 - len(pay) % 4) << 6) + pkt = pkt[0:2] + struct.pack("!B", pad_byte) + pkt[3:] + pay += bytes(pad) + + # Take care for length updation if the acf_msg_length field is not set + if self.acf_msg_length is None: + acf_length = int((len(pkt) + len(pay)) / 4) & 0x1FF + first_byte = (pkt[0] & 0xFE) | ((acf_length >> 8) & 0x01) + pkt = ( + struct.pack("!B", first_byte) + + struct.pack("!B", acf_length & 0xFF) + + pkt[2:] + ) + + return pkt + pay + + +class AvtpAcfCanBriefHeader(AvtpAcfHeader): + """ + Header for CAN/CAN-FD Brief Messages - Clause 9.4.4 - IEEE 1722 - 2025 + """ + + name = "ACF CAN BRIEF Header" + fields_desc = [ + BitEnumField( + name="acf_msg_type", + default=AvtpAcfType.ACF_CAN_BRIEF, + size=7, + enum={i.name: i.value for i in AvtpAcfType}, + ), + BitField(name="acf_msg_length", size=9, default=None), + BitField(name="padlength", size=2, default=0), + BitField(name="mtv", size=1, default=0), + BitField(name="rtr", size=1, default=0), + BitField(name="eff", size=1, default=0), + BitField(name="brs", size=1, default=0), + BitField(name="fdf", size=1, default=0), + BitField(name="esi", size=1, default=0), + BitField(name="rsv1", size=3, default=0), + BitField(name="can_bus_id", size=5, default=0), + BitField(name="rsv2", size=3, default=0), + BitField(name="can_identifier", size=29, default=0), + ] + + def do_dissect_payload(self, s): + return super().do_dissect_payload(s[: len(s) - self.padlength]) + + def extract_padding(self, s): + payload_length = self.acf_msg_length * 4 - len(self.self_build()) + return s[0:payload_length], s[payload_length:] + + def post_build(self, pkt: bytes, pay: bytes) -> bytes: + # Do padding + if len(pay) % 4: + pad = [0] * (4 - len(pay) % 4) + pad_byte = pkt[2] | ((4 - len(pay) % 4) << 6) + pkt = pkt[0:2] + struct.pack("!B", pad_byte) + pkt[3:] + pay += bytes(pad) + + # Take care for length updation if the acf_msg_length field is not set + if self.acf_msg_length is None: + acf_length = int((len(pkt) + len(pay)) / 4) & 0x1FF + first_byte = (pkt[0] & 0xFE) | ((acf_length >> 8) & 0x01) + pkt = ( + struct.pack("!B", first_byte) + + struct.pack("!B", acf_length & 0xFF) + + pkt[2:] + ) + + return pkt + pay + + +class AvtpAcfCanBriefV2Header(AvtpAcfHeader): + """ + Header for CAN/CAN-FD Brief version 2 Messages - Clause 9.4.4 - IEEE 1722 - 2025 + """ + + name = "ACF CAN Brief V2 Header" + + fields_desc = [ + BitEnumField( + name="acf_msg_type", + default=AvtpAcfType.ACF_CAN_BRIEF_V2, + size=7, + enum={i.name: i.value for i in AvtpAcfType}, + ), + BitField(name="acf_msg_length", size=9, default=None), + BitField(name="padlength", size=2, default=0), + BitField(name="mtv", size=1, default=0), + BitField(name="rtr", size=1, default=0), + BitField(name="eff", size=1, default=0), + BitField(name="can_bus_id", size=11, default=0), + BitField(name="brs", size=1, default=0), + BitField(name="fdf", size=1, default=0), + BitField(name="esi", size=1, default=0), + BitField(name="can_identifier", size=29, default=0), + ] + + def do_dissect_payload(self, s): + return super().do_dissect_payload(s[: len(s) - self.padlength]) + + def extract_padding(self, s): + payload_length = self.acf_msg_length * 4 - len(self.self_build()) + return s[0:payload_length], s[payload_length:] + + def post_build(self, pkt: bytes, pay: bytes) -> bytes: + # Do padding + if len(pay) % 4: + pad = [0] * (4 - len(pay) % 4) + pad_byte = pkt[2] | ((4 - len(pay) % 4) << 6) + pkt = pkt[0:2] + struct.pack("!B", pad_byte) + pkt[3:] + pay += bytes(pad) + + # Take care for length updation if the acf_msg_length field is not set + if self.acf_msg_length is None: + acf_length = int((len(pkt) + len(pay)) / 4) & 0x1FF + first_byte = (pkt[0] & 0xFE) | ((acf_length >> 8) & 0x01) + pkt = ( + struct.pack("!B", first_byte) + + struct.pack("!B", acf_length & 0xFF) + + pkt[2:] + ) + + return pkt + pay + + +class AvtpAcfLinHeader(AvtpAcfHeader): + """ + Header for LIN Messages - Clause 9.4.5 - IEEE 1722 - 2025 + """ + + name = "ACF LIN Header" + fields_desc = [ + BitEnumField( + name="acf_msg_type", + default=AvtpAcfType.ACF_LIN, + size=7, + enum={i.name: i.value for i in AvtpAcfType}, + ), + BitField(name="acf_msg_length", size=9, default=None), + BitField(name="padlength", size=2, default=0), + BitField(name="mtv", size=1, default=0), + BitField(name="lin_bus_id", size=5, default=0), + XByteField(name="lin_identifier", default=0), + LongField(name="message_timestamp", default=0), + ] + + def do_dissect_payload(self, s): + return super().do_dissect_payload(s[: len(s) - self.padlength]) + + def extract_padding(self, s): + payload_length = self.acf_msg_length * 4 - len(self.self_build()) + return s[0:payload_length], s[payload_length:] + + def post_build(self, pkt: bytes, pay: bytes) -> bytes: + # Do padding + if len(pay) % 4: + pad = [0] * (4 - len(pay) % 4) + pad_byte = pkt[2] | ((4 - len(pay) % 4) << 6) + pkt = pkt[0:2] + struct.pack("!B", pad_byte) + pkt[3:] + pay += bytes(pad) + + # Take care for length updation if the acf_msg_length field is not set + if self.acf_msg_length is None: + acf_length = int((len(pkt) + len(pay)) / 4) & 0x1FF + first_byte = (pkt[0] & 0xFE) | ((acf_length >> 8) & 0x01) + pkt = ( + struct.pack("!B", first_byte) + + struct.pack("!B", acf_length & 0xFF) + + pkt[2:] + ) + + return pkt + pay + + +class AvtpAcfLinV2Header(AvtpAcfHeader): + """ + Header for LIN Messages - Clause 9.4.5 - IEEE 1722 - 2025 + """ + + name = "ACF LIN V2 Header" + fields_desc = [ + BitEnumField( + name="acf_msg_type", + default=AvtpAcfType.ACF_LIN_V2, + size=7, + enum={i.name: i.value for i in AvtpAcfType}, + ), + BitField(name="acf_msg_length", size=9, default=None), + BitField(name="padlength", size=2, default=0), + BitField(name="mtv", size=1, default=0), + BitField(name="rsv", size=2, default=0), + BitField(name="lin_bus_id", size=11, default=0), + LongField(name="message_timestamp", default=0), + StrFixedLenField("reserved", None, length=3), + XByteField(name="lin_identifier", default=0), + ] + + def do_dissect_payload(self, s): + return super().do_dissect_payload(s[: len(s) - self.padlength]) + + def extract_padding(self, s): + payload_length = self.acf_msg_length * 4 - len(self.self_build()) + return s[0:payload_length], s[payload_length:] + + def post_build(self, pkt: bytes, pay: bytes) -> bytes: + # Do padding + if len(pay) % 4: + pad = [0] * (4 - len(pay) % 4) + pad_byte = pkt[2] | ((4 - len(pay) % 4) << 6) + pkt = pkt[0:2] + struct.pack("!B", pad_byte) + pkt[3:] + pay += bytes(pad) + + # Take care for length updation if the acf_msg_length field is not set + if self.acf_msg_length is None: + acf_length = int((len(pkt) + len(pay)) / 4) & 0x1FF + first_byte = (pkt[0] & 0xFE) | ((acf_length >> 8) & 0x01) + pkt = ( + struct.pack("!B", first_byte) + + struct.pack("!B", acf_length & 0xFF) + + pkt[2:] + ) + + return pkt + pay + + +class AvtpAcfMostHeader(AvtpAcfHeader): + """ + Header for MOST Messages - Clause 9.4.6 - IEEE 1722 - 2025 + """ + + name = "ACF MOST Header" + fields_desc = [ + BitEnumField( + name="acf_msg_type", + default=AvtpAcfType.ACF_MOST, + size=7, + enum={i.name: i.value for i in AvtpAcfType}, + ), + BitField(name="acf_msg_length", size=9, default=None), + BitField(name="padlength", size=2, default=0), + BitField(name="mtv", size=1, default=0), + BitField(name="most_net_id", size=5, default=0), + ByteField(name="reserved_1", default=0), + LongField(name="message_timestamp", default=0), + XShortField(name="device_id", default=0), + XByteField(name="fblock_id", default=0), + XByteField(name="inst_id", default=0), + BitField(name="func_id", size=12, default=0), + BitField(name="op_type", size=4, default=0), + XShortField(name="reserved_2", default=0), + ] + + def do_dissect_payload(self, s): + return super().do_dissect_payload(s[: len(s) - self.padlength]) + + def extract_padding(self, s): + payload_length = self.acf_msg_length * 4 - len(self.self_build()) + return s[0:payload_length], s[payload_length:] + + def post_build(self, pkt: bytes, pay: bytes) -> bytes: + # Do padding + if len(pay) % 4: + pad = [0] * (4 - len(pay) % 4) + pad_byte = pkt[2] | ((4 - len(pay) % 4) << 6) + pkt = pkt[0:2] + struct.pack("!B", pad_byte) + pkt[3:] + pay += bytes(pad) + + # Take care for length updation if the acf_msg_length field is not set + if self.acf_msg_length is None: + acf_length = int((len(pkt) + len(pay)) / 4) & 0x1FF + first_byte = (pkt[0] & 0xFE) | ((acf_length >> 8) & 0x01) + pkt = ( + struct.pack("!B", first_byte) + + struct.pack("!B", acf_length & 0xFF) + + pkt[2:] + ) + + return pkt + pay + + +class AvtpAcfGpcHeader(AvtpAcfHeader): + """ + Header for GPC Messages - Clause 9.4.7 - IEEE 1722 - 2025 + """ + + name = "ACF GPC Header" + fields_desc = [ + BitEnumField( + name="acf_msg_type", + default=AvtpAcfType.ACF_GPC, + size=7, + enum={i.name: i.value for i in AvtpAcfType}, + ), + BitField(name="acf_msg_length", size=9, default=None), + BitField(name="gpc_msgid", size=48, default=0), + ] + + def post_build(self, pkt: bytes, pay: bytes) -> bytes: + # Do padding + if len(pay) % 4: + pad = [0] * (4 - len(pay) % 4) + pay += bytes(pad) + + # Take care for length updation if the acf_msg_length field is not set + if self.acf_msg_length is None: + acf_length = int((len(pkt) + len(pay)) / 4) & 0x1FF + first_byte = (pkt[0] & 0xFE) | ((acf_length >> 8) & 0x01) + pkt = ( + struct.pack("!B", first_byte) + + struct.pack("!B", acf_length & 0xFF) + + pkt[2:] + ) + + return pkt + pay + + +class AvtpAcfSerialHeader(AvtpAcfHeader): + """ + Header for Serial Messages - Clause 9.4.8 - IEEE 1722 - 2025 + """ + + name = "ACF Serial Header" + fields_desc = [ + BitEnumField( + name="acf_msg_type", + default=AvtpAcfType.ACF_SERIAL, + size=7, + enum={i.name: i.value for i in AvtpAcfType}, + ), + BitField(name="acf_msg_length", size=9, default=None), + BitField(name="padlength", size=2, default=0), + BitField(name="reserved", size=8, default=0), + BitField(name="dcd", size=1, default=0), + BitField(name="dtr", size=1, default=0), + BitField(name="dsr", size=1, default=0), + BitField(name="rts", size=1, default=0), + BitField(name="cts", size=1, default=0), + BitField(name="ri", size=1, default=0), + ] + + def do_dissect_payload(self, s): + return super().do_dissect_payload(s[: len(s) - self.padlength]) + + def extract_padding(self, s): + payload_length = self.acf_msg_length * 4 - len(self.self_build()) + return s[0:payload_length], s[payload_length:] + + def post_build(self, pkt: bytes, pay: bytes) -> bytes: + # Do padding + if len(pay) % 4: + pad = [0] * (4 - len(pay) % 4) + pad_byte = pkt[2] | ((4 - len(pay) % 4) << 6) + pkt = pkt[0:2] + struct.pack("!B", pad_byte) + pkt[3:] + pay += bytes(pad) + + # Take care for length updation if the acf_msg_length field is not set + if self.acf_msg_length is None: + acf_length = int((len(pkt) + len(pay)) / 4) & 0x1FF + first_byte = (pkt[0] & 0xFE) | ((acf_length >> 8) & 0x01) + pkt = ( + struct.pack("!B", first_byte) + + struct.pack("!B", acf_length & 0xFF) + + pkt[2:] + ) + + return pkt + pay + + +class AvtpAcfParallelHeader(AvtpAcfHeader): + """ + Header for Parallel Messages - Clause 9.4.9 - IEEE 1722 - 2025 + """ + + name = "ACF Parallel Header" + fields_desc = [ + BitEnumField( + name="acf_msg_type", + default=AvtpAcfType.ACF_PARALLEL, + size=7, + enum={i.name: i.value for i in AvtpAcfType}, + ), + BitField(name="acf_msg_length", size=9, default=None), + XByteField(name="reserved", default=0), + XByteField(name="bit_width", default=0), + ] + + def post_build(self, pkt: bytes, pay: bytes) -> bytes: + # Do padding + if len(pay) % 4: + pad = [0] * (4 - len(pay) % 4) + pay += bytes(pad) + + # Take care for length updation if the acf_msg_length field is not set + if self.acf_msg_length is None: + acf_length = int((len(pkt) + len(pay)) / 4) & 0x1FF + first_byte = (pkt[0] & 0xFE) | ((acf_length >> 8) & 0x01) + pkt = ( + struct.pack("!B", first_byte) + + struct.pack("!B", acf_length & 0xFF) + + pkt[2:] + ) + + return pkt + pay + + +class AvtpAcfSensorHeader(AvtpAcfHeader): + """ + Header for Sensor Messages - Clause 9.4.10 - IEEE 1722 - 2025 + """ + + name = "ACF Sensor Header" + fields_desc = [ + BitEnumField( + name="acf_msg_type", + default=AvtpAcfType.ACF_SENSOR, + size=7, + enum={i.name: i.value for i in AvtpAcfType}, + ), + BitField(name="acf_msg_length", size=9, default=None), + BitField(name="mtv", size=1, default=None), + BitField(name="num_sensors", size=7, default=None), + BitField(name="sz", size=2, default=None), + BitField(name="sensor_group", size=6, default=None), + XLongField(name="message_timestamp", default=None), + ] + + def post_build(self, pkt: bytes, pay: bytes) -> bytes: + # Do padding + if len(pay) % 4: + pad = [0] * (4 - len(pay) % 4) + pay += bytes(pad) + + # Take care for length updation if the acf_msg_length field is not set + if self.acf_msg_length is None: + acf_length = int((len(pkt) + len(pay)) / 4) & 0x1FF + first_byte = (pkt[0] & 0xFE) | ((acf_length >> 8) & 0x01) + pkt = ( + struct.pack("!B", first_byte) + + struct.pack("!B", acf_length & 0xFF) + + pkt[2:] + ) + + return pkt + pay + + +class AvtpAcfSensorBriefHeader(AvtpAcfHeader): + """ + Header for Sensor Brief Messages - Clause 9.4.11 - IEEE 1722 - 2025 + """ + + name = "ACF Sensor Brief Header" + fields_desc = [ + BitEnumField( + name="acf_msg_type", + default=AvtpAcfType.ACF_SENSOR_BRIEF, + size=7, + enum={i.name: i.value for i in AvtpAcfType}, + ), + BitField(name="acf_msg_length", size=9, default=None), + BitField(name="mtv", size=1, default=None), + BitField(name="num_sensors", size=7, default=None), + BitField(name="sz", size=2, default=None), + BitField(name="sensor_group", size=6, default=None), + ] + + def post_build(self, pkt: bytes, pay: bytes) -> bytes: + # Do padding + if len(pay) % 4: + pad = [0] * (4 - len(pay) % 4) + pay += bytes(pad) + + # Take care for length updation if the acf_msg_length field is not set + if self.acf_msg_length is None: + acf_length = int((len(pkt) + len(pay)) / 4) & 0x1FF + first_byte = (pkt[0] & 0xFE) | ((acf_length >> 8) & 0x01) + pkt = ( + struct.pack("!B", first_byte) + + struct.pack("!B", acf_length & 0xFF) + + pkt[2:] + ) + + return pkt + pay + + +class AvtpAcfAecpHeader(AvtpAcfHeader): + """ + Header for AECP Messages - Clause 9.4.12 - IEEE 1722 - 2025 + """ + + name = "ACF AECP Header" + fields_desc = [ + BitEnumField( + name="acf_msg_type", + default=AvtpAcfType.ACF_AECP, + size=7, + enum={i.name: i.value for i in AvtpAcfType}, + ), + BitField(name="acf_msg_length", size=9, default=None), + XShortField(name="reserved", default=0), + ] + + def post_build(self, pkt: bytes, pay: bytes) -> bytes: + # Do padding + if len(pay) % 4: + pad = [0] * (4 - len(pay) % 4) + pay += bytes(pad) + + # Take care for length updation if the acf_msg_length field is not set + if self.acf_msg_length is None: + acf_length = int((len(pkt) + len(pay)) / 4) & 0x1FF + first_byte = (pkt[0] & 0xFE) | ((acf_length >> 8) & 0x01) + pkt = ( + struct.pack("!B", first_byte) + + struct.pack("!B", acf_length & 0xFF) + + pkt[2:] + ) + + return pkt + pay + + +class AvtpAcfAncillaryHeader(AvtpAcfHeader): + """ + Header for Ancillary Messages - Clause 9.4.13 - IEEE 1722 - 2025 + """ + + class AncMode(Enum): + """ + Ancillary Mode - Clause 9.4.13.3.2 + """ + + ANC_8BIT = 0 + ANC_10BIT = 1 + + name = "ACF Ancillary Header" + fields_desc = [ + BitEnumField( + name="acf_msg_type", + default=AvtpAcfType.ACF_ANCILLARY, + size=7, + enum={i.name: i.value for i in AvtpAcfType}, + ), + BitField(name="acf_msg_length", size=9, default=None), + BitField(name="padlength", size=2, default=0), + BitField(name="reserved", size=10, default=0), + BitEnumField( + name="mode", + default=AncMode.ANC_8BIT, + size=2, + enum={i.name: i.value for i in AncMode}, + ), + BitField(name="fp", size=1, default=0), + BitField(name="lp", size=1, default=0), + XShortField(name="line_number", default=0), + XByteField(name="did", default=0), + XByteField(name="sdid_dbn", default=0), + ] + + def do_dissect_payload(self, s): + return super().do_dissect_payload(s[: len(s) - self.padlength]) + + def extract_padding(self, s): + payload_length = self.acf_msg_length * 4 - len(self.self_build()) + return s[0:payload_length], s[payload_length:] + + def post_build(self, pkt: bytes, pay: bytes) -> bytes: + # Do padding + if len(pay) % 4: + pad = [0] * (4 - len(pay) % 4) + pad_byte = pkt[2] | ((4 - len(pay) % 4) << 6) + pkt = pkt[0:2] + struct.pack("!B", pad_byte) + pkt[3:] + pay += bytes(pad) + + # Take care for length updation if the acf_msg_length field is not set + if self.acf_msg_length is None: + acf_length = int((len(pkt) + len(pay)) / 4) & 0x1FF + first_byte = (pkt[0] & 0xFE) | ((acf_length >> 8) & 0x01) + pkt = ( + struct.pack("!B", first_byte) + + struct.pack("!B", acf_length & 0xFF) + + pkt[2:] + ) + + return pkt + pay + + +class AvtpAcfGenericByteBusHeader(AvtpAcfHeader): + """ + Header for Generic Byte Bus Messages - Clause 9.4.14 - IEEE 1722 - 2025 + """ + + name = "ACF Generic Byte Bus Header" + fields_desc = [ + BitEnumField( + name="acf_msg_type", + default=AvtpAcfType.ACF_GBB, + size=7, + enum={i.name: i.value for i in AvtpAcfType}, + ), + BitField(name="acf_msg_length", size=9, default=None), + BitField(name="padlength", size=2, default=0), + BitField(name="mtv", size=1, default=0), + BitField(name="rsv", size=2, default=0), + BitField(name="byte_bus_id", size=11, default=0), + LongField(name="message_timestamp", default=0), + BitField(name="evt", size=4, default=0), + BitField(name="rsv2", size=2, default=0), + BitField(name="hs", size=1, default=0), + BitField(name="cs", size=1, default=0), + XByteField(name="transaction_num", default=0), + BitField(name="op", size=1, default=0), + BitField(name="rsp", size=1, default=0), + BitField(name="err", size=1, default=0), + BitField(name="ms", size=1, default=0), + BitField(name="read_size_segment_num", size=12, default=0), + ] + + def do_dissect_payload(self, s): + return super().do_dissect_payload(s[: len(s) - self.padlength]) + + def extract_padding(self, s): + payload_length = self.acf_msg_length * 4 - len(self.self_build()) + return s[0:payload_length], s[payload_length:] + + def post_build(self, pkt: bytes, pay: bytes) -> bytes: + # Do padding + if len(pay) % 4: + pad = [0] * (4 - len(pay) % 4) + pad_byte = pkt[2] | ((4 - len(pay) % 4) << 6) + pkt = pkt[0:2] + struct.pack("!B", pad_byte) + pkt[3:] + pay += bytes(pad) + + # Take care for length updation if the acf_msg_length field is not set + if self.acf_msg_length is None: + acf_length = int((len(pkt) + len(pay)) / 4) & 0x1FF + first_byte = (pkt[0] & 0xFE) | ((acf_length >> 8) & 0x01) + pkt = ( + struct.pack("!B", first_byte) + + struct.pack("!B", acf_length & 0xFF) + + pkt[2:] + ) + + return pkt + pay + + +class AvtpAcfAbbreviatedByteBusHeader(AvtpAcfHeader): + """ + Header for Abbreviated Byte Bus Messages - Clause 9.4.15 - IEEE 1722 - 2025 + """ + + name = "ACF Abbreviated Byte Bus Header" + fields_desc = [ + BitEnumField( + name="acf_msg_type", + default=AvtpAcfType.ACF_ABB, + size=7, + enum={i.name: i.value for i in AvtpAcfType}, + ), + BitField(name="acf_msg_length", size=9, default=None), + BitField(name="padlength", size=2, default=0), + BitField(name="mtv", size=1, default=0), + BitField(name="rsv", size=2, default=0), + BitField(name="byte_bus_id", size=11, default=0), + BitField(name="evt", size=4, default=0), + BitField(name="rsv2", size=2, default=0), + BitField(name="hs", size=1, default=0), + BitField(name="cs", size=1, default=0), + XByteField(name="transaction_num", default=0), + BitField(name="op", size=1, default=0), + BitField(name="rsp", size=1, default=0), + BitField(name="err", size=1, default=0), + BitField(name="ms", size=1, default=0), + BitField(name="read_size_segment_num", size=12, default=0), + ] + + def do_dissect_payload(self, s): + return super().do_dissect_payload(s[: len(s) - self.padlength]) + + def extract_padding(self, s): + payload_length = self.acf_msg_length * 4 - len(self.self_build()) + return s[0:payload_length], s[payload_length:] + + def post_build(self, pkt: bytes, pay: bytes) -> bytes: + # Do padding + if len(pay) % 4: + pad = [0] * (4 - len(pay) % 4) + pad_byte = pkt[2] | ((4 - len(pay) % 4) << 6) + pkt = pkt[0:2] + struct.pack("!B", pad_byte) + pkt[3:] + pay += bytes(pad) + + # Take care for length updation if the acf_msg_length field is not set + if self.acf_msg_length is None: + acf_length = int((len(pkt) + len(pay)) / 4) & 0x1FF + first_byte = (pkt[0] & 0xFE) | ((acf_length >> 8) & 0x01) + pkt = ( + struct.pack("!B", first_byte) + + struct.pack("!B", acf_length & 0xFF) + + pkt[2:] + ) + + return pkt + pay + + +class AvtpAcfI2CHeader(AvtpAcfHeader): + """ + Header for I2C Messages - Clause 9.4.16 - IEEE 1722 - 2025 + """ + + name = "ACF I2C Header" + fields_desc = [ + BitEnumField( + name="acf_msg_type", + default=AvtpAcfType.ACF_I2C, + size=7, + enum={i.name: i.value for i in AvtpAcfType}, + ), + BitField(name="acf_msg_length", size=9, default=None), + BitField(name="padlength", size=2, default=0), + BitField(name="mtv", size=1, default=0), + BitField(name="rsv", size=2, default=0), + BitField(name="i2c_bus_id", size=11, default=0), + LongField(name="message_timestamp", default=0), + BitField(name="i2c_code", size=4, default=0), + BitField(name="trr", size=1, default=0), + BitField(name="reserved", size=3, default=0), + XByteField(name="transaction_num", default=0), + BitField(name="evt", size=4, default=0), + BitField(name="exception_code", size=4, default=0), + ] + + def post_build(self, pkt: bytes, pay: bytes) -> bytes: + + # A correct I2C message has a 1-byte payload. + # If the payload is not 1 byte, we will pad it to 1 byte. + + # Take care for length updation if the acf_msg_length field is not set + if self.acf_msg_length is None: + acf_length = int((len(pkt) + len(pay)) / 4) & 0x1FF + first_byte = (pkt[0] & 0xFE) | ((acf_length >> 8) & 0x01) + pkt = ( + struct.pack("!B", first_byte) + + struct.pack("!B", acf_length & 0xFF) + + pkt[2:] + ) + + return pkt + pay + + +class AvtpAcfI2CBriefHeader(AvtpAcfHeader): + """ + Header for I2C Brief Messages - Clause 9.4.17 - IEEE 1722 - 2025 + """ + + name = "ACF I2C Brief Header" + fields_desc = [ + BitEnumField( + name="acf_msg_type", + default=AvtpAcfType.ACF_I2C_BRIEF, + size=7, + enum={i.name: i.value for i in AvtpAcfType}, + ), + BitField(name="acf_msg_length", size=9, default=None), + BitField(name="padlength", size=2, default=0), + BitField(name="mtv", size=1, default=0), + BitField(name="rsv", size=2, default=0), + BitField(name="i2c_bus_id", size=11, default=0), + BitField(name="i2c_code", size=4, default=0), + BitField(name="trr", size=1, default=0), + BitField(name="reserved", size=3, default=0), + XByteField(name="transaction_num", default=0), + BitField(name="evt", size=4, default=0), + BitField(name="exception_code", size=4, default=0), + ] + + def post_build(self, pkt: bytes, pay: bytes) -> bytes: + + # A correct I2C brief message has a 1-byte payload. + # If the payload is not 1 byte, we will pad it to 1 byte. + + # Take care for length updation if the acf_msg_length field is not set + if self.acf_msg_length is None: + acf_length = int((len(pkt) + len(pay)) / 4) & 0x1FF + first_byte = (pkt[0] & 0xFE) | ((acf_length >> 8) & 0x01) + pkt = ( + struct.pack("!B", first_byte) + + struct.pack("!B", acf_length & 0xFF) + + pkt[2:] + ) + + return pkt + pay + + +class AvtpAcfCanXlHeader(AvtpAcfHeader): + """ + Header for CAN XL Messages - Clause 9.4.18 - IEEE 1722 - 2025 + """ + + name = "ACF CAN XL Header" + fields_desc = [ + BitEnumField( + name="acf_msg_type", + default=AvtpAcfType.ACF_CAN_XL, + size=7, + enum={i.name: i.value for i in AvtpAcfType}, + ), + BitField(name="acf_msg_length", size=9, default=None), + BitField(name="padlength", size=2, default=0), + BitField(name="mtv", size=1, default=0), + BitField(name="rsv", size=2, default=0), + BitField(name="can_bus_id", size=11, default=0), + LongField(name="message_timestamp", default=0), + XByteField(name="vcid", default=0), + XByteField(name="sdt", default=0), + BitField(name="reserved", size=3, default=0), + BitField(name="rrs", size=1, default=0), + BitField(name="sec", size=1, default=0), + BitField(name="priority_id", size=11, default=0), + XIntField(name="acceptance_field", default=0), + ByteField(name="reserved_2", default=0), + XByteField(name="transaction_num", default=0), + BitField(name="reserved_3", size=3, default=0), + BitField(name="ms", size=1, default=0), + BitField(name="segment_num", size=12, default=0), + ] + + def do_dissect_payload(self, s): + return super().do_dissect_payload(s[: len(s) - self.padlength]) + + def extract_padding(self, s): + """Extract padding from the payload.""" + payload_length = self.acf_msg_length * 4 - len(self.self_build()) + return s[0:payload_length], s[payload_length:] + + def post_build(self, pkt: bytes, pay: bytes) -> bytes: + # Do padding + if len(pay) % 4: + pad = [0] * (4 - len(pay) % 4) + pad_byte = pkt[2] | ((4 - len(pay) % 4) << 6) + pkt = pkt[0:2] + struct.pack("!B", pad_byte) + pkt[3:] + pay += bytes(pad) + + # Take care for length updation if the acf_msg_length field is not set + if self.acf_msg_length is None: + acf_length = int((len(pkt) + len(pay)) / 4) & 0x1FF + first_byte = (pkt[0] & 0xFE) | ((acf_length >> 8) & 0x01) + pkt = ( + struct.pack("!B", first_byte) + + struct.pack("!B", acf_length & 0xFF) + + pkt[2:] + ) + + return pkt + pay + + +class AvtpAcfCanXlBriefHeader(AvtpAcfHeader): + """ + Header for CAN XL Brief Messages - Clause 9.4.19 - IEEE 1722 - 2025 + """ + + name = "ACF CAN XL Brief Header" + fields_desc = [ + BitEnumField( + name="acf_msg_type", + default=AvtpAcfType.ACF_CAN_XL_BRIEF, + size=7, + enum={i.name: i.value for i in AvtpAcfType}, + ), + BitField(name="acf_msg_length", size=9, default=None), + BitField(name="padlength", size=2, default=0), + BitField(name="mtv", size=1, default=0), + BitField(name="rsv", size=2, default=0), + BitField(name="can_bus_id", size=11, default=0), + XByteField(name="vcid", default=0), + XByteField(name="sdt", default=0), + BitField(name="reserved", size=3, default=0), + BitField(name="rrs", size=1, default=0), + BitField(name="sec", size=1, default=0), + BitField(name="priority_id", size=11, default=0), + XIntField(name="acceptance_field", default=0), + ByteField(name="reserved_2", default=0), + XByteField(name="transaction_num", default=0), + BitField(name="reserved_3", size=3, default=0), + BitField(name="ms", size=1, default=0), + BitField(name="segment_num", size=12, default=0), + ] + + def do_dissect_payload(self, s): + return super().do_dissect_payload(s[: len(s) - self.padlength]) + + def extract_padding(self, s): + """Extract padding from the payload.""" + payload_length = self.acf_msg_length * 4 - len(self.self_build()) + return s[0:payload_length], s[payload_length:] + + def post_build(self, pkt: bytes, pay: bytes) -> bytes: + # Do padding + if len(pay) % 4: + pad = [0] * (4 - len(pay) % 4) + pad_byte = pkt[2] | ((4 - len(pay) % 4) << 6) + pkt = pkt[0:2] + struct.pack("!B", pad_byte) + pkt[3:] + pay += bytes(pad) + + # Take care for length updation if the acf_msg_length field is not set + if self.acf_msg_length is None: + acf_length = int((len(pkt) + len(pay)) / 4) & 0x1FF + first_byte = (pkt[0] & 0xFE) | ((acf_length >> 8) & 0x01) + pkt = ( + struct.pack("!B", first_byte) + + struct.pack("!B", acf_length & 0xFF) + + pkt[2:] + ) + + return pkt + pay + + +class AvtpAcfChecksumHeader(AvtpAcfHeader): + """ + Header for Checksum Messages - Clause 9.4.20 - IEEE 1722 - 2025 + """ + + name = "ACF Checksum Header" + fields_desc = [ + BitEnumField( + name="acf_msg_type", + default=AvtpAcfType.ACF_CHECKSUM, + size=7, + enum={i.name: i.value for i in AvtpAcfType}, + ), + BitField(name="acf_msg_length", size=9, default=1), + XShortField(name="checksum", default=0), + ] + + +class AvtpAcfCrcHeader(AvtpAcfHeader): + """ + Header for CRC Messages - Clause 9.4.21 - IEEE 1722 - 2025 + """ + + name = "ACF CRC Header" + fields_desc = [ + BitEnumField( + name="acf_msg_type", + default=AvtpAcfType.ACF_CRC, + size=7, + enum={i.name: i.value for i in AvtpAcfType}, + ), + BitField(name="acf_msg_length", size=9, default=None), + BitField(name="reserved", size=12, default=0), + BitField(name="crc_type", size=4, default=0), + ] + + def post_build(self, pkt: bytes, pay: bytes) -> bytes: + # Do padding + if len(pay) % 4: + pad = [0] * (4 - len(pay) % 4) + pay += bytes(pad) + + # Take care for length updation if the acf_msg_length field is not set + if self.acf_msg_length is None: + acf_length = int((len(pkt) + len(pay)) / 4) & 0x1FF + first_byte = (pkt[0] & 0xFE) | ((acf_length >> 8) & 0x01) + pkt = ( + struct.pack("!B", first_byte) + + struct.pack("!B", acf_length & 0xFF) + + pkt[2:] + ) + + return pkt + pay + + +class AvtpAcfGisfHeader(AvtpAcfHeader): + """ + Header for GISF Messages - Clause 18 - IEEE 1722 - 2025 + """ + + name = "ACF GISF Header" + fields_desc = [ + BitEnumField( + name="acf_msg_type", + default=AvtpAcfType.ACF_GISF, + size=7, + enum={i.name: i.value for i in AvtpAcfType}, + ), + BitField(name="acf_msg_length", size=9, default=None), + BitField(name="padlength", size=2, default=0), + BitField(name="mtv", size=1, default=0), + BitField(name="rsv", size=2, default=0), + BitField(name="image_sensor_id", size=11, default=0), + LongField(name="message_timestamp", default=0), + ShortField(name="reserved", default=0), + BitField(name="r", size=1, default=0), + BitField(name="el", size=1, default=0), + BitField(name="t1", size=1, default=0), + BitField(name="ef", size=1, default=0), + BitField(name="evt", size=4, default=0), + BitField(name="rsv2", size=2, default=0), + BitField(name="bf", size=1, default=0), + BitField(name="line_type_id", size=5, default=0), + XByteField(name="evt2", default=0), + XByteField(name="i_seq_num", default=0), + XShortField(name="line_number", default=0), + ] + + def do_dissect_payload(self, s): + return super().do_dissect_payload(s[: len(s) - self.padlength]) + + def extract_padding(self, s): + """Extract padding from the payload.""" + payload_length = self.acf_msg_length * 4 - len(self.self_build()) + return s[0:payload_length], s[payload_length:] + + def post_build(self, pkt: bytes, pay: bytes) -> bytes: + # Do padding + if len(pay) % 4: + pad = [0] * (4 - len(pay) % 4) + pad_byte = pkt[2] | ((4 - len(pay) % 4) << 6) + pkt = pkt[0:2] + struct.pack("!B", pad_byte) + pkt[3:] + pay += bytes(pad) + + # Take care for length updation if the acf_msg_length field is not set + if self.acf_msg_length is None: + acf_length = int((len(pkt) + len(pay)) / 4) & 0x1FF + first_byte = (pkt[0] & 0xFE) | ((acf_length >> 8) & 0x01) + pkt = ( + struct.pack("!B", first_byte) + + struct.pack("!B", acf_length & 0xFF) + + pkt[2:] + ) + + return pkt + pay + + +class _AvtpNtscfHeaderV0(_AlternativeHeaderV0): + name = "AVTP NTSCF Header v0" + fields_desc = [ + ConditionalField( + IntField(name="encapsulation_sequence_num", default=0), + lambda pkt: isinstance(pkt.underlayer, UDP), + ), + XByteEnumField( + name="subtype", + default=AvtpStreamType.NTSCF, + enum={i.name: i.value for i in AvtpStreamType}, + ), + BitField(name="h", size=1, default=0), + BitField(name="version", size=3, default=0), + BitField(name="r", size=1, default=0), + BitFieldLenField( + name="ntscf_data_length", size=11, default=None, length_of="acf_tlv" + ), + XByteField(name="sequence_num", default=0), + XLongField(name="stream_id", default=0), + PacketListField(name="acf_tlv", default=[], pkt_cls=AvtpAcfHeader), + ] + + +class _AvtpNtscfHeaderV1(_AlternativeHeaderV1): + name = "AVTP NTSCF Header v1" + fields_desc = [ + ConditionalField( + IntField(name="encapsulation_sequence_num", default=0), + lambda pkt: isinstance(pkt.underlayer, UDP), + ), + XByteEnumField( + name="subtype", + default=AvtpStreamType.NTSCF, + enum={i.name: i.value for i in AvtpStreamType}, + ), + BitField(name="h", size=1, default=0), + BitField(name="version", size=3, default=1), + BitField(name="reserved_1", size=20, default=0), + XIntField(name="sequence_num", default=0), + XLongField(name="gptp_grandmaster_identity", default=0), + BitField(name="reserved_2", size=12, default=0), + BitField(name="r", size=1, default=0), + BitFieldLenField( + name="ntscf_data_length", size=11, default=None, length_of="acf_tlv" + ), + XByteField(name="sequence_num_lsb", default=0), + XLongField(name="stream_id", default=0), + PacketListField(name="acf_tlv", default=[], pkt_cls=AvtpAcfHeader), + ] + + +class AvtpNtscfHeader(AvtpCommonHeader): + """ + Header for Non-Time-Synchronous Control Format - Clause 9.2 - IEEE 1722 - 2025 + """ + + name = "AVTP NTSCF Header" + + @classmethod + def dispatch_hook(cls, pkt=None, **kargs): + version = 0 + if "version" in kargs: + version = kargs.get("version", 0) == 1 + else: + if pkt is not None: + # We do not check for the underlayer here. + # This is only if the packet is being dissected from + # raw bytes using the AvtpNtscfHeader class directly. + parsed_version = (ord(pkt[1:2]) & 0x70) >> 4 + version = parsed_version == 1 + return _AvtpNtscfHeaderV1 if version else _AvtpNtscfHeaderV0 + + +class _AvtpTscfHeaderV0(_CommonStreamHeaderV0): + name = "AVTP TSCF Header v0" + fields_desc = [ + ConditionalField( + IntField(name="encapsulation_sequence_num", default=0), + lambda pkt: isinstance(pkt.underlayer, UDP), + ), + XByteEnumField( + name="subtype", + default=AvtpStreamType.TSCF, + enum={i.name: i.value for i in AvtpStreamType}, + ), + BitField(name="sv", size=1, default=0), + BitField(name="version", size=3, default=0), + BitField(name="mr", size=1, default=0), + BitField(name="rsv", size=2, default=0), + BitField(name="tv", size=1, default=0), + XByteField(name="sequence_num", default=0), + BitField(name="reserved_1", size=7, default=0), + BitField(name="tu", size=1, default=0), + XLongField(name="stream_id", default=0), + XIntField(name="avtp_timestamp", default=0), + XIntField(name="reserved_2", default=0), + FieldLenField(name="stream_data_length", default=None, length_of="acf_tlv"), + XShortField(name="reserved_3", default=0), + PacketListField(name="acf_tlv", default=[], pkt_cls=AvtpAcfHeader), + ] + + +class _AvtpTscfHeaderV1(_CommonStreamHeaderV1): + name = "AVTP TSCF Header v1" + fields_desc = [ + ConditionalField( + IntField(name="encapsulation_sequence_num", default=0), + lambda pkt: isinstance(pkt.underlayer, UDP), + ), + XByteEnumField( + name="subtype", + default=AvtpStreamType.TSCF, + enum={i.name: i.value for i in AvtpStreamType}, + ), + BitField(name="sv", size=1, default=0), + BitField(name="version", size=3, default=1), + BitField(name="mr", size=1, default=0), + BitField(name="rsv", size=2, default=0), + BitField(name="tv", size=1, default=0), + XByteField(name="sequence_num_lsb", default=0), + BitField(name="reserved_1", size=7, default=0), + BitField(name="tu", size=1, default=0), + XLongField(name="stream_id", default=0), + XIntField(name="sequence_num", default=0), + XLongField(name="avtp_timestamp", default=0), + XLongField(name="ptp_grandmaster_identity", default=0), + XIntField(name="reserved_2", default=0), + FieldLenField(name="stream_data_length", default=None, length_of="acf_tlv"), + XShortField(name="reserved_3", default=0), + PacketListField(name="acf_tlv", default=[], pkt_cls=AvtpAcfHeader), + ] + + +class AvtpTscfHeader(AvtpCommonStreamHeader): + """ + Header for Time-Synchronous Control Format - Clause 9.3 - IEEE 1722 - 2025 + """ + + name = "AVTP TSCF Header" + + @classmethod + def dispatch_hook(cls, pkt=None, **kargs): + version = 0 + if "version" in kargs: + version = kargs.get("version", 0) == 1 + else: + if pkt is not None: + # We do not check for the underlayer here. + # This is only if the packet is being dissected from + # raw bytes using the AvtpTscfHeader class directly. + parsed_version = (ord(pkt[1:2]) & 0x70) >> 4 + version = parsed_version == 1 + return _AvtpTscfHeaderV1 if version else _AvtpTscfHeaderV0 + + +bind_layers(Ether, AvtpCommonHeader, type=0x22F0) +bind_layers(UDP, AvtpCommonHeader, dport=17220) diff --git a/test/contrib/ieee1722.uts b/test/contrib/ieee1722.uts new file mode 100644 index 00000000000..1034d6e20f7 --- /dev/null +++ b/test/contrib/ieee1722.uts @@ -0,0 +1,390 @@ +% IEEE 1722 test campaign + +# SPDX-License-Identifier: GPL-2.0-only +# This file is part of Scapy +# See https://scapy.net/ for more information +# Copyright (C) Naresh Nayak + ++ Load tests += Load module +load_contrib("ieee1722", globals_dict=globals()) +AvtpAcfFlexrayHeader() +assert AvtpAcfHeader.acf_variants[0] is AvtpAcfFlexrayHeader + ++ AvtpAcfCan Tests += Build packet +acf_pkt = AvtpAcfCanHeader() +assert acf_pkt.acf_msg_type == 1 +assert hasattr(acf_pkt, "acf_msg_length") +assert acf_pkt.acf_msg_length is None + += Set invalid length +acf_pkt = AvtpAcfCanHeader(acf_msg_length= 0x1, can_bus_id=0x14, can_identifier=0x45) / Raw(load="\x01\x02\x03") +pktstr = bytes(acf_pkt) +binstr = b"\x02\x01\x40\x14\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x45\x01\x02\x03\x00" +assert pktstr == binstr +assert acf_pkt.acf_msg_type == 1 +assert acf_pkt.acf_msg_length == 1 +assert acf_pkt.can_identifier == 0x45 +assert acf_pkt.can_bus_id == 0x14 + += Dont set any length +acf_pkt = AvtpAcfCanHeader(can_identifier=0x45) / Raw(load="\x01\x02\x03") +pktstr = bytes(acf_pkt) +binstr = b"\x02\x05\x40\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x45\x01\x02\x03\x00" +assert pktstr == binstr +acf_pkt_recv = AvtpAcfCanHeader(binstr) +assert acf_pkt_recv.acf_msg_type == 1 +assert acf_pkt_recv.acf_msg_length == 5 +assert acf_pkt_recv.padlength == 1 + += Encapsulations with version 0 +acf_can_pkt = AvtpAcfCanHeader() / Raw( + load="\x01\x02\x03" +) +cf_pkt = AvtpNtscfHeader(acf_tlv=[acf_can_pkt]) +pkt = Ether()/cf_pkt +assert pkt.encapsulation_sequence_num == None +assert len(pkt.layers()) == 2 +assert type(pkt.layers()[0]()) == Ether +assert type(pkt.layers()[1]()) == type(AvtpNtscfHeader()) + +pkt = Ether()/IP()/UDP()/cf_pkt +assert pkt.encapsulation_sequence_num == 0 +assert len(pkt.layers()) == 4 +assert pkt.layers()[0] == Ether +assert pkt.layers()[1] == IP +assert pkt.layers()[2] == UDP +assert type(pkt.layers()[3]()) == type(AvtpNtscfHeader()) + += Encapsulations with version 1 +acf_can_pkt = AvtpAcfCanHeader() / Raw( + load="\x01\x02\x03" +) +cf_pkt = AvtpNtscfHeader(version=1, acf_tlv=[acf_can_pkt]) +pkt = Ether()/cf_pkt +assert pkt.encapsulation_sequence_num == None +assert len(pkt.layers()) == 2 +assert type(pkt.layers()[0]()) == Ether +assert type(pkt.layers()[1]()) == type(AvtpNtscfHeader(version=1)) + ++ Common header tests += Stream header versions +stream_v0 = AvtpCommonStreamHeader(version=0) +stream_v1 = AvtpCommonStreamHeader(version=1) +assert type(stream_v0).__name__ == "_CommonStreamHeaderV0" +assert type(stream_v1).__name__ == "_CommonStreamHeaderV1" +assert stream_v0.version == 0 +assert stream_v1.version == 1 +assert len(stream_v0.fields_desc) == 12 +assert len(stream_v1.fields_desc) == 17 + += Control and alternative header versions +control = AvtpCommonControlHeader( + subtype=AvtpStreamType.AECP.value, stream_id=0x0102030405060708 +) / Raw(load="abc") +assert control.control_data_length == None +assert control.subtype == AvtpStreamType.AECP.value +assert control.stream_id == 0x0102030405060708 +alternative_v0 = AvtpAlternativeHeader(version=0) +alternative_v1 = AvtpAlternativeHeader(version=1) +assert type(alternative_v0).__name__ == "_AlternativeHeaderV0" +assert type(alternative_v1).__name__ == "_AlternativeHeaderV1" +assert len(alternative_v0.fields_desc) == 4 +assert len(alternative_v1.fields_desc) == 8 + ++ ACF header tests += ACF variant defaults +acf_headers = [ + (AvtpAcfFlexrayHeader, AvtpAcfType.ACF_FLEXRAY.value), + (AvtpAcfCanHeader, AvtpAcfType.ACF_CAN.value), + (AvtpAcfCanV2Header, AvtpAcfType.ACF_CAN_V2.value), + (AvtpAcfCanBriefHeader, AvtpAcfType.ACF_CAN_BRIEF.value), + (AvtpAcfCanBriefV2Header, AvtpAcfType.ACF_CAN_BRIEF_V2.value), + (AvtpAcfLinHeader, AvtpAcfType.ACF_LIN.value), + (AvtpAcfLinV2Header, AvtpAcfType.ACF_LIN_V2.value), + (AvtpAcfMostHeader, AvtpAcfType.ACF_MOST.value), + (AvtpAcfGpcHeader, AvtpAcfType.ACF_GPC.value), + (AvtpAcfSerialHeader, AvtpAcfType.ACF_SERIAL.value), + (AvtpAcfParallelHeader, AvtpAcfType.ACF_PARALLEL.value), +] +for acf_class, acf_type in acf_headers: + acf_pkt = acf_class() + assert acf_pkt.acf_msg_type == acf_type + assert acf_pkt.acf_msg_length is None + += ACF payload lengths and padding +can_v2 = AvtpAcfCanV2Header( + can_bus_id=0x123, can_identifier=0x12345, message_timestamp=0x10203040 +) / Raw(load="abc") +can_v2_bytes = bytes(can_v2) +assert len(can_v2_bytes) == 20 +can_v2_decoded = AvtpAcfCanV2Header(can_v2_bytes) +assert can_v2_decoded.acf_msg_length == 5 +assert can_v2_decoded.padlength == 1 + +lin_v2 = AvtpAcfLinV2Header( + lin_bus_id=0x123, lin_identifier=0x22, reserved=b"abc" +) / Raw(load="abc") +assert len(bytes(lin_v2)) == 20 +lin_v2_decoded = AvtpAcfLinV2Header(bytes(lin_v2)) +assert lin_v2_decoded.acf_msg_length == 5 +assert lin_v2_decoded.padlength == 1 + +gpc = AvtpAcfGpcHeader(gpc_msgid=0x010203040506) / Raw(load="abc") +gpc_decoded = AvtpAcfGpcHeader(bytes(gpc)) +assert len(bytes(gpc)) == 12 +assert gpc_decoded.acf_msg_length == 3 + +parallel = AvtpAcfParallelHeader(bit_width=8) / Raw(load="abc") +parallel_decoded = AvtpAcfParallelHeader(bytes(parallel)) +assert len(bytes(parallel)) == 8 +assert parallel_decoded.acf_msg_length == 2 + += ACF message lengths beyond 8 bits +can_length_256 = AvtpAcfCanHeader() / Raw(load=b"x" * (256 * 4 - 16)) +can_length_256_bytes = bytes(can_length_256) +assert len(can_length_256_bytes) == 256 * 4 +assert can_length_256_bytes[:2] == b"\x03\x00" +can_length_256_decoded = AvtpAcfCanHeader(can_length_256_bytes) +assert can_length_256_decoded.acf_msg_length == 256 + +can_length_257 = AvtpAcfCanHeader() / Raw(load=b"x" * (257 * 4 - 16)) +can_length_257_bytes = bytes(can_length_257) +assert len(can_length_257_bytes) == 257 * 4 +assert can_length_257_bytes[:2] == b"\x03\x01" +can_length_257_decoded = AvtpAcfCanHeader(can_length_257_bytes) +assert can_length_257_decoded.acf_msg_length == 257 + += ACF field values +flexray = AvtpAcfFlexrayHeader( + fr_bus_id=3, fr_frame_id=0x123, cycle=0x12 +) +assert flexray.fr_bus_id == 3 +assert flexray.fr_frame_id == 0x123 +assert flexray.cycle == 0x12 + +can_brief = AvtpAcfCanBriefHeader( + can_bus_id=4, can_identifier=0x12345, brs=1, fdf=1 +) +assert can_brief.can_bus_id == 4 +assert can_brief.can_identifier == 0x12345 +assert can_brief.brs == 1 +assert can_brief.fdf == 1 + +most = AvtpAcfMostHeader( + most_net_id=2, device_id=0x1234, fblock_id=3, inst_id=4, func_id=5 +) +assert most.most_net_id == 2 +assert most.device_id == 0x1234 +assert most.fblock_id == 3 +assert most.inst_id == 4 +assert most.func_id == 5 + +serial = AvtpAcfSerialHeader(dcd=1, dtr=1, dsr=1, rts=1, cts=1, ri=1) +assert serial.dcd == 1 +assert serial.dtr == 1 +assert serial.dsr == 1 +assert serial.rts == 1 +assert serial.cts == 1 +assert serial.ri == 1 + += ACF Serial header serialization +serial_pkt = AvtpAcfSerialHeader(dcd=1, dtr=1, dsr=1, rts=1, cts=1, ri=1) +serial_bytes = bytes(serial_pkt) +serial_decoded = AvtpAcfSerialHeader(serial_bytes) +assert serial_decoded.dcd == 1 +assert serial_decoded.dtr == 1 +assert serial_decoded.dsr == 1 +assert serial_decoded.rts == 1 +assert serial_decoded.cts == 1 +assert serial_decoded.ri == 1 + +serial_pkt_with_payload = AvtpAcfSerialHeader() / Raw(load="abc") +serial_with_payload_bytes = bytes(serial_pkt_with_payload) +serial_with_payload_decoded = AvtpAcfSerialHeader(serial_with_payload_bytes) +assert bytes(serial_with_payload_decoded.payload) == b"abc" + += ACF do_dissect_payload drops payload when padlength is 0 +padlength_zero_classes = [ + AvtpAcfFlexrayHeader, + AvtpAcfCanHeader, + AvtpAcfCanV2Header, + AvtpAcfCanBriefHeader, + AvtpAcfCanBriefV2Header, + AvtpAcfLinHeader, + AvtpAcfLinV2Header, + AvtpAcfMostHeader, + AvtpAcfSerialHeader, +] + +for acf_class in padlength_zero_classes: + built = acf_class() / Raw(load=b"WXYZ") + decoded = acf_class(bytes(built)) + assert decoded.padlength == 0 + assert bytes(decoded.payload) == b"WXYZ" + +for acf_class in padlength_zero_classes: + built = acf_class() / Raw(load=b"WXY") + decoded = acf_class(bytes(built)) + assert decoded.padlength == 1 + assert bytes(decoded.payload) == b"WXY" + +for acf_class in padlength_zero_classes: + built = acf_class() / Raw(load=b"WX") + decoded = acf_class(bytes(built)) + assert decoded.padlength == 2 + assert bytes(decoded.payload) == b"WX" + +for acf_class in padlength_zero_classes: + built = acf_class() / Raw(load=b"W") + decoded = acf_class(bytes(built)) + assert decoded.padlength == 3 + assert bytes(decoded.payload) == b"W" + += Versioned NTSCF and TSCF headers +acf_tlv = [AvtpAcfCanHeader(can_identifier=0x123) / Raw(load="abc")] +ntscf_v0 = AvtpNtscfHeader(version=0, acf_tlv=acf_tlv) +ntscf_v1 = AvtpNtscfHeader(version=1, acf_tlv=acf_tlv) +assert ntscf_v0.subtype == AvtpStreamType.NTSCF.value +assert ntscf_v1.subtype == AvtpStreamType.NTSCF.value +assert len(bytes(ntscf_v0)) == 32 +assert len(bytes(ntscf_v1)) == 48 +assert len(ntscf_v0.acf_tlv) == 1 +assert len(ntscf_v1.acf_tlv) == 1 + +tscf_v0 = AvtpTscfHeader(version=0, acf_tlv=acf_tlv) +tscf_v1 = AvtpTscfHeader(version=1, acf_tlv=acf_tlv) +assert tscf_v0.subtype == AvtpStreamType.TSCF.value +assert tscf_v1.subtype == AvtpStreamType.TSCF.value +assert len(bytes(tscf_v0)) == 44 +assert len(bytes(tscf_v1)) == 60 +assert len(tscf_v0.acf_tlv) == 1 +assert len(tscf_v1.acf_tlv) == 1 + += Additional ACF class defaults +acf_headers += [ + (AvtpAcfSensorHeader, AvtpAcfType.ACF_SENSOR.value), + (AvtpAcfSensorBriefHeader, AvtpAcfType.ACF_SENSOR_BRIEF.value), + (AvtpAcfAecpHeader, AvtpAcfType.ACF_AECP.value), + (AvtpAcfAncillaryHeader, AvtpAcfType.ACF_ANCILLARY.value), + (AvtpAcfGenericByteBusHeader, AvtpAcfType.ACF_GBB.value), + (AvtpAcfAbbreviatedByteBusHeader, AvtpAcfType.ACF_ABB.value), + (AvtpAcfCanXlHeader, AvtpAcfType.ACF_CAN_XL.value), + (AvtpAcfCanXlBriefHeader, AvtpAcfType.ACF_CAN_XL_BRIEF.value), + (AvtpAcfCrcHeader, AvtpAcfType.ACF_CRC.value), + (AvtpAcfI2CHeader, AvtpAcfType.ACF_I2C.value), + (AvtpAcfI2CBriefHeader, AvtpAcfType.ACF_I2C_BRIEF.value), + (AvtpAcfGisfHeader, AvtpAcfType.ACF_GISF.value) +] +for acf_class, acf_type in acf_headers: + acf_pkt = acf_class() + assert acf_pkt.acf_msg_type == acf_type + assert acf_pkt.acf_msg_length is None + += ACF dispatch variants +for acf_class in [ + AvtpAcfFlexrayHeader, + AvtpAcfCanHeader, + AvtpAcfCanBriefHeader, + AvtpAcfLinHeader, + AvtpAcfMostHeader, + AvtpAcfGpcHeader, + AvtpAcfSerialHeader, + AvtpAcfParallelHeader, + AvtpAcfSensorHeader, + AvtpAcfSensorBriefHeader, + AvtpAcfAecpHeader, + AvtpAcfAncillaryHeader, + AvtpAcfGenericByteBusHeader, + AvtpAcfAbbreviatedByteBusHeader, + AvtpAcfI2CHeader, + AvtpAcfI2CBriefHeader, + AvtpAcfCanXlHeader, + AvtpAcfCanXlBriefHeader, + AvtpAcfGisfHeader +]: + acf_pkt = acf_class() + decoded = AvtpAcfHeader(bytes(acf_pkt)) + assert type(decoded) is acf_class + += ACF Message Length defaults for fixed length padlength_zero_classes +acf_pkt = AvtpAcfChecksumHeader() +assert acf_pkt.acf_msg_type == AvtpAcfType.ACF_CHECKSUM.value +assert acf_pkt.acf_msg_length == 1 + += Additional payload handling tests +can_xl = AvtpAcfCanXlHeader( + can_bus_id=0x123, priority_id=0x456, acceptance_field=0x12345678 +) / Raw(load="abc") +can_xl_decoded = AvtpAcfCanXlHeader(bytes(can_xl)) +assert len(bytes(can_xl)) == 28 +assert can_xl_decoded.acf_msg_length == 7 +assert can_xl_decoded.padlength == 1 + +can_xl_brief = AvtpAcfCanXlBriefHeader( + can_bus_id=0x123, priority_id=0x456, acceptance_field=0x12345678 +) / Raw(load="abc") +can_xl_brief_decoded = AvtpAcfCanXlBriefHeader(bytes(can_xl_brief)) +assert len(bytes(can_xl_brief)) == 20 +assert can_xl_brief_decoded.acf_msg_length == 5 +assert can_xl_brief_decoded.padlength == 1 + +checksum = AvtpAcfChecksumHeader(checksum=0xABCD) +assert checksum.checksum == 0xABCD +assert checksum.acf_msg_type == AvtpAcfType.ACF_CHECKSUM.value + +crc = AvtpAcfCrcHeader(crc_type=0x3) +assert crc.crc_type == 0x3 +assert crc.acf_msg_type == AvtpAcfType.ACF_CRC.value + +gisf = AvtpAcfGisfHeader( + image_sensor_id=0x123, line_number=0x456 +) / Raw(load="abc") +gisf_decoded = AvtpAcfGisfHeader(bytes(gisf)) +assert len(bytes(gisf)) == 24 +assert gisf_decoded.acf_msg_type == AvtpAcfType.ACF_GISF.value +assert gisf_decoded.image_sensor_id == 0x123 +assert gisf_decoded.line_number == 0x456 +assert gisf_decoded.acf_msg_length == 6 +assert gisf_decoded.padlength == 1 +assert bytes(gisf_decoded.payload) == b"abc" + += Ethernet and UDP bindings +ether_pkt = Ether(type=0x22F0) / AvtpTscfHeader() +assert type(ether_pkt.payload).__name__ == "_AvtpTscfHeaderV0" +ether_pkt = Ether(type=0x22F0) / AvtpTscfHeader(version=1) +assert type(ether_pkt.payload).__name__ == "_AvtpTscfHeaderV1" +ether_pkt_bytes = bytes(ether_pkt) +ether_pkt_recd = Ether(ether_pkt_bytes) +assert type(ether_pkt_recd.payload).__name__ == "_AvtpTscfHeaderV1" + +ether_pkt = Ether(type=0x22F0) / AvtpNtscfHeader() +assert type(ether_pkt.payload).__name__ == "_AvtpNtscfHeaderV0" +ether_pkt = Ether(type=0x22F0) / AvtpNtscfHeader(version=1) +assert type(ether_pkt.payload).__name__ == "_AvtpNtscfHeaderV1" +ether_pkt_bytes = bytes(ether_pkt) +ether_pkt_recd = Ether(ether_pkt_bytes) +assert type(ether_pkt_recd.payload).__name__ == "_AvtpNtscfHeaderV1" + +udp_pkt = IP() / UDP(dport=17220) / AvtpNtscfHeader() +assert udp_pkt.payload.payload.encapsulation_sequence_num == 0 +assert type(udp_pkt.payload.payload).__name__ == "_AvtpNtscfHeaderV0" +udp_pkt = IP() / UDP(dport=17220) / AvtpNtscfHeader(version=1) +assert udp_pkt.payload.payload.encapsulation_sequence_num == 0 +assert type(udp_pkt.payload.payload).__name__ == "_AvtpNtscfHeaderV1" + += TSCF & NTSCF parsing +wire = bytes(AvtpNtscfHeader(version=1)) +decoded = AvtpNtscfHeader(wire) +assert type(decoded).__name__ == "_AvtpNtscfHeaderV1" +wire = bytes(AvtpNtscfHeader()) +decoded = AvtpNtscfHeader(wire) +assert type(decoded).__name__ == "_AvtpNtscfHeaderV0" + +wire = bytes(AvtpTscfHeader(version=1)) +decoded = AvtpTscfHeader(wire) +assert type(decoded).__name__ == "_AvtpTscfHeaderV1" +wire = bytes(AvtpTscfHeader()) +decoded = AvtpTscfHeader(wire) +assert type(decoded).__name__ == "_AvtpTscfHeaderV0"