diff --git a/pcapkit/protocols/data/transport/sctp.py b/pcapkit/protocols/data/transport/sctp.py index 2ee6a48a28..77fc00129f 100644 --- a/pcapkit/protocols/data/transport/sctp.py +++ b/pcapkit/protocols/data/transport/sctp.py @@ -104,7 +104,11 @@ class ErrorCause(Data): if TYPE_CHECKING: #: Trailing padding octets, as captured; present only when they are - #: not all zeros. + #: not all zeros, or fewer than the ``length`` calls for (:issue:`1474`). + #: Read back by ``from_data`` only, where fewer octets are honoured on + #: the last item of its list alone and zero-filled to the full width + #: anywhere else. ``make`` builds from keywords and takes no + #: ``padding``: every item it builds is padded in zeros. padding: 'bytes' @@ -267,7 +271,11 @@ class Parameter(Data): if TYPE_CHECKING: #: Trailing padding octets, as captured; present only when they are - #: not all zeros. + #: not all zeros, or fewer than the ``length`` calls for (:issue:`1474`). + #: Read back by ``from_data`` only, where fewer octets are honoured on + #: the last item of its list alone and zero-filled to the full width + #: anywhere else. ``make`` builds from keywords and takes no + #: ``padding``: every item it builds is padded in zeros. padding: 'bytes' @@ -380,7 +388,11 @@ class Chunk(Data): if TYPE_CHECKING: #: Trailing padding octets, as captured; present only when they are - #: not all zeros. + #: not all zeros, or fewer than the ``length`` calls for (:issue:`1474`). + #: Read back by ``from_data`` only, where fewer octets are honoured on + #: the last item of its list alone and zero-filled to the full width + #: anywhere else. ``make`` builds from keywords and takes no + #: ``padding``: every item it builds is padded in zeros. padding: 'bytes' diff --git a/pcapkit/protocols/schema/transport/sctp.py b/pcapkit/protocols/schema/transport/sctp.py index 7e92d5a438..5285a3333d 100644 --- a/pcapkit/protocols/schema/transport/sctp.py +++ b/pcapkit/protocols/schema/transport/sctp.py @@ -96,18 +96,45 @@ def padding_length(pkt: 'dict[str, Any]') -> 'int': Returns: Number of padding bytes, clamped to the number of bytes left in the enclosing structure, since :rfc:`9260#section-3.2` allows the final - padding of a packet to be omitted. + padding of a packet to be omitted. When packing, it is clamped as well + to the ``__padding__`` width :func:`pre_pack_padding` hands over, so a + final chunk captured without (all of) its padding rebuilds without it + (:issue:`1474`). """ length = pkt.get('length') or 0 padding = -length % 4 + width = pkt.get('__padding__') + if isinstance(width, int) and width >= 0: + padding = min(padding, width) + remaining = pkt.get('__length__') if not isinstance(remaining, int) or remaining < 0: return padding return min(padding, remaining) +def pre_pack_padding(schema: 'Schema', packet: 'dict[str, Any]') -> 'None': + """Hand a schema's recorded padding width to :func:`padding_length`. + + Args: + schema: Chunk, parameter or error cause being packed. + packet: Packet data, shared with the enclosing and sibling schemas. + + A schema whose padding was captured short -- only the last chunk of a + packet can be -- carries the width as captured under ``__padding__`` in its + ``__dict__``, set by :func:`~pcapkit.protocols.transport.sctp._restore_padding` + from the chunk's ``padding`` as parsed (:issue:`1474`). The key is written to + ``packet`` for *every* schema, ``None`` when there is no record, since + :meth:`Schema.pack ` shares one + ``packet`` mapping across siblings and a record left behind by one would + otherwise shorten the padding of the next. + + """ + packet['__padding__'] = schema.__dict__.get('__padding__') + + def bounded(length: 'Callable[[dict[str, Any]], int]') -> 'Callable[[dict[str, Any]], int]': """Clamp a list field's computed length to the bytes actually available. @@ -261,6 +288,10 @@ class ErrorCause(EnumSchema[Enum_CauseCode]): #: Cause length. length: 'int' = UInt16Field() + def pre_pack(self, packet: 'dict[str, Any]') -> 'None': + """Prepare ``packet`` data for packing process; see :func:`pre_pack_padding`.""" + pre_pack_padding(self, packet) + @schema_final class UnknownCause(ErrorCause): @@ -457,6 +488,10 @@ class Parameter(EnumSchema[Enum_Parameter]): #: Parameter length. length: 'int' = UInt16Field() + def pre_pack(self, packet: 'dict[str, Any]') -> 'None': + """Prepare ``packet`` data for packing process; see :func:`pre_pack_padding`.""" + pre_pack_padding(self, packet) + @schema_final class UnknownParameter(Parameter): @@ -606,6 +641,10 @@ class Chunk(EnumSchema[Enum_Chunk]): #: Chunk length, excluding any trailing padding. length: 'int' = UInt16Field() + def pre_pack(self, packet: 'dict[str, Any]') -> 'None': + """Prepare ``packet`` data for packing process; see :func:`pre_pack_padding`.""" + pre_pack_padding(self, packet) + @schema_final class UnknownChunk(Chunk): diff --git a/pcapkit/protocols/transport/sctp.py b/pcapkit/protocols/transport/sctp.py index 6eca9b5343..0c096e8b7b 100644 --- a/pcapkit/protocols/transport/sctp.py +++ b/pcapkit/protocols/transport/sctp.py @@ -182,6 +182,7 @@ Data_T = TypeVar('Data_T', bound=Data) Schema_T = TypeVar('Schema_T', bound=Schema) + List_T = TypeVar('List_T', bound=list) ChunkParser = Callable[[Schema_Chunk, NamedArg(Chunks, 'chunks')], Data_Chunk] ChunkConstructor = Callable[[Enum_Chunk, DefaultArg(Optional[Data_Chunk]), @@ -218,16 +219,34 @@ def _keep_padding(schema: 'Schema', data: 'Data_T') -> 'Data_T': Returns: ``data``, with a ``padding`` attribute holding the trailing padding as - captured if it is not all zeros -- the form a fresh build writes - (:issue:`1223`). :func:`_restore_padding` writes it back. + captured if it is not what a fresh build writes: all zeros + (:issue:`1223`), and as many as the ``length`` calls for. A final chunk + may arrive with fewer, or none (:issue:`1474`), in which case + ``padding`` holds the octets that were there, possibly ``b''``. + :func:`_restore_padding` writes it back. """ padding = schema.__dict__.get('padding') - if isinstance(padding, bytes) and any(padding): + if isinstance(padding, bytes) and (any(padding) or len(padding) < _padding_width(schema)): data.__update__([('padding', padding)]) return data +def _padding_width(schema: 'Schema') -> 'int': + """Number of padding octets a fresh build writes after ``schema``. + + Args: + schema: Chunk, parameter or error cause schema. + + Returns: + The octets that round its ``length`` up to a multiple of four + (:rfc:`9260#section-3.2`). + + """ + length = schema.__dict__.get('length') + return -length % 4 if isinstance(length, int) else 0 + + def _restore_padding(data: 'Optional[Data]', schema: 'Schema_T') -> 'Schema_T': """Write back the trailing padding :func:`_keep_padding` kept, if any. @@ -236,15 +255,60 @@ def _restore_padding(data: 'Optional[Data]', schema: 'Schema_T') -> 'Schema_T': schema: Constructed schema. Returns: - ``schema``, with its padding set from ``data``. + ``schema``, with its padding set from ``data``. Padding kept shorter + than ``length`` calls for also sets the schema's ``__padding__`` to its + width, which :func:`~pcapkit.protocols.schema.transport.sctp.padding_length` + packs to rather than zero-filling the octets that were never captured + (:issue:`1474`). """ padding = getattr(data, 'padding', None) if isinstance(padding, bytes) and 'padding' in schema.__fields__: schema.padding = padding + if len(padding) < _padding_width(schema): + schema.__dict__['__padding__'] = len(padding) return schema +def _last_item_only(schemas: 'List_T') -> 'List_T': + """Drop the short-padding record from every item but the last. + + Args: + schemas: Constructed chunk, parameter or cause list. + + Returns: + ``schemas``, in which only the last item may keep the ``__padding__`` + width :func:`_restore_padding` set. Only the end of a packet may be + short of its padding (:issue:`1474`); a short record on any earlier + item -- from an edited ``info``, say -- would misalign everything after + it, so that item packs to the full width, its ``padding`` octets + zero-filled to it. The same goes for the parameters or causes nested in + an earlier chunk, whose last item ends that chunk rather than the packet. + + """ + for schema in schemas[:-1]: + _drop_padding_record(schema) + return schemas + + +def _drop_padding_record(schema: 'Schema | bytes') -> 'None': + """Drop ``schema``'s short-padding record, and its nested items' too. + + Args: + schema: Chunk, parameter or cause that does not end the packet. + + """ + if not isinstance(schema, Schema): + return + if schema.__dict__.pop('__padding__', None) is not None: + schema.__updated__ = True + for name in ('parameters', 'error'): + nested = schema.__dict__.get(name) + if isinstance(nested, list): + for item in nested: + _drop_padding_record(item) + + class SCTP(Transport[Data_SCTP, Schema_SCTP], schema=Schema_SCTP, data=Data_SCTP): """This class implements Stream Control Transmission Protocol. @@ -987,11 +1051,11 @@ def _make_sctp_chunks(self, chunks: 'list[Schema_Chunk | tuple[Enum_Chunk, dict[ else: code, args = cast('tuple[Enum_Chunk, dict[str, Any]]', schema) chunks_list.append(self._make_sctp_chunk(code, None, **args)) - return chunks_list + return _last_item_only(chunks_list) for code, chunk in chunks.items(multi=True): chunks_list.append(self._make_sctp_chunk(code, chunk)) - return chunks_list + return _last_item_only(chunks_list) def _make_sctp_chunk(self, code: 'Enum_Chunk', chunk: 'Optional[Data_Chunk]' = None, **kwargs: 'Any') -> 'Schema_Chunk': @@ -1059,11 +1123,11 @@ def _make_sctp_parameters(self, parameters: 'list[Schema_Parameter | tuple[Enum_ else: code, args = cast('tuple[Enum_Parameter, dict[str, Any]]', schema) parameters_list.append(self._make_sctp_parameter(code, None, **args)) - return parameters_list + return _last_item_only(parameters_list) for code, parameter in parameters.items(multi=True): parameters_list.append(self._make_sctp_parameter(code, parameter)) - return parameters_list + return _last_item_only(parameters_list) def _make_sctp_parameter(self, code: 'Enum_Parameter', parameter: 'Optional[Data_Parameter]' = None, **kwargs: 'Any') -> 'Schema_Parameter': @@ -1131,11 +1195,11 @@ def _make_sctp_causes(self, causes: 'list[Schema_ErrorCause | tuple[Enum_CauseCo else: code, args = cast('tuple[Enum_CauseCode, dict[str, Any]]', schema) causes_list.append(self._make_sctp_cause(code, None, **args)) - return causes_list + return _last_item_only(causes_list) for code, cause in causes.items(multi=True): causes_list.append(self._make_sctp_cause(code, cause)) - return causes_list + return _last_item_only(causes_list) def _make_sctp_cause(self, code: 'Enum_CauseCode', cause: 'Optional[Data_ErrorCause]' = None, **kwargs: 'Any') -> 'Schema_ErrorCause': diff --git a/tests/protocols/transport/test_sctp_unpadded_last_chunk_unit.py b/tests/protocols/transport/test_sctp_unpadded_last_chunk_unit.py new file mode 100644 index 0000000000..c84e9650a1 --- /dev/null +++ b/tests/protocols/transport/test_sctp_unpadded_last_chunk_unit.py @@ -0,0 +1,185 @@ +"""#1474 -- a last SCTP chunk captured without (all of) its padding rebuilds as captured. + +:rfc:`9260#section-3.2` pads every chunk to four octets, but the last chunk of a +packet may arrive with fewer pad octets, or none. Parsing already read what was +there; the rebuild zero-filled the rest, so 17 octets in gave 20 out. The chunk's +``padding`` now records the octets as captured whenever they are short, and the +rebuild writes exactly those. + +""" +from __future__ import annotations + +import importlib.util +import struct +import unittest + +from tests._support import reimport_once_per_class + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) + +#: SCTP common header with a zero checksum. +COMMON = struct.pack('!HHII', 5000, 5001, 0xdeadbeef, 0) +#: DATA chunk TSN, stream, sequence and PPID. +DATA_HEAD = b'\x00\x00\x00\x01' + b'\x00\x01\x00\x02' + bytes(4) +#: INIT chunk fixed part. +INIT_HEAD = b'\x00\x00\x00\x01' + b'\x00\x00\xff\xff' + b'\x00\x01\x00\x01' + b'\x00\x00\x00\x01' + + +def chunk(type_: int, flags: int, value: bytes, pad: 'bytes | None' = None) -> bytes: + """An SCTP chunk, zero-padded to four octets unless ``pad`` is given.""" + length = 4 + len(value) + pad = bytes(-length % 4) if pad is None else pad + return struct.pack('!BBH', type_, flags, length) + value + pad + + +#: Packets whose last chunk is short of its padding, by name. +SHORT = { + 'unknown-chunk-no-padding': COMMON + chunk(0x3f, 0, b'a', pad=b''), + 'data-chunk-one-of-three-zero-pad': COMMON + chunk(0, 0x03, DATA_HEAD + b'X', pad=b'\x00'), + 'data-chunk-two-of-three-zero-pad': COMMON + chunk(0, 0x03, DATA_HEAD + b'X', pad=b'\x00\x00'), + 'data-chunk-two-of-three-nonzero-pad': + COMMON + chunk(0, 0x03, DATA_HEAD + b'X', pad=b'\xff\xff'), + 'cookie-echo-no-padding': COMMON + chunk(10, 0, b'abcde', pad=b''), + 'second-chunk-no-padding': COMMON + chunk(11, 0, b'') + chunk(0xbf, 0, b'abcde', pad=b''), +} + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class SCTPUnpaddedLastChunkUnitTests(unittest.TestCase): + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_rebuilds_from_info(self) -> None: + """#1474: ``from_data(info)`` writes back only the pad octets captured.""" + from pcapkit.protocols.transport.sctp import SCTP + + for name, raw in SHORT.items(): + with self.subTest(case=name): + self.assertEqual(SCTP.from_data(SCTP(raw).info).data.hex(), raw.hex()) + + def test_rebuilds_from_to_dict(self) -> None: + """#1474: the record survives ``info.to_dict()``.""" + from pcapkit.protocols.transport.sctp import SCTP + + for name, raw in SHORT.items(): + with self.subTest(case=name): + self.assertEqual(SCTP.from_data(SCTP(raw).info.to_dict()).data.hex(), raw.hex()) + + def test_info_records_padding_as_captured(self) -> None: + """#1474: the chunk's ``padding`` holds the octets that were there.""" + from pcapkit.protocols.transport.sctp import SCTP + + info = SCTP(SHORT['unknown-chunk-no-padding']).info + last = list(info.chunks.items(multi=True))[-1][1] + self.assertEqual(last.padding, b'') + + info = SCTP(SHORT['data-chunk-two-of-three-zero-pad']).info + last = list(info.chunks.items(multi=True))[-1][1] + self.assertEqual(last.padding, b'\x00\x00') + + def test_whole_zero_padding_is_not_recorded(self) -> None: + """#1474: a chunk padded as a fresh build pads keeps no ``padding`` key.""" + from pcapkit.protocols.transport.sctp import SCTP + + raw = COMMON + chunk(0x3f, 0, b'a') + last = list(SCTP(raw).info.chunks.items(multi=True))[-1][1] + self.assertNotIn('padding', last) + self.assertEqual(SCTP.from_data(SCTP(raw).info).data, raw) + + def test_parsed_short_chunk_is_short_only_when_last(self) -> None: + """#1474: a parsed short chunk moved ahead of another pads in full; last, it stays short.""" + from pcapkit.corekit.multidict import OrderedMultiDict + from pcapkit.protocols.transport.sctp import SCTP + + short = SCTP(COMMON + chunk(0x3f, 0, b'a', pad=b'')).info + fresh = SCTP(COMMON + chunk(0x3f, 0, b'ab')).info + chunks = OrderedMultiDict() # type: OrderedMultiDict + for info in (short, fresh): + for code, value in info.chunks.items(multi=True): + chunks.add(code, value) + + data = SCTP(srcport=5000, dstport=5001, vtag=0xdeadbeef, chksum=bytes(4), + chunks=chunks).data + self.assertEqual(data[12:], chunk(0x3f, 0, b'a') + chunk(0x3f, 0, b'ab')) + + chunks = OrderedMultiDict() + for info in (fresh, short): + for code, value in info.chunks.items(multi=True): + chunks.add(code, value) + data = SCTP(srcport=5000, dstport=5001, vtag=0xdeadbeef, chksum=bytes(4), + chunks=chunks).data + self.assertEqual(data[12:], chunk(0x3f, 0, b'ab') + chunk(0x3f, 0, b'a', pad=b'')) + + def test_short_record_on_a_non_last_chunk_rebuilds_aligned(self) -> None: + """#1474: only the last chunk may end short; an earlier one pads in full.""" + from pcapkit.protocols.transport.sctp import SCTP + + raw = COMMON + chunk(0x3f, 0, b'a') + chunk(0xbf, 0, b'b') + info = SCTP(raw).info + first = next(iter(info.chunks.values())) + first.__update__({'padding': b''}) + self.assertEqual(SCTP.from_data(info).data.hex(), raw.hex()) + + data = info.to_dict() + first = next(iter(data['chunks'].values())) + first.__update__({'padding': b''}) + self.assertEqual(SCTP.from_data(data).data.hex(), raw.hex()) + + def test_short_record_on_a_non_last_parameter_rebuilds_aligned(self) -> None: + """#1474: the same holds for parameters, which end their chunk, not the packet.""" + from pcapkit.protocols.transport.sctp import SCTP + + param = struct.pack('!HH', 0xbfff, 5) + b'q\x00\x00\x00' + raw = COMMON + chunk(1, 0, INIT_HEAD + param + param) + chunk(11, 0, b'') + info = SCTP(raw).info + init = next(iter(info.chunks.values())) + for item in init.parameters.values(): + item.__update__({'padding': b''}) + self.assertEqual(SCTP.from_data(info).data.hex(), raw.hex()) + + def test_short_last_parameter_in_a_non_last_chunk_rebuilds_aligned(self) -> None: + """#1474: a non-last chunk drops its last parameter's short record too.""" + from pcapkit.protocols.transport.sctp import SCTP + + # The INIT length excludes the last parameter's padding (RFC 9260 Sec. 3.2). + value = INIT_HEAD + struct.pack('!HH', 0xbfff, 5) + b'q' + raw = COMMON + struct.pack('!BBH', 1, 0, 4 + len(value)) + value + bytes(3) \ + + chunk(8, 0, b'') + info = SCTP(raw).info + init = list(info.chunks.items(multi=True))[0][1] + last = list(init.parameters.items(multi=True))[-1][1] + last.__update__({'padding': b''}) + + data = SCTP.from_data(info).data + self.assertEqual(len(data), 44) + self.assertEqual(data.hex(), raw.hex()) + + def test_short_last_cause_in_a_non_last_chunk_rebuilds_aligned(self) -> None: + """#1474: a non-last ABORT drops its last error cause's short record too.""" + from pcapkit.protocols.transport.sctp import SCTP + + # The ABORT length excludes the last cause's padding (RFC 9260 Sec. 3.2). + value = struct.pack('!HH', 0xffff, 5) + b'q' + raw = COMMON + struct.pack('!BBH', 6, 0, 4 + len(value)) + value + bytes(3) \ + + chunk(8, 0, b'') + info = SCTP(raw).info + abort = list(info.chunks.items(multi=True))[0][1] + last = list(abort.error.items(multi=True))[-1][1] + last.__update__({'padding': b''}) + + data = SCTP.from_data(info).data + self.assertEqual(len(data), 28) + self.assertEqual(data.hex(), raw.hex()) + + def test_make_still_pads_the_last_chunk(self) -> None: + """#1474: a chunk built from keywords is padded in full, last or not.""" + from pcapkit.const.sctp.chunk import Chunk + from pcapkit.protocols.transport.sctp import SCTP + + data = SCTP(chunks=[(Chunk.State_Cookie, {'cookie': b'abcde'})]).data + self.assertEqual(data[12:], chunk(10, 0, b'abcde')) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/transport/test_transport_edge_roundtrip_unit.py b/tests/protocols/transport/test_transport_edge_roundtrip_unit.py index 475b8eb197..e23a676a26 100644 --- a/tests/protocols/transport/test_transport_edge_roundtrip_unit.py +++ b/tests/protocols/transport/test_transport_edge_roundtrip_unit.py @@ -18,7 +18,7 @@ import unittest from tests.protocols import _edge_roundtrip as edge -from tests.protocols._edge_roundtrip import Case, Gap, MakeCase, Reject +from tests.protocols._edge_roundtrip import Case, MakeCase, Reject TCP = 'pcapkit.protocols.transport.tcp:TCP' UDP = 'pcapkit.protocols.transport.udp:UDP' @@ -120,18 +120,7 @@ def chunk(type_: int, flags: int, value: bytes, pad: 'bytes | None' = None) -> b 'sctp/data-no-user-data': Reject('ProtocolError', 'SCTP: [Chunk 0] invalid format'), } -KNOWN_FAILURES = ( - # ``sctp/last-chunk-unpadded`` is uncut: a final chunk sent without its pad - # octets rebuilds with three zero octets. The four cuts end inside the last - # chunk's pad the same way. - Gap(1474, 'A last SCTP chunk whose padding is absent rebuilds with it zero-filled: ' - '``padding_length`` (pcapkit/protocols/schema/transport/sctp.py:83), which ' - 'chunks, parameters and causes share, always writes the full padding.', - 'PADDED', '', - ('sctp/last-chunk-unpadded', 'sctp/data-1-octet-3-pad/cut31', - 'sctp/data-nonzero-pad/cut31', 'sctp/unassigned-type-action-00/cut19', - 'sctp/unassigned-type-action-10/cut23')), -) +KNOWN_FAILURES = () # type: tuple[edge.Gap, ...] @unittest.skipUnless(edge.HAS_RUNTIME, 'runtime dependencies not installed')