ieee1722: Initial implementation of parts of IEEE 1722. by nayakned · Pull Request #5185 · secdev/scapy

nayakned · 2026-09-18T08:48:11Z

AI Assisted: no for the implementation/copilot for unit tests.

Implemented the Chapter 9 of the IEEE 1722-2025 spec.
Added unit test cases for dissecting and generating packets belonging to this protocol

Checklist :

Check the contribution guide at https://github.com/secdev/scapy/blob/master/CONTRIBUTING.md (esp. section submitting-pull-requests) ✅
Have good commit hygiene. They must have the AI-Assisted tag as explained in the contributing guide. Please squash commits that belong together, and split commits that contain multiple features. ✅
AI: You must make sure that you understood the internal concepts of Scapy and have good test coverage (like >90%). Please review ALL the code you generated. ✅
Add unit tests or explain why they are not relevant. ✅
If the PR is still not finished, please create a Draft Pull Request ✅
If this PR contains more than 500 lines of code (excluding unit tests), consider splitting it. -> IEEE 1722 aka Audio/Video Transport Protocol (AVTP) specifies a large number of serialization formats over Ethernet. In this PR only the serialization formats mentioned in the Chapter 9 (these are the ones relevant for automotive usecases) are implemented. While the number of lines may seem large, the PR only has features/serialization formats belonging together.
New protocols: I considered interoperability tests with existing packages or utilities to ensure conformity of a newly generated protocol ✅ Tested conformance along with Ethernet, UDP/IP

Description

IEEE 1722 specifies the Audio/Video Transport Protocol (AVTP) along with several serialization formats for data which can be transported over Ethernet (e.g., various audio formats, video formats etc.). The 2016 version of the specification extended the spec to also include so-called control formats including automotive fieldbus frames, e.g., CAN, LIN, FlexRay etc. Now the recent 2025 version also includes serialization formats for I2C, SPI etc.

In this PR, we focus on the AVTP control formats (the ones described in the Chapter 6 of the IEEE 1722 spec.). This is not an exhaustive implementation of IEEE 1722 as we do not focus on the audio/video formats. The reason for contributing this to the upstream project is that we see currently traction for using IEEE 1722. We (myself and a few like-minded colleagues) are working on integrating this protocol into open source projects and encourage adoption. We are also working on:

AI Assisted: no for the implementation/copilot for unit tests. - Implemented the Chapter 9 of the IEEE 1722-2025 spec. - Added unit test cases Signed-off-by: Naresh Nayak <Naresh.Nayak@hs-furtwangen.de>

polybassa

Focus on simplicity and Scapy-likeness: the protocol fields themselves are mostly fine, but ACF framing, UDP encapsulation, and duplicated post_build/extract_padding should live in the packet hierarchy once.

Main priorities from this review: ACF boundary handling → I2C length fix → UDP wrapper layer → public version classes → collapse duplicated ACF plumbing.

This review was written with the help of AI (ChatGPT).

polybassa · 2026-09-21T07:40:10Z

+        ),
+        XByteField(name="sequence_num", default=0),
+        XLongField(name="stream_id", default=0),
+        PacketListField(name="acf_tlv", default=[], pkt_cls=AvtpAcfHeader),


Related to the ACF framing comment: ntscf_data_length is associated with acf_tlv, but this PacketListField does not use length_from.

Please pass:

length_from=lambda pkt: pkt.ntscf_data_length

(and the same for TSCF's stream_data_length) so outer payload after the ACF list is not eaten by the list field.

Updated as suggested.

polybassa · 2026-09-21T07:40:10Z

+
+    def post_build(self, pkt: bytes, pay: bytes) -> bytes:
+
+        # A correct I2C message has a 1-byte payload.


P1: this comment says a correct I2C message has a 1-byte payload and will be padded to 1 byte, but the implementation only updates acf_msg_length and returns pkt + pay.

The fixed I2C header is 15 bytes; with one data byte it becomes 16 (aligned). Without that byte:

I2C builds 15 bytes but can declare acf_msg_length == 3 → 12 bytes

I2C Brief has the same class of inconsistency

If the byte is structurally mandatory, model it as a field and drop the special post_build():

XByteField("data", 0)

If /Raw(...) must stay as the API, enforce exactly one byte there. The current halfway state should go. Same issue on AvtpAcfI2CBriefHeader.post_build().

The i2c_data is now modelled as an XByteField. The length is set by default to 4 for I2C format and to 2 for I2C_BRIEF format. To create custom frames, one must explicitly set the acf_msg_length to None.

polybassa · 2026-09-21T07:42:53Z

Thanks for your PR

- Removed unused class AvtpHeaderVersion - Added AvtpUdpEncapsulation as an independent layer - Use of AvtpStreamType instead of flattening the enum - Removed usage of "_underlayer" to check for UDP - Made private hidden classes now public and used match_subclass - Reduced bolier plate code for ACF data formats - Added additional unit tests for regression Signed-off-by: Naresh Nayak <Naresh.Nayak@hs-furtwangen.de>

- Removed unused class AvtpHeaderVersion - Added AvtpUdpEncapsulation as an independent layer - Use of AvtpStreamType instead of flattening the enum - Removed usage of "_underlayer" to check for UDP - Made private hidden classes now public and used match_subclass - Reduced bolier plate code for ACF data formats - Added additional unit tests for regression - AI-Assisted: no

nayakned · 2026-09-22T02:59:03Z

Thanks for your PR

Thanks for the exhaustive review. Got to know more about Scapy with this PR.
I have addressed most of your findings.
Perhaps where you can help me:

Please have an explicit look at how AvtpAcfTscfHeader and AvtpAcfNtscfHeader and their versions for their "Scapyness".
AvtpAcfI2CHeader also contains the payload. So I am toying with the idea to call it just AvtpAcfI2C. Same for the brief format.

Let me know if you have further findings or want me to rearrange the commit history.

polybassa

Follow-up on the latest tip (9f7e707): the structural cleanup looks good. Two concrete P1 regressions remain.

This review was written with the help of AI (ChatGPT).

polybassa · 2026-09-22T11:44:30Z

+                int.from_bytes(pkt[-10:-8], byteorder="big") & 0xF800
+            ) | pay_length
+            pkt = pkt[:-10] + struct.pack("!H", current_length) + pkt[-8:]
+            pkt += pay


P1 (follow-up): preserving an explicit control_data_length is correct, but pkt += pay ended up inside the if self.control_data_length is None: branch.

So:

AvtpCommonControlHeader(control_data_length=2) / Raw(b"abcd")

serializes without abcd at all (header-only). Auto-length still appends the payload.

Please keep the length update conditional, but always append:

def post_build(self, pkt, pay): if self.control_data_length is None: pay_length = len(pay) if len(pay) < 2**11 else 0 current_length = ( int.from_bytes(pkt[-10:-8], "big") & 0xF800 ) | pay_length pkt = pkt[:-10] + struct.pack("!H", current_length) + pkt[-8:] return pkt + pay

The existing test that only asserts decoded.control_data_length == 2 misses this; also check that the payload survives, e.g. assert bytes(avtp_pkt).endswith(b"\\x00" * 4) or inspect the decoded Raw.

Thats a good catch. I have fixed it now and added a regression.

- Fixed appending of payload to packet in AvtpCommonControlHeader. - Fixed padding arithmetic of AvtpAcfHeader - Renamed AvtpAcfI2CHeader to AvtpAcfI2CMessage - Renamed AvtpAcfI2CBriefHeader to AvtpAcfI2CBriefMessage - AI-Assisted: no

polybassa · 2026-09-23T18:45:57Z

+        return AvtpCommonStreamHeaderV1 if version else AvtpCommonStreamHeaderV0
+
+
+class AvtpCommonStreamHeaderV0(AvtpCommonStreamHeader):


P1/P2: Remove or redesign incomplete intermediate AVTP packet classes.

AvtpCommonStreamHeaderV0/V1 and AvtpAlternativeHeaderV0/V1 currently behave more like protocol fragments than complete Scapy Packet classes.

For example:

class AvtpCommonStreamHeaderV0(AvtpCommonStreamHeader): fields_desc = [ XByteEnumField("subtype", ...), # 8 bits BitField("h", 0, 1), # +1 BitField("version", 0, 3), # +3 -> 12 bits XByteField("sequence_num", 0), # byte field starts unaligned ... ]

After h + version, only four bits of the second byte have been defined. A regular XByteField cannot cleanly follow an unfinished BitField sequence.

AvtpAlternativeHeaderV0 is even more obviously incomplete:

class AvtpAlternativeHeaderV0(...): fields_desc = [ XByteEnumField("subtype", ...), BitField("h", 0, 1), BitField("version", 0, 3), ]

This ends after 12 bits.

Preferred solution

Do not add artificial reserved fields merely to align these classes.

If these types are only specification fragments, remove them as public packet classes and keep only the actual complete packets:

AvtpTscfHeaderV0 AvtpTscfHeaderV1 AvtpNtscfHeaderV0 AvtpNtscfHeaderV1

If reuse is necessary, use private field fragments rather than pretending the fragments are independently usable Scapy layers.

Regression test if they remain public

bytes(AvtpCommonStreamHeaderV0()) bytes(AvtpAlternativeHeaderV0())

Both should be valid if the classes remain public Packet classes. If writing sensible tests for these classes is difficult because they are only incomplete header fragments, that is another indication that they should not be public Scapy packets.

Thank you for your feedback here. I agree with you about the incomplete protocol fragments. The specification describes a common header from which the common control header, the common stream header, and the alternative header (and, of course, their versions) are derived. These headers are not expected to stand alone as individual formats. Individual formats, e.g., TSCF and NTSCF extend one of these headers. Additionally, the common header from which everything else is derived and the alternative header are incomplete and therefore not serializable.

So here is what I have done/propose:

All serializable classes are public. E.g.

AvtpCommonStreamHeader, AvtpCommonStreamHeaderV0, AvtpCommonStreamHeaderV1

AvtpCommonControlHeader

Non-serializable classes are private with the exception of AvtpCommonHeader as this is the main entry into the protocol in case any one wants to create their own customized/proprietary dataformat.

P.S.: There was an error in the AvtpCommonStreamHeaderV0. I somehow missed adding 4 fields which actually makes the class serializable.

polybassa · 2026-09-23T18:47:06Z

+        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:


P2: Have exactly one implementation of acf_msg_length serialization.

AvtpAcfHeader.post_build() and _AvtpAcfPaddedHeader.post_build() still duplicate the difficult part:

acf_length = (len(pkt) + len(pay)) // 4 & 0x1FF first_byte = ... second_byte = ... pkt = struct.pack(...) + struct.pack(...) + pkt[2:]

The padded version only differs because it must additionally: (1) calculate padding, (2) write padlength, (3) append padding bytes. The 9-bit ACF-length serialization should not exist twice.

Prefer one post_build() in AvtpAcfHeader, with padding factored into _pad_payload() that the subclass overrides:

class AvtpAcfHeader(Packet): def _pad_payload(self, pkt, pay): pad = -(len(pkt) + len(pay)) % 4 return pkt, pay + b"\x00" * pad def post_build(self, pkt, pay): if self.acf_msg_length is not None: return pkt + pay pkt, pay = self._pad_payload(pkt, pay) acf_length = (len(pkt) + len(pay)) // 4 & 0x1FF pkt = bytes([ (pkt[0] & 0xFE) | (acf_length >> 8), acf_length & 0xFF, ]) + pkt[2:] return pkt + pay class _AvtpAcfPaddedHeader(AvtpAcfHeader): def _pad_payload(self, pkt, pay): pad = -(len(pkt) + len(pay)) % 4 if pad: pkt = pkt[:2] + bytes([pkt[2] | (pad << 6)]) + pkt[3:] pay += b"\x00" * pad return pkt, pay

Goal: one post_build(), one acf_msg_length implementation, one copy of the 9-bit bit manipulation.

Updated as suggested

polybassa · 2026-09-23T18:47:26Z

+    def do_dissect_payload(self, s):
+        return super().do_dissect_payload(s[: len(s) - self.padlength])
+
+    def extract_padding(self, s):


P2: Remove duplicated extract_padding().

_AvtpAcfPaddedHeader currently repeats functionality already provided by AvtpAcfHeader:

def extract_padding(self, s): length = self.acf_msg_length * 4 - len(self.self_build()) return s[:length], s[length:]

The subclass implementation is effectively identical. Delete it and inherit from AvtpAcfHeader. There should be one implementation because the ACF boundary rule is identical for padded and unpadded messages.

Function removed as suggested in the review.

polybassa · 2026-09-23T18:50:29Z

+ether_pkt_recd = Ether(ether_pkt_bytes)
+assert type(ether_pkt_recd.payload) == AvtpNtscfHeaderV1
+
+udp_pkt = IP() / UDP(dport=17220) / AvtpUdpEncapsulation() / AvtpNtscfHeader()


P2: Add an actual UDP serialization/dissection round trip.

The UDP tests currently mainly validate the manually constructed in-memory layer tree. The new AvtpUdpEncapsulation architecture should also be tested through actual wire serialization and dissection, e.g.:

pkt = ( IP() / UDP(dport=17220) / AvtpUdpEncapsulation(encapsulation_sequence_num=123) / AvtpNtscfHeader(version=1) ) wire = bytes(pkt) decoded = IP(wire) assert AvtpUdpEncapsulation in decoded assert AvtpNtscfHeaderV1 in decoded assert decoded[AvtpUdpEncapsulation].encapsulation_sequence_num == 123

This specifically tests UDP → AvtpUdpEncapsulation → AvtpCommonHeader dispatch → NTSCF V1, rather than only testing manually stacked Python objects.

Test cases added/adapted

polybassa · 2026-09-23T18:51:00Z

+assert AvtpAcfGpcHeader in decoded
+
+= AvtpAcfI2CMessage and AvtpAcfI2CBriefMessage
+acf_i2c = AvtpAcfI2CMessage() / Raw(b"\x00\x00")


P2/P3: Clean up the I2C tests.

AvtpAcfI2CMessage now contains the fixed I2C data byte as a proper field (XByteField("i2c_data", 0)). A normal message should not need AvtpAcfI2CMessage() / Raw(b"\x00\x00") while still keeping the default fixed acf_msg_length — that constructs an intentionally inconsistent packet.

Prefer a normal round trip:

pkt = AvtpAcfI2CMessage(i2c_data=0x42) wire = bytes(pkt) decoded = AvtpAcfI2CMessage(wire) assert decoded.i2c_data == 0x42 assert decoded.acf_msg_length == 4

If malformed/custom messages are intentionally being tested, make that explicit, e.g. AvtpAcfI2CMessage(acf_msg_length=None) / Raw(b"\x00\x00"), or explicitly provide a deliberately inconsistent length and state that this is a fuzzing/crafting test. Normal protocol tests and malformed-packet tests should not accidentally overlap.

Test cases cleaned up.

polybassa · 2026-09-23T18:51:22Z

+import struct
+from enum import Enum
+
+from scapy.all import UDP, Ether, bind_layers  # pylint: disable=no-name-in-module


P3: Avoid scapy.all from within Scapy.

Instead of:

from scapy.all import UDP, Ether, bind_layers

prefer direct internal imports:

from scapy.packet import Packet, bind_layers from scapy.layers.inet import UDP from scapy.layers.l2 import Ether

scapy.all is primarily the user-facing convenience namespace.

Implemented as suggested.

polybassa · 2026-09-23T18:51:45Z

+            name="mode",
+            default=AncMode.ANC_8BIT,
+            size=2,
+            enum={i.name: i.value for i in AncMode},


P3: Pass Enum classes directly.

Where code still contains:

enum={i.name: i.value for i in AncMode}

prefer:

enum=AncMode

Scapy enum fields already support Python Enum classes. This is already done correctly in several other places in the PR, so the remaining instances should be made consistent.

Implemented as suggested.

polybassa · 2026-09-23T18:53:26Z

+        Dispatch the appropriate class based on the parsed subtype and version.
+        """
+        if pkt is not None:
+            parsed_type = ord(pkt[0:1])


P3: Simplify byte access.

Instead of ord(pkt[0:1]), use pkt[0]. And instead of unnecessary struct.pack() calls for individual bytes, bytes([value]) can often make the bit manipulation clearer.

Do this only where readability improves; don't rewrite working code merely for stylistic uniformity. Same pattern appears in the other dispatch hooks and ACF post_build() paths.

Implemented as suggested.

polybassa · 2026-09-23T18:55:33Z

Thanks for the PR, just a few more comments

- Replaced convenience import scapy.all with individual items import. - Consolidated padding functions in the AvtpAcfHeader and _AvtpAcfPaddedHeader. - AI-Assisted: no

- Added a serializable variant of AvtpCommonHeader as a fallback for unknown variants - Modified dispatch_hooks to avoid silent falling back to version 0 - Added missing fields to AvtpCommonStreamHeaderV0 - Incomplete header classes (other than AvtpCommonHeader) are now private classes - Improved unit test cases to check for unknown versions and UDP round trips - AI-Assisted: no

nayakned · 2026-09-25T06:22:48Z

@polybassa I have often discussed with my colleagues the best way to model the class/packets for this protocol.
Personally, I prefer to align with the specification. It is not our job to fix issues in the specification.
Therefore, I prefer to keep the packet classes AvtpCommon*, even though they will never be seen in a standalone correct system. This allows future developers to inherit or extend these classes or the IEEE 1722 module.

I hope exposing only serializable classes and making non-serializable classes (with the exception of AvtpCommonHeader) is an acceptable solution.

polybassa

Thanks for your work. I think we are almost there.

polybassa · 2026-09-26T04:34:09Z

+        """
+        Dispatch the appropriate class based on the parsed subtype and version.
+        """
+        if pkt is not None:


Please add a length check, otherwise the code two lines later can crash

polybassa · 2026-09-26T04:34:47Z

+    def dispatch_hook(cls, pkt=None, **kargs):
+        if "version" in kargs:
+            version = kargs["version"]
+        elif pkt:


polybassa · 2026-09-26T04:36:08Z

+        if self.control_data_length is None:
+            # Update the length fields on packet building
+            pay_length = len(pay) if len(pay) < 2**11 else 0
+            current_length = (


Please add a length check before accessing pkt by index

polybassa · 2026-09-26T04:36:24Z

+    def dispatch_hook(cls, pkt=None, **kargs):
+        if "version" in kargs:
+            version = kargs["version"]
+        elif pkt:


Length check

polybassa · 2026-09-26T04:37:09Z

+        """
+        Dispatch to the appropriate ACF header variant based on the acf_msg_type field.
+        """
+        if pkt is not None:


Length check

polybassa · 2026-09-26T04:51:20Z

+    acf_msg_type = AvtpAcfType.ACF_CAN_XL.value
+    fields_desc = [
+        _AvtpAcfPaddedHeader,
+        BitField(name="mtv", size=1, default=0),


CANXL is about to be merged, please also check this one for field name consistency

polybassa · 2026-09-26T04:52:05Z

+    fields_desc = [
+        AvtpAcfHeader,
+        BitField(name="reserved", size=12, default=0),
+        BitField(name="crc_type", size=4, default=0),


Could that be an enum field?

polybassa · 2026-09-26T04:52:35Z

+    def dispatch_hook(cls, pkt=None, **kargs):
+        if "version" in kargs:
+            version = kargs["version"]
+        elif pkt:


Length check

polybassa · 2026-09-26T04:54:17Z

+    def dispatch_hook(cls, pkt=None, **kargs):
+        if "version" in kargs:
+            version = kargs["version"]
+        elif pkt:


Length check

ieee1722: Initial implementation of parts of IEEE 1722.

61cbda8

AI Assisted: no for the implementation/copilot for unit tests. - Implemented the Chapter 9 of the IEEE 1722-2025 spec. - Added unit test cases Signed-off-by: Naresh Nayak <Naresh.Nayak@hs-furtwangen.de>

nayakned force-pushed the add_proto/avtp branch from d399543 to 61cbda8 Compare September 18, 2026 13:34

nayakned marked this pull request as ready for review September 18, 2026 13:54

nayakned force-pushed the add_proto/avtp branch from 61cbda8 to e5634dd Compare September 20, 2026 12:12

polybassa reviewed Sep 21, 2026

View reviewed changes

nayakned force-pushed the add_proto/avtp branch from e5634dd to 61cbda8 Compare September 21, 2026 13:39

nayakned force-pushed the add_proto/avtp branch from e4d1955 to b40d02e Compare September 22, 2026 02:11

nayakned force-pushed the add_proto/avtp branch from b40d02e to 9f7e707 Compare September 22, 2026 02:53

nayakned requested a review from polybassa September 22, 2026 02:59

polybassa reviewed Sep 22, 2026

View reviewed changes

nayakned requested a review from polybassa September 22, 2026 12:30

polybassa reviewed Sep 23, 2026

View reviewed changes

Comment thread scapy/contrib/ieee1722.py

polybassa reviewed Sep 23, 2026

View reviewed changes

Comment thread scapy/contrib/ieee1722.py

polybassa reviewed Sep 23, 2026

View reviewed changes

nayakned added 2 commits September 24, 2026 03:33

Minor fixes in the IEEE 1722 implementation

33d32c5

- Replaced convenience import scapy.all with individual items import. - Consolidated padding functions in the AvtpAcfHeader and _AvtpAcfPaddedHeader. - AI-Assisted: no

nayakned requested a review from polybassa September 25, 2026 06:22

polybassa requested changes Sep 26, 2026

View reviewed changes


		def post_build(self, pkt: bytes, pay: bytes) -> bytes:

		# A correct I2C message has a 1-byte payload.

		return AvtpCommonStreamHeaderV1 if version else AvtpCommonStreamHeaderV0


		class AvtpCommonStreamHeaderV0(AvtpCommonStreamHeader):

Uh oh!

Conversation

nayakned commented Sep 18, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Uh oh!

polybassa left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

polybassa commented Sep 21, 2026

Uh oh!

nayakned commented Sep 22, 2026

Uh oh!

polybassa left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Preferred solution

Regression test if they remain public

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

polybassa commented Sep 23, 2026

Uh oh!

nayakned commented Sep 25, 2026

Uh oh!

polybassa left a comment

Choose a reason for hiding this comment

Uh oh!

nayakned commented Sep 18, 2026 •

edited

Loading