fix(sctp)!: keep the octets after the last whole chunk as captured (#1468)

pcapkit/protocols/data/transport/sctp.py

@@ -614,4 +614,8 @@ class SCTP(Protocol):
     chunks: 'OrderedMultiDict[ChunkType, Chunk]'
 
     if TYPE_CHECKING:
+        #: Octets captured after the last whole chunk, present only when there
+        #: are any.
+        trailer: 'bytes'
+
         def __init__(self, srcport: 'AppType', dstport: 'AppType', vtag: 'int', chksum: 'bytes', chunks: 'OrderedMultiDict[ChunkType, Chunk]') -> 'None': ...  # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin

pcapkit/protocols/schema/transport/sctp.py

@@ -2,6 +2,7 @@
 # mypy: disable-error-code=assignment
 """header schema for stream control transmission protocol"""
 
+import io
 from typing import TYPE_CHECKING
 
 from pcapkit.const.reg.apptype import AppType as Enum_AppType
@@ -13,7 +14,7 @@
     PayloadProtocolIdentifier as Enum_PayloadProtocolIdentifier
 from pcapkit.corekit.fields.collections import ListField, OptionField
 from pcapkit.corekit.fields.ipaddress import IPv4AddressField, IPv6AddressField
-from pcapkit.corekit.fields.misc import SchemaField
+from pcapkit.corekit.fields.misc import PayloadField, SchemaField
 from pcapkit.corekit.fields.numbers import EnumField, UInt16Field, UInt32Field
 from pcapkit.corekit.fields.strings import BitField, BytesField, PaddingField
 from pcapkit.protocols.schema.schema import EnumSchema, Schema, schema_final
@@ -49,7 +50,7 @@
 
 if TYPE_CHECKING:
     from ipaddress import IPv4Address, IPv6Address
-    from typing import Any, Callable
+    from typing import IO, Any, Callable
 
     from pcapkit.protocols.protocol import ProtocolBase
 
@@ -890,6 +891,70 @@ class ShutdownCompleteChunk(Chunk, code=Enum_Chunk.Shutdown_Complete):
         def __init__(self, type: 'Enum_Chunk', flags: 'TBitFlags', length: 'int') -> 'None': ...
 
 
+def whole_chunks_length(data: 'bytes') -> 'int':
+    """Length of the run of whole chunks at the front of ``data``.
+
+    Args:
+        data: Chunk area of an SCTP packet.
+
+    Returns:
+        Number of octets up to the end of the last chunk whose declared
+        ``length`` octets are all present, with as much of its padding as there
+        is, c.f. :func:`padding_length`. A ``length`` below the four-octet chunk
+        header (:rfc:`9260#section-3.2`) delimits nothing, so the run stops
+        there too; four zero octets would otherwise read as a DATA chunk.
+
+    """
+    end, size = 0, len(data)
+    while size - end >= 4:
+        length = int.from_bytes(data[end + 2:end + 4], 'big')
+        stop = end + length
+        if length < 4 or stop > size:
+            break
+        end = min(stop + (-length % 4), size)
+    return end
+
+
+class ChunkListField(OptionField):
+    """Chunk list that stops at the last whole chunk.
+
+    SCTP has no packet length field, so octets after the last whole chunk --
+    fewer than a chunk header, or a chunk whose ``length`` runs past the data --
+    cannot be told apart from a chunk cut short. They are not parsed but
+    reported as option padding, which :meth:`Schema.unpack
+    <pcapkit.protocols.schema.schema.Schema.unpack>` hands back to the stream for
+    :attr:`SCTP.trailer` to keep as captured (:issue:`1468`).
+
+    """
+
+    def unpack(self, buffer: 'bytes | IO[bytes]', packet: 'dict[str, Any]') -> 'list[Chunk]':
+        """Unpack field value from :obj:`bytes`.
+
+        Args:
+            buffer: Field buffer.
+            packet: Packet data.
+
+        Returns:
+            Unpacked whole chunks.
+
+        """
+        if isinstance(buffer, bytes):
+            data = buffer
+        else:
+            start = buffer.tell()
+            data = buffer.read(max(self._length, 0))
+            buffer.seek(start, io.SEEK_SET)
+
+        area, whole = self._length, whole_chunks_length(data)
+        self._length = whole
+        try:
+            chunks = super().unpack(data[:whole], packet)
+        finally:
+            self._length = area
+        self._option_padding += len(data) - whole
+        return chunks
+
+
 @schema_final
 class SCTP(Schema):
     """Header schema for SCTP packets.
@@ -913,14 +978,18 @@ class SCTP(Schema):
     #: Checksum, as a CRC32c over the whole packet with this field zeroed.
     chksum: 'bytes' = BytesField(length=4)
     #: Chunks.
-    chunks: 'list[Chunk]' = OptionField(
+    chunks: 'list[Chunk]' = ChunkListField(
         length=lambda pkt: max(pkt['__length__'], 0),
         base_schema=Chunk,
         type_name='type',
         registry=Chunk.registry,
     )
+    #: Octets captured after the last whole chunk, c.f. :class:`ChunkListField`
+    #: and :attr:`IPv4.trailer <pcapkit.protocols.schema.internet.ipv4.IPv4.trailer>`.
+    trailer: 'bytes' = PayloadField(default=b'')
 
     if TYPE_CHECKING:
         def __init__(self, srcport: 'Enum_AppType | int', dstport: 'Enum_AppType | int',
                      vtag: 'int', chksum: 'bytes',
-                     chunks: 'list[Chunk | bytes] | bytes') -> 'None': ...
+                     chunks: 'list[Chunk | bytes] | bytes',
+                     trailer: 'bytes' = b'') -> 'None': ...

pcapkit/protocols/transport/sctp.py

@@ -566,6 +566,14 @@ def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_SCTP':
             chunks=self._read_sctp_chunks(),
         )
 
+        # NOTE: With no packet length field, octets after the last whole chunk
+        # cannot be told apart from a chunk cut short, so they are kept as
+        # captured rather than rejected (:issue:`1468`).
+        if schema.trailer:
+            sctp.__update__([
+                ('trailer', schema.trailer),
+            ])
+
         # NOTE: The next layer is named by the DATA chunk's payload protocol
         # identifier, not by a port number. A packet may bundle several DATA
         # chunks; we dispatch on the first one, and the rest stay recorded in
@@ -587,6 +595,7 @@ def make(self,
              vtag: 'int' = 0,
              chksum: 'Optional[bytes]' = None,
              chunks: 'Optional[list[Schema_Chunk | tuple[Enum_Chunk, dict[str, Any]] | bytes] | Chunks]' = None,  # pylint: disable=line-too-long
+             trailer: 'bytes' = b'',
              **kwargs: 'Any') -> 'Schema_SCTP':
         """Make (construct) packet data.
 
@@ -598,6 +607,8 @@ def make(self,
                 constructed packet is computed and inserted, per
                 :rfc:`9260#section-6.8`.
             chunks: SCTP chunks.
+            trailer: Octets after the last chunk, written as is and covered by
+                the computed checksum.
             **kwargs: Arbitrary keyword arguments.
 
         Returns:
@@ -620,6 +631,7 @@ def make(self,
             vtag=vtag,
             chksum=b'\x00\x00\x00\x00' if chksum is None else chksum,
             chunks=chunks_value,
+            trailer=trailer,
         )
 
         if chksum is None:
@@ -856,6 +868,7 @@ def _make_data(cls, data: 'Data_SCTP') -> 'dict[str, Any]':  # type: ignore[over
             'vtag': data.vtag,
             'chksum': data.chksum,
             'chunks': data.chunks,
+            'trailer': data.get('trailer', b''),
         }
 
     def _get_payload(self) -> 'bytes':
@@ -918,10 +931,20 @@ def _read_sctp_chunks(self) -> 'Chunks':
         Returns:
             Extracted SCTP chunks.
 
+        Raises:
+            ProtocolError: If a raw chunk given to :meth:`make` is not a whole
+                chunk. Parsed, its octets would fall to the trailer rather than
+                to the chunk list, so it is rejected, as it was before
+                :issue:`1468`.
+
         """
         chunks = OrderedMultiDict()  # type: Chunks
 
-        for schema in self.__header__.chunks:
+        for schema in cast('list[Schema_Chunk | bytes]', self.__header__.chunks):
+            if isinstance(schema, bytes):
+                raise ProtocolError(f'{self.alias}: raw chunk {schema!r} runs past the end of '
+                                    'the data, so it is not a whole chunk')
+
             code = schema.type
             name = self._lookup_registry(self.__chunk__, code)
 

tests/protocols/test_declared_length_overrun_unit.py

@@ -73,8 +73,9 @@ def _ipv6(proto: int, payload: bytes) -> bytes:
                 bytes.fromhex('3b010000 00000000 010400000000 0103')),
     'HOPOPT PadN': ('pcapkit.protocols.internet.hopopt', 'HOPOPT', _ext6(bytes.fromhex('010700000000'))),
     'IPv6-Opts PadN': ('pcapkit.protocols.internet.ipv6_opts', 'IPv6_Opts', _ext6(bytes.fromhex('010700000000'))),
-    'SCTP INIT parameter': ('pcapkit.protocols.transport.sctp', 'SCTP', _sctp(
-        bytes.fromhex('01000020 00000001 00010000 00010001 00000001 000c0010 00050000'))),
+    # An SCTP chunk declaring more octets than the packet holds is not here:
+    # with no packet length field it is kept as the packet's trailer instead,
+    # c.f. ``tests/protocols/transport/test_sctp_stray_octets_unit.py`` (#1468).
     # an unassigned kind declaring 12 octets, 8 present
     'TCP unassigned': ('pcapkit.protocols.transport.tcp', 'TCP', _tcp(bytes.fromhex('4f0caabb ccddeeff'))),
     # an unassigned option declaring 12 octets, 8 present in a complete
@@ -103,8 +104,6 @@ def _ipv6(proto: int, payload: bytes) -> bytes:
     'MH PadN': ('pcapkit.protocols.internet.ipv6', 'IPv6', _ipv6(135, OVERRUNS['MH PadN'][2])),
     'HOPOPT PadN': ('pcapkit.protocols.internet.ipv6', 'IPv6', _ipv6(0, OVERRUNS['HOPOPT PadN'][2])),
     'IPv6-Opts PadN': ('pcapkit.protocols.internet.ipv6', 'IPv6', _ipv6(60, OVERRUNS['IPv6-Opts PadN'][2])),
-    'SCTP INIT parameter': ('pcapkit.protocols.internet.ipv4', 'IPv4',
-                            _ipv4(b'', ihl=5, proto=132, payload=OVERRUNS['SCTP INIT parameter'][2])),
     'TCP unassigned': ('pcapkit.protocols.internet.ipv4', 'IPv4',
                        _ipv4(b'', ihl=5, proto=6, payload=OVERRUNS['TCP unassigned'][2])),
     'IPv4 unassigned': ('pcapkit.protocols.link.ethernet', 'Ethernet',

tests/protocols/transport/test_sctp_stray_octets_unit.py

@@ -0,0 +1,146 @@
+# -*- coding: utf-8 -*-
+"""Octets after the last whole SCTP chunk parse and survive a rebuild.
+
+GitHub issue #1468: SCTP has no packet length field, so octets left after the
+last whole chunk -- fewer than a chunk header, or a chunk whose declared length
+runs past the data -- cannot be told apart from a chunk cut short. They made
+:class:`~pcapkit.protocols.transport.sctp.SCTP` raise ``ProtocolError: Field
+chunks has an option that runs past the end of the data``, so the packet had no
+round trip at all.
+
+Per the #1458 ruling they are now recorded rather than rejected: the chunk list
+stops at the last whole chunk and the rest is kept as ``trailer``, as IPv4,
+IPv6 (#1209), L2TPv2 and OSPF (#1455) do. :meth:`make` writes a ``trailer``
+after the chunks.
+
+Every case builds its own octets in memory and reads no capture. Classes are
+imported inside each test, after :func:`~tests._support.reimport_once_per_class`,
+c.f. ``tests/protocols/link/test_ethernet_mac_roundtrip_unit.py``.
+
+"""
+
+import struct
+import unittest
+import warnings
+
+from tests._support import reimport_once_per_class
+
+#: SCTP common header.
+COMMON = struct.pack('!HHII', 1, 2, 0, 0)
+#: COOKIE ACK chunk.
+COOKIE_ACK = struct.pack('!BBH', 11, 0, 4)
+#: DATA chunk carrying ``b'hi'`` under PPID 0, padded to eight octets.
+DATA = struct.pack('!BBHIHHI', 0, 3, 18, 1, 0, 0, 0) + b'hi' + bytes(2)
+#: INIT chunk declaring 32 octets, 28 present.
+INIT_CUT = bytes.fromhex('01000020 00000001 00010000 00010001 00000001 000c0010 00050000')
+
+#: Packets with octets after the last whole chunk: (octets, chunk count, trailer).
+STRAY = {
+    '1 octet': (COMMON + COOKIE_ACK + b'x', 1, b'x'),
+    '3 octets': (COMMON + COOKIE_ACK + b'xyz', 1, b'xyz'),
+    '4 octets': (COMMON + COOKIE_ACK + b'wxyz', 1, b'wxyz'),
+    '4 zero octets': (COMMON + COOKIE_ACK + bytes(4), 1, bytes(4)),
+    '5 octets': (COMMON + COOKIE_ACK + b'vwxyz', 1, b'vwxyz'),
+    'after DATA': (COMMON + DATA + b'x', 1, b'x'),
+    'chunk length past the data': (COMMON + COOKIE_ACK + struct.pack('!BBH', 11, 0, 40) + b'abcd', 1,
+                                   struct.pack('!BBH', 11, 0, 40) + b'abcd'),
+    'first chunk past the data': (COMMON + DATA[:14], 0, DATA[:14]),
+    'chunk length below its header': (COMMON + COOKIE_ACK + struct.pack('!BBH', 11, 0, 2), 1,
+                                      struct.pack('!BBH', 11, 0, 2)),
+    'under a chunk header': (COMMON + b'ab', 0, b'ab'),
+    # formerly rejected by tests/protocols/test_declared_length_overrun_unit.py
+    'INIT past the data': (COMMON + INIT_CUT, 0, INIT_CUT),
+}
+
+#: Packets that end on a chunk boundary.
+NO_STRAY = {
+    'header only': COMMON,
+    'COOKIE ACK': COMMON + COOKIE_ACK,
+    'DATA then COOKIE ACK': COMMON + DATA + COOKIE_ACK,
+}
+
+
+class TestSCTPStrayOctets(unittest.TestCase):
+    """Pin the octets after the last whole SCTP chunk through ``from_data``."""
+
+    def setUp(self) -> None:
+        reimport_once_per_class(self)
+
+    def test_packet_parses_and_rebuilds_byte_for_byte(self) -> None:
+        from pcapkit.protocols.transport.sctp import SCTP
+
+        for name, (raw, count, trailer) in STRAY.items():
+            with self.subTest(packet=name):
+                parsed = SCTP(raw, len(raw))
+                self.assertEqual(len(parsed.info.chunks), count)
+                self.assertEqual(parsed.info.trailer, trailer)
+                self.assertEqual(SCTP.from_data(parsed.info).data, raw)
+                self.assertEqual(SCTP.from_data(parsed.info.to_dict()).data, raw)
+
+    def test_schema_keeps_the_trailer(self) -> None:
+        from pcapkit.protocols.schema.transport.sctp import SCTP as Schema_SCTP
+
+        for name, (raw, count, trailer) in STRAY.items():
+            with self.subTest(packet=name):
+                schema = Schema_SCTP.unpack(raw)
+                self.assertEqual(len(schema.chunks), count)
+                self.assertEqual(schema.trailer, trailer)
+                self.assertEqual(schema.pack(), raw)
+
+    def test_no_trailer_key_on_a_chunk_boundary(self) -> None:
+        from pcapkit.protocols.transport.sctp import SCTP
+
+        for name, raw in NO_STRAY.items():
+            with self.subTest(packet=name):
+                parsed = SCTP(raw, len(raw))
+                self.assertNotIn('trailer', parsed.info)
+                self.assertEqual(SCTP.from_data(parsed.info).data, raw)
+
+    def test_data_chunk_still_dispatches(self) -> None:
+        from pcapkit.protocols.transport.sctp import SCTP
+
+        raw = COMMON + DATA + b'x'
+        parsed = SCTP(raw, len(raw))
+        self.assertEqual(parsed.info.trailer, b'x')
+        self.assertEqual(bytes(parsed.payload.data), b'hi')
+
+    def test_parent_rebuilds_byte_for_byte(self) -> None:
+        from pcapkit.protocols.internet.ipv4 import IPv4
+        from pcapkit.protocols.transport.sctp import SCTP
+
+        for name, (raw, _, trailer) in STRAY.items():
+            with self.subTest(packet=name):
+                frame = (struct.pack('!BBHHHBBH4s4s', 0x45, 0, 20 + len(raw), 1, 0, 64, 132, 0,
+                                     bytes(4), bytes(4)) + raw)
+                with warnings.catch_warnings():
+                    warnings.simplefilter('ignore')
+                    parsed = IPv4(frame, len(frame))
+                self.assertIsInstance(parsed.payload, SCTP)
+                self.assertEqual(parsed.payload.info.trailer, trailer)
+                self.assertEqual(IPv4.from_data(parsed.info).data, frame)
+
+    def test_raw_chunk_that_is_not_whole_is_rejected(self) -> None:
+        from pcapkit.protocols.transport.sctp import SCTP
+        from pcapkit.utilities.exceptions import ProtocolError
+
+        with warnings.catch_warnings():
+            warnings.simplefilter('ignore')
+            with self.assertRaisesRegex(ProtocolError, r"raw chunk b'\\x01' runs past the end of the data"):
+                SCTP(chunks=[b'\x01'])
+
+    def test_make_writes_the_trailer_under_the_checksum(self) -> None:
+        from pcapkit.const.sctp.chunk import Chunk as Enum_Chunk
+        from pcapkit.protocols.transport.sctp import SCTP
+
+        with warnings.catch_warnings():
+            warnings.simplefilter('ignore')
+            raw = SCTP(srcport=1, dstport=2, chunks=[(Enum_Chunk.Cookie_Acknowledgement, {})],
+                       trailer=b'xyz').data
+        self.assertEqual(raw[12:], COOKIE_ACK + b'xyz')
+        parsed = SCTP(raw, len(raw))
+        self.assertEqual(parsed.info.trailer, b'xyz')
+        self.assertTrue(parsed.checksum_valid)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/protocols/transport/test_sctp_unit.py

@@ -854,12 +854,15 @@ def test_malformed_chunk_lengths_raise(self) -> None:
                     '0002' '0003' '0005' '0005',
             # SHUTDOWN must be exactly 8.
             'shutdown': '07' '00' '0004' '0a0b0c10',
-            # SHUTDOWN ACK must be exactly 4.
-            'shutdown_ack': '08' '00' '0008',
+            # SHUTDOWN ACK must be exactly 4. The schema reads only the
+            # header, so the other four declared octets are a well-formed
+            # chunk of the same type; with them absent, the chunk would run
+            # past the data and be kept as the packet's trailer (#1468).
+            'shutdown_ack': '08' '00' '0008' '08000004',
             # COOKIE ACK must be exactly 4.
-            'cookie_ack': '0b' '00' '0008',
+            'cookie_ack': '0b' '00' '0008' '0b000004',
             # SHUTDOWN COMPLETE must be exactly 4.
-            'shutdown_complete': '0e' '00' '0008',
+            'shutdown_complete': '0e' '00' '0008' '0e000004',
         }
         for name, body in cases.items():
             with self.subTest(chunk=name):