From 3d721832a872ba111ac9cf089a969199d83d2404 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Fri, 9 Oct 2026 13:55:13 -0400 Subject: [PATCH] fix(sctp): accept a final parameter or cause short of its padding (#1479) The span of a chunk's nested parameter or error cause list counts the chunk's trailing padding, which the last chunk of a packet may omit (RFC 9260 Sec. 3.2). The last item then read past the data and parsing raised ProtocolError for INIT, INIT ACK, HEARTBEAT, HEARTBEAT ACK, ABORT and ERROR. - Clamp the span to the octets present by wrapping nested_length in bounded, so the final item reads only the padding there is; the #1477 short-padding record then rebuilds it byte-exactly. New tests/protocols/transport/test_sctp_unpadded_final_param_unit.py. --- pcapkit/protocols/schema/transport/sctp.py | 17 +-- .../test_sctp_unpadded_final_param_unit.py | 112 ++++++++++++++++++ 2 files changed, 122 insertions(+), 7 deletions(-) create mode 100644 tests/protocols/transport/test_sctp_unpadded_final_param_unit.py diff --git a/pcapkit/protocols/schema/transport/sctp.py b/pcapkit/protocols/schema/transport/sctp.py index 5285a3333d..7d17478449 100644 --- a/pcapkit/protocols/schema/transport/sctp.py +++ b/pcapkit/protocols/schema/transport/sctp.py @@ -188,7 +188,10 @@ def nested_length(base: 'int') -> 'Callable[[dict[str, Any]], int]': final parameter's padding out of the chunk length, so the padding has to be consumed whether or not the length accounts for it. Folding it into the list's span does that, and leaves it visible as the field's - ``__option_padding__``. + ``__option_padding__``. The last chunk of a packet may omit that padding + altogether, so the chunks wrap this callback in :func:`bounded`: the span + then ends with the data, and the final parameter or cause reads only the + padding that is there (:issue:`1479`). * On construction, a separate padding field could not compute its own size: :meth:`Schema.pack ` passes one shared ``packet`` mapping down into the nested schemas, and each @@ -709,7 +712,7 @@ class INITChunk(Chunk, code=Enum_Chunk.Initiation): #: Optional and variable-length parameters, including the chunk's own #: trailing padding; see :func:`nested_length`. parameters: 'list[Parameter]' = OptionField( - length=nested_length(20), + length=bounded(nested_length(20)), base_schema=Parameter, type_name='type', registry=Parameter.registry, @@ -739,7 +742,7 @@ class INITACKChunk(Chunk, code=Enum_Chunk.Initiation_Acknowledgement): #: Optional and variable-length parameters, including the chunk's own #: trailing padding; see :func:`nested_length`. parameters: 'list[Parameter]' = OptionField( - length=nested_length(20), + length=bounded(nested_length(20)), base_schema=Parameter, type_name='type', registry=Parameter.registry, @@ -794,7 +797,7 @@ class HeartbeatChunk(Chunk, code=Enum_Chunk.Heartbeat_Request): #: Heartbeat information parameters, including the chunk's own trailing #: padding; see :func:`nested_length`. parameters: 'list[Parameter]' = OptionField( - length=nested_length(4), + length=bounded(nested_length(4)), base_schema=Parameter, type_name='type', registry=Parameter.registry, @@ -812,7 +815,7 @@ class HeartbeatACKChunk(Chunk, code=Enum_Chunk.Heartbeat_Acknowledgement): #: Heartbeat information parameters, including the chunk's own trailing #: padding; see :func:`nested_length`. parameters: 'list[Parameter]' = OptionField( - length=nested_length(4), + length=bounded(nested_length(4)), base_schema=Parameter, type_name='type', registry=Parameter.registry, @@ -835,7 +838,7 @@ class AbortChunk(Chunk, code=Enum_Chunk.Abort): #: Zero or more error causes, including the chunk's own trailing padding; #: see :func:`nested_length`. error: 'list[ErrorCause]' = OptionField( - length=nested_length(4), + length=bounded(nested_length(4)), base_schema=ErrorCause, type_name='code', registry=ErrorCause.registry, @@ -878,7 +881,7 @@ class ErrorChunk(Chunk, code=Enum_Chunk.Operation_Error): #: One or more error causes, including the chunk's own trailing padding; #: see :func:`nested_length`. error: 'list[ErrorCause]' = OptionField( - length=nested_length(4), + length=bounded(nested_length(4)), base_schema=ErrorCause, type_name='code', registry=ErrorCause.registry, diff --git a/tests/protocols/transport/test_sctp_unpadded_final_param_unit.py b/tests/protocols/transport/test_sctp_unpadded_final_param_unit.py new file mode 100644 index 0000000000..2784b0c026 --- /dev/null +++ b/tests/protocols/transport/test_sctp_unpadded_final_param_unit.py @@ -0,0 +1,112 @@ +"""#1479 -- a final SCTP parameter or error cause sent without its padding parses and rebuilds. + +:rfc:`9260#section-3.2` lets the last chunk of a packet omit its padding, and in +a chunk carrying a parameter or cause list that is the last item's padding. The +nested list's span counted that padding whether or not it was there, so the last +item read past the data and parsing raised :exc:`ProtocolError`. The span is now +clamped to the octets present. + +""" +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) +#: INIT and INIT ACK chunk fixed part. +INIT_HEAD = struct.pack('!IIHHI', 1, 1000, 10, 10, 1) + + +def tlv(type_: int, value: bytes, pad: 'bytes | None' = None) -> bytes: + """A parameter or cause, 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('!HH', type_, length) + value + pad + + +def chunk(type_: int, value: bytes) -> bytes: + """A chunk whose ``length`` covers ``value`` exactly, with no padding after it.""" + return struct.pack('!BBH', type_, 0, 4 + len(value)) + value + + +HOST = tlv(11, b'abcd\x00', pad=b'') +CAUSE = tlv(12, b'abc', pad=b'') + +#: Packets whose final parameter or cause has no padding, by name. +SHORT = { + 'init-sole-hostname': COMMON + chunk(1, INIT_HEAD + HOST), + 'init-ipv4-then-hostname': COMMON + chunk(1, INIT_HEAD + tlv(5, b'\x0a\x00\x00\x01') + HOST), + 'init-two-hostnames': COMMON + chunk(1, INIT_HEAD + tlv(11, b'abcd\x00') + HOST), + 'init-final-type-0x8000': COMMON + chunk(1, INIT_HEAD + tlv(0x8000, b'a', pad=b'')), + 'init-ack-hostname': COMMON + chunk(2, INIT_HEAD + HOST), + 'abort-sole-cause': bytes.fromhex('13881389deadbeef000000000600000b000c0007616263'), + 'abort-two-causes': COMMON + chunk(6, tlv(12, b'a') + CAUSE), + 'error-sole-cause': COMMON + chunk(9, CAUSE), + 'error-two-causes': COMMON + chunk(9, tlv(12, b'a') + CAUSE), + 'heartbeat': COMMON + chunk(4, tlv(1, b'a', pad=b'')), + 'heartbeat-ack': COMMON + chunk(5, tlv(1, b'a', pad=b'')), +} + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class SCTPUnpaddedFinalParameterUnitTests(unittest.TestCase): + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_parses_and_rebuilds_through_data(self) -> None: + """#1479: the packet parses and ``.data`` is the octets as captured.""" + from pcapkit.protocols.transport.sctp import SCTP + + for name, raw in SHORT.items(): + with self.subTest(case=name): + self.assertEqual(SCTP(raw).data.hex(), raw.hex()) + + def test_rebuilds_from_info(self) -> None: + """#1479: ``from_data(info)`` writes no padding the final item lacked.""" + 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: + """#1479: 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_only_the_final_item_records_short_padding(self) -> None: + """#1479: the final item keeps ``padding=b''``; the one before it keeps none.""" + from pcapkit.protocols.transport.sctp import SCTP + + for name, field in (('init-two-hostnames', 'parameters'), ('abort-two-causes', 'error')): + with self.subTest(case=name): + last = list(SCTP(SHORT[name]).info.chunks.items(multi=True))[-1][1] + items = [item for _, item in last[field].items(multi=True)] + self.assertEqual(len(items), 2) + self.assertNotIn('padding', items[0]) + self.assertEqual(items[1].padding, b'') + + def test_issue_init_repro_parses(self) -> None: + """#1479: the INIT from the issue parses, its host name read in full.""" + from pcapkit.protocols.transport.sctp import SCTP + + raw = bytes.fromhex('13881389deadbeef000000000100001d00000001000003e8000a000a' + '00000001000b00096162636465') + init = list(SCTP(raw).info.chunks.items(multi=True))[-1][1] + host = list(init.parameters.items(multi=True))[-1][1] + self.assertEqual(host.name, b'abcde') + self.assertEqual(host.padding, b'') + + +if __name__ == '__main__': + unittest.main()