diff --git a/pcapkit/protocols/data/transport/sctp.py b/pcapkit/protocols/data/transport/sctp.py index 88d9dd2d81..2ee6a48a28 100644 --- a/pcapkit/protocols/data/transport/sctp.py +++ b/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 diff --git a/pcapkit/protocols/schema/transport/sctp.py b/pcapkit/protocols/schema/transport/sctp.py index 91cd5a875d..7e92d5a438 100644 --- a/pcapkit/protocols/schema/transport/sctp.py +++ b/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 + ` 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 `. + 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': ... diff --git a/pcapkit/protocols/transport/sctp.py b/pcapkit/protocols/transport/sctp.py index 2003ae6a94..6eca9b5343 100644 --- a/pcapkit/protocols/transport/sctp.py +++ b/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) diff --git a/tests/protocols/test_declared_length_overrun_unit.py b/tests/protocols/test_declared_length_overrun_unit.py index 2055b83ac3..a2503f1c05 100644 --- a/tests/protocols/test_declared_length_overrun_unit.py +++ b/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', diff --git a/tests/protocols/transport/test_sctp_stray_octets_unit.py b/tests/protocols/transport/test_sctp_stray_octets_unit.py new file mode 100644 index 0000000000..7b8d777313 --- /dev/null +++ b/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() diff --git a/tests/protocols/transport/test_sctp_unit.py b/tests/protocols/transport/test_sctp_unit.py index 7f97baebfb..45d40dee2c 100644 --- a/tests/protocols/transport/test_sctp_unit.py +++ b/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):