diff --git a/pcapkit/protocols/data/transport/sctp.py b/pcapkit/protocols/data/transport/sctp.py index 5df0a08211..29e22a52b6 100644 --- a/pcapkit/protocols/data/transport/sctp.py +++ b/pcapkit/protocols/data/transport/sctp.py @@ -52,6 +52,8 @@ class DATAChunkFlags(Data): """Data model for SCTP DATA chunk flags.""" + #: Reserved bits, kept as received. + reserved: 'int' #: (I)mmediate bit, i.e., request a SACK chunk without delay. I: 'bool' #: (U)nordered bit, i.e., no stream sequence number is assigned. @@ -62,7 +64,7 @@ class DATAChunkFlags(Data): E: 'bool' if TYPE_CHECKING: - def __init__(self, I: 'bool', U: 'bool', B: 'bool', E: 'bool') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + def __init__(self, reserved: 'int', I: 'bool', U: 'bool', B: 'bool', E: 'bool') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @info_final @@ -70,11 +72,13 @@ class TBitFlags(Data): """Data model for SCTP chunk flags carrying only the T bit, i.e., ABORT and SHUTDOWN COMPLETE chunks.""" + #: Reserved bits, kept as received. + reserved: 'int' #: T bit, i.e., the verification tag has been reflected. T: 'bool' if TYPE_CHECKING: - def __init__(self, T: 'bool') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + def __init__(self, reserved: 'int', T: 'bool') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @info_final @@ -116,9 +120,11 @@ class InvalidStreamIdentifierCause(ErrorCause): #: Stream identifier of the offending DATA chunk. stream_id: 'int' + #: Reserved field, kept as received. + reserved: 'bytes' if TYPE_CHECKING: - def __init__(self, code: 'CauseCode', length: 'int', stream_id: 'int') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + def __init__(self, code: 'CauseCode', length: 'int', stream_id: 'int', reserved: 'bytes') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @info_final @@ -401,6 +407,8 @@ def __init__(self, type: 'ChunkType', length: 'int', flags: 'DATAChunkFlags', ts class INITChunk(Chunk): """Data model for SCTP INIT chunks.""" + #: Raw chunk flags, kept as received. + flags: 'bytes' #: Initiate tag. init_tag: 'int' #: Advertised receiver window credit. @@ -415,13 +423,15 @@ class INITChunk(Chunk): parameters: 'OrderedMultiDict[ParameterType, Parameter]' if TYPE_CHECKING: - def __init__(self, type: 'ChunkType', length: 'int', init_tag: 'int', a_rwnd: 'int', outbound_streams: 'int', inbound_streams: 'int', init_tsn: 'int', parameters: 'OrderedMultiDict[ParameterType, Parameter]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + def __init__(self, type: 'ChunkType', length: 'int', flags: 'bytes', init_tag: 'int', a_rwnd: 'int', outbound_streams: 'int', inbound_streams: 'int', init_tsn: 'int', parameters: 'OrderedMultiDict[ParameterType, Parameter]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @info_final class INITACKChunk(Chunk): """Data model for SCTP INIT ACK chunks.""" + #: Raw chunk flags, kept as received. + flags: 'bytes' #: Initiate tag. init_tag: 'int' #: Advertised receiver window credit. @@ -436,13 +446,15 @@ class INITACKChunk(Chunk): parameters: 'OrderedMultiDict[ParameterType, Parameter]' if TYPE_CHECKING: - def __init__(self, type: 'ChunkType', length: 'int', init_tag: 'int', a_rwnd: 'int', outbound_streams: 'int', inbound_streams: 'int', init_tsn: 'int', parameters: 'OrderedMultiDict[ParameterType, Parameter]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + def __init__(self, type: 'ChunkType', length: 'int', flags: 'bytes', init_tag: 'int', a_rwnd: 'int', outbound_streams: 'int', inbound_streams: 'int', init_tsn: 'int', parameters: 'OrderedMultiDict[ParameterType, Parameter]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @info_final class SACKChunk(Chunk): """Data model for SCTP SACK chunks.""" + #: Raw chunk flags, kept as received. + flags: 'bytes' #: Cumulative TSN ack. cum_tsn_ack: 'int' #: Advertised receiver window credit. @@ -457,29 +469,33 @@ class SACKChunk(Chunk): dup_tsn: 'tuple[int, ...]' if TYPE_CHECKING: - def __init__(self, type: 'ChunkType', length: 'int', cum_tsn_ack: 'int', a_rwnd: 'int', num_gap_blocks: 'int', num_dup_tsn: 'int', gap_blocks: 'tuple[GapAckBlock, ...]', dup_tsn: 'tuple[int, ...]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + def __init__(self, type: 'ChunkType', length: 'int', flags: 'bytes', cum_tsn_ack: 'int', a_rwnd: 'int', num_gap_blocks: 'int', num_dup_tsn: 'int', gap_blocks: 'tuple[GapAckBlock, ...]', dup_tsn: 'tuple[int, ...]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @info_final class HeartbeatChunk(Chunk): """Data model for SCTP HEARTBEAT chunks.""" + #: Raw chunk flags, kept as received. + flags: 'bytes' #: Heartbeat information parameters. parameters: 'OrderedMultiDict[ParameterType, Parameter]' if TYPE_CHECKING: - def __init__(self, type: 'ChunkType', length: 'int', parameters: 'OrderedMultiDict[ParameterType, Parameter]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + def __init__(self, type: 'ChunkType', length: 'int', flags: 'bytes', parameters: 'OrderedMultiDict[ParameterType, Parameter]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @info_final class HeartbeatACKChunk(Chunk): """Data model for SCTP HEARTBEAT ACK chunks.""" + #: Raw chunk flags, kept as received. + flags: 'bytes' #: Heartbeat information parameters. parameters: 'OrderedMultiDict[ParameterType, Parameter]' if TYPE_CHECKING: - def __init__(self, type: 'ChunkType', length: 'int', parameters: 'OrderedMultiDict[ParameterType, Parameter]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + def __init__(self, type: 'ChunkType', length: 'int', flags: 'bytes', parameters: 'OrderedMultiDict[ParameterType, Parameter]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @info_final @@ -499,49 +515,61 @@ def __init__(self, type: 'ChunkType', length: 'int', flags: 'TBitFlags', error: class ShutdownChunk(Chunk): """Data model for SCTP SHUTDOWN chunks.""" + #: Raw chunk flags, kept as received. + flags: 'bytes' #: Cumulative TSN ack. cum_tsn_ack: 'int' if TYPE_CHECKING: - def __init__(self, type: 'ChunkType', length: 'int', cum_tsn_ack: 'int') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + def __init__(self, type: 'ChunkType', length: 'int', flags: 'bytes', cum_tsn_ack: 'int') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @info_final class ShutdownACKChunk(Chunk): """Data model for SCTP SHUTDOWN ACK chunks.""" + #: Raw chunk flags, kept as received. + flags: 'bytes' + if TYPE_CHECKING: - def __init__(self, type: 'ChunkType', length: 'int') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + def __init__(self, type: 'ChunkType', length: 'int', flags: 'bytes') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @info_final class ErrorChunk(Chunk): """Data model for SCTP ERROR chunks.""" + #: Raw chunk flags, kept as received. + flags: 'bytes' #: One or more error causes. error: 'OrderedMultiDict[CauseCode, ErrorCause]' if TYPE_CHECKING: - def __init__(self, type: 'ChunkType', length: 'int', error: 'OrderedMultiDict[CauseCode, ErrorCause]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + def __init__(self, type: 'ChunkType', length: 'int', flags: 'bytes', error: 'OrderedMultiDict[CauseCode, ErrorCause]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @info_final class CookieEchoChunk(Chunk): """Data model for SCTP COOKIE ECHO chunks.""" + #: Raw chunk flags, kept as received. + flags: 'bytes' #: State cookie, as received in the INIT ACK chunk's state cookie parameter. cookie: 'bytes' if TYPE_CHECKING: - def __init__(self, type: 'ChunkType', length: 'int', cookie: 'bytes') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + def __init__(self, type: 'ChunkType', length: 'int', flags: 'bytes', cookie: 'bytes') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @info_final class CookieACKChunk(Chunk): """Data model for SCTP COOKIE ACK chunks.""" + #: Raw chunk flags, kept as received. + flags: 'bytes' + if TYPE_CHECKING: - def __init__(self, type: 'ChunkType', length: 'int') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + def __init__(self, type: 'ChunkType', length: 'int', flags: 'bytes') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @info_final diff --git a/pcapkit/protocols/schema/transport/sctp.py b/pcapkit/protocols/schema/transport/sctp.py index b4de7b503d..91cd5a875d 100644 --- a/pcapkit/protocols/schema/transport/sctp.py +++ b/pcapkit/protocols/schema/transport/sctp.py @@ -59,6 +59,8 @@ class DATAChunkFlags(TypedDict): """SCTP DATA chunk flags.""" + #: Reserved bits. + reserved: int #: (I)mmediate bit, i.e., request a SACK chunk without delay. I: int #: (U)nordered bit, i.e., no stream sequence number is assigned. @@ -72,6 +74,8 @@ class TBitFlags(TypedDict): """SCTP chunk flags carrying only the T bit, i.e., ABORT and SHUTDOWN COMPLETE chunks.""" + #: Reserved bits. + reserved: int #: T bit, i.e., the verification tag has been reflected. T: int @@ -162,9 +166,10 @@ def nested_length(base: 'int') -> 'Callable[[dict[str, Any]], int]': one shared ``packet`` mapping down into the nested schemas, and each nested parameter overwrites ``packet['length']`` with *its* length, so a trailing field would size itself from the last parameter rather than from - the chunk. Folding avoids the question: the constructors always declare a - chunk length that already covers every parameter's padding, hence a - multiple of four, hence no chunk-level padding to emit. + the chunk. Folding avoids the question: every nested parameter packs its + own padding, the last one's included, so the list already ends on a + four-byte boundary whether or not the chunk length counts that final + padding. """ def callback(pkt: 'dict[str, Any]') -> 'int': @@ -276,12 +281,13 @@ class InvalidStreamIdentifierCause(ErrorCause, code=Enum_CauseCode.Invalid_Strea #: Stream identifier of the offending DATA chunk. stream_id: 'int' = UInt16Field() #: Reserved. - reserved: 'bytes' = PaddingField(length=2) + reserved: 'bytes' = BytesField(length=2) #: Padding. padding: 'bytes' = PaddingField(length=padding_length) if TYPE_CHECKING: - def __init__(self, code: 'Enum_CauseCode', length: 'int', stream_id: 'int') -> 'None': ... + def __init__(self, code: 'Enum_CauseCode', length: 'int', stream_id: 'int', + reserved: 'bytes') -> 'None': ... @schema_final @@ -620,6 +626,7 @@ class DATAChunk(Chunk, code=Enum_Chunk.Payload_Data): #: Chunk flags. flags: 'DATAChunkFlags' = BitField(length=1, namespace={ + 'reserved': (0, 4), 'I': (4, 1), 'U': (5, 1), 'B': (6, 1), @@ -782,6 +789,7 @@ class AbortChunk(Chunk, code=Enum_Chunk.Abort): #: Chunk flags. flags: 'TBitFlags' = BitField(length=1, namespace={ + 'reserved': (0, 7), 'T': (7, 1), }) #: Zero or more error causes, including the chunk's own trailing padding; @@ -872,6 +880,7 @@ class ShutdownCompleteChunk(Chunk, code=Enum_Chunk.Shutdown_Complete): #: Chunk flags. flags: 'TBitFlags' = BitField(length=1, namespace={ + 'reserved': (0, 7), 'T': (7, 1), }) #: Padding. diff --git a/pcapkit/protocols/transport/sctp.py b/pcapkit/protocols/transport/sctp.py index fb0fea6db6..fc65f8f1d9 100644 --- a/pcapkit/protocols/transport/sctp.py +++ b/pcapkit/protocols/transport/sctp.py @@ -161,7 +161,7 @@ if TYPE_CHECKING: from ipaddress import IPv4Address, IPv6Address - from typing import Any, Callable, DefaultDict, Optional, Type + from typing import Any, Callable, DefaultDict, Optional, Sequence, Type from mypy_extensions import DefaultArg, KwArg, NamedArg from typing_extensions import Literal @@ -1060,6 +1060,34 @@ def _make_sctp_cause(self, code: 'Enum_CauseCode', cause: 'Optional[Data_ErrorCa meth = name[1] return meth(code, cause, **kwargs) + def _make_nested_length(self, base: 'int', items: 'Sequence[Schema_Parameter | Schema_ErrorCause | bytes]', # pylint: disable=line-too-long + declared: 'Optional[int]' = None) -> 'int': + """Compute the length of a chunk carrying a nested parameter or error + cause list. + + Per :rfc:`9260#section-3.2`, the chunk length counts the padding of + every nested item except the last. The RFC also asks receivers to + accept a length that counts the last item's padding too, so a parsed + chunk declaring that longer form keeps it. + + Args: + base: size of the chunk's fixed-length fields, including the + four-byte chunk header + items: constructed nested items; a raw :obj:`bytes` item is + counted as given + declared: chunk length declared by the parsed chunk being + rebuilt, if any + + Returns: + Chunk length. + + """ + sizes = [len(item) if isinstance(item, bytes) else len(item.pack()) for item in items] + padded = base + sum(sizes) + if not items or isinstance(items[-1], bytes) or declared == padded: + return padded + return padded - max(sizes[-1] - items[-1].length, 0) + def _read_chunk_donone(self, schema: 'Schema_UnknownChunk', *, chunks: 'Chunks') -> 'Data_UnknownChunk': # pylint: disable=unused-argument """Read SCTP chunk of an unsupported type. @@ -1127,6 +1155,7 @@ def _read_chunk_data(self, schema: 'Schema_DATAChunk', *, chunks: 'Chunks') -> ' type=schema.type, length=schema.length, flags=Data_DATAChunkFlags( + reserved=schema.flags['reserved'], I=bool(schema.flags['I']), U=bool(schema.flags['U']), B=bool(schema.flags['B']), @@ -1174,10 +1203,6 @@ def _read_chunk_init(self, schema: 'Schema_INITChunk', *, chunks: 'Chunks') -> ' Raises: ProtocolError: If ``length`` is **NOT** at least ``20``. - Note: - The chunk flags are reserved by :rfc:`9260#section-3.3.2` and are - therefore not exposed on the data model. - """ if schema.length < 20: raise ProtocolError(f'{self.alias}: [Chunk {schema.type}] invalid format') @@ -1185,6 +1210,7 @@ def _read_chunk_init(self, schema: 'Schema_INITChunk', *, chunks: 'Chunks') -> ' return Data_INITChunk( type=schema.type, length=schema.length, + flags=schema.flags, init_tag=schema.init_tag, a_rwnd=schema.a_rwnd, outbound_streams=schema.outbound_streams, @@ -1235,6 +1261,7 @@ def _read_chunk_init_ack(self, schema: 'Schema_INITACKChunk', *, chunks: 'Chunks return Data_INITACKChunk( type=schema.type, length=schema.length, + flags=schema.flags, init_tag=schema.init_tag, a_rwnd=schema.a_rwnd, outbound_streams=schema.outbound_streams, @@ -1288,6 +1315,7 @@ def _read_chunk_sack(self, schema: 'Schema_SACKChunk', *, chunks: 'Chunks') -> ' return Data_SACKChunk( type=schema.type, length=schema.length, + flags=schema.flags, cum_tsn_ack=schema.cum_tsn_ack, a_rwnd=schema.a_rwnd, num_gap_blocks=schema.num_gap_blocks, @@ -1340,6 +1368,7 @@ def _read_chunk_heartbeat(self, schema: 'Schema_HeartbeatChunk', *, chunks: 'Chu return Data_HeartbeatChunk( type=schema.type, length=schema.length, + flags=schema.flags, parameters=self._read_sctp_parameters(schema.parameters), ) @@ -1377,6 +1406,7 @@ def _read_chunk_heartbeat_ack(self, schema: 'Schema_HeartbeatACKChunk', *, chunk return Data_HeartbeatACKChunk( type=schema.type, length=schema.length, + flags=schema.flags, parameters=self._read_sctp_parameters(schema.parameters), ) @@ -1415,6 +1445,7 @@ def _read_chunk_abort(self, schema: 'Schema_AbortChunk', *, chunks: 'Chunks') -> type=schema.type, length=schema.length, flags=Data_TBitFlags( + reserved=schema.flags['reserved'], T=bool(schema.flags['T']), ), error=self._read_sctp_causes(schema.error), @@ -1452,6 +1483,7 @@ def _read_chunk_shutdown(self, schema: 'Schema_ShutdownChunk', *, chunks: 'Chunk return Data_ShutdownChunk( type=schema.type, length=schema.length, + flags=schema.flags, cum_tsn_ack=schema.cum_tsn_ack, ) @@ -1485,6 +1517,7 @@ def _read_chunk_shutdown_ack(self, schema: 'Schema_ShutdownACKChunk', *, chunks: return Data_ShutdownACKChunk( type=schema.type, length=schema.length, + flags=schema.flags, ) def _read_chunk_error(self, schema: 'Schema_ErrorChunk', *, chunks: 'Chunks') -> 'Data_ErrorChunk': # pylint: disable=unused-argument @@ -1521,6 +1554,7 @@ def _read_chunk_error(self, schema: 'Schema_ErrorChunk', *, chunks: 'Chunks') -> return Data_ErrorChunk( type=schema.type, length=schema.length, + flags=schema.flags, error=self._read_sctp_causes(schema.error), ) @@ -1563,6 +1597,7 @@ def _read_chunk_cookie_echo(self, schema: 'Schema_CookieEchoChunk', *, chunks: ' return Data_CookieEchoChunk( type=schema.type, length=schema.length, + flags=schema.flags, cookie=schema.cookie, ) @@ -1596,6 +1631,7 @@ def _read_chunk_cookie_ack(self, schema: 'Schema_CookieACKChunk', *, chunks: 'Ch return Data_CookieACKChunk( type=schema.type, length=schema.length, + flags=schema.flags, ) def _read_chunk_shutdown_complete(self, schema: 'Schema_ShutdownCompleteChunk', *, chunks: 'Chunks') -> 'Data_ShutdownCompleteChunk': # pylint: disable=unused-argument @@ -1629,6 +1665,7 @@ def _read_chunk_shutdown_complete(self, schema: 'Schema_ShutdownCompleteChunk', type=schema.type, length=schema.length, flags=Data_TBitFlags( + reserved=schema.flags['reserved'], T=bool(schema.flags['T']), ), ) @@ -1668,6 +1705,7 @@ def _make_chunk_donone(self, code: 'Enum_Chunk', chunk: 'Optional[Data_UnknownCh ) def _make_chunk_data(self, code: 'Enum_Chunk', chunk: 'Optional[Data_DATAChunk]' = None, *, + reserved: 'int' = 0, I: 'bool' = False, # noqa: E741 U: 'bool' = False, B: 'bool' = True, @@ -1683,6 +1721,7 @@ def _make_chunk_data(self, code: 'Enum_Chunk', chunk: 'Optional[Data_DATAChunk]' Args: code: chunk type chunk: chunk data + reserved: reserved flag bits I: immediate bit U: unordered bit B: beginning fragment bit @@ -1703,6 +1742,7 @@ def _make_chunk_data(self, code: 'Enum_Chunk', chunk: 'Optional[Data_DATAChunk]' """ if chunk is not None: + reserved = chunk.flags.reserved I = chunk.flags.I # noqa: E741 U = chunk.flags.U B = chunk.flags.B @@ -1719,6 +1759,7 @@ def _make_chunk_data(self, code: 'Enum_Chunk', chunk: 'Optional[Data_DATAChunk]' return Schema_DATAChunk( type=code, flags={ + 'reserved': reserved, 'I': int(I), 'U': int(U), 'B': int(B), @@ -1733,6 +1774,7 @@ def _make_chunk_data(self, code: 'Enum_Chunk', chunk: 'Optional[Data_DATAChunk]' ) def _make_chunk_init(self, code: 'Enum_Chunk', chunk: 'Optional[Data_INITChunk]' = None, *, + flags: 'bytes' = b'\x00', init_tag: 'int' = 0, a_rwnd: 'int' = 1500, # minimum permitted by [RFC 9260] outbound_streams: 'int' = 1, @@ -1745,6 +1787,7 @@ def _make_chunk_init(self, code: 'Enum_Chunk', chunk: 'Optional[Data_INITChunk]' Args: code: chunk type chunk: chunk data + flags: raw chunk flags, as a single byte init_tag: initiate tag a_rwnd: advertised receiver window credit outbound_streams: number of outbound streams @@ -1756,12 +1799,16 @@ def _make_chunk_init(self, code: 'Enum_Chunk', chunk: 'Optional[Data_INITChunk]' Returns: Constructed chunk schema. + Raises: + ProtocolError: If ``flags`` is **NOT** exactly one byte. + Note: - The chunk flags are reserved by :rfc:`9260#section-3.3.2` and are - always emitted as zero. + The chunk flags are reserved by :rfc:`9260#section-3.3.2`, so they + default to zero; a parsed chunk's flags are kept as received. """ if chunk is not None: + flags = chunk.flags init_tag = chunk.init_tag a_rwnd = chunk.a_rwnd outbound_streams = chunk.outbound_streams @@ -1769,16 +1816,19 @@ def _make_chunk_init(self, code: 'Enum_Chunk', chunk: 'Optional[Data_INITChunk]' init_tsn = chunk.init_tsn parameters = chunk.parameters + if len(flags) != 1: + raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid format') + if parameters is not None: parameters_value = self._make_sctp_parameters(parameters) else: parameters_value = [] - length = 20 + sum(len(param) if isinstance(param, bytes) else len(param.pack()) - for param in parameters_value) + length = self._make_nested_length(20, parameters_value, + None if chunk is None else chunk.length) return Schema_INITChunk( type=code, - flags=b'\x00', + flags=flags, length=length, init_tag=init_tag, a_rwnd=a_rwnd, @@ -1789,6 +1839,7 @@ def _make_chunk_init(self, code: 'Enum_Chunk', chunk: 'Optional[Data_INITChunk]' ) def _make_chunk_init_ack(self, code: 'Enum_Chunk', chunk: 'Optional[Data_INITACKChunk]' = None, *, + flags: 'bytes' = b'\x00', init_tag: 'int' = 0, a_rwnd: 'int' = 1500, # minimum permitted by [RFC 9260] outbound_streams: 'int' = 1, @@ -1801,6 +1852,7 @@ def _make_chunk_init_ack(self, code: 'Enum_Chunk', chunk: 'Optional[Data_INITACK Args: code: chunk type chunk: chunk data + flags: raw chunk flags, as a single byte init_tag: initiate tag a_rwnd: advertised receiver window credit outbound_streams: number of outbound streams @@ -1812,8 +1864,12 @@ def _make_chunk_init_ack(self, code: 'Enum_Chunk', chunk: 'Optional[Data_INITACK Returns: Constructed chunk schema. + Raises: + ProtocolError: If ``flags`` is **NOT** exactly one byte. + """ if chunk is not None: + flags = chunk.flags init_tag = chunk.init_tag a_rwnd = chunk.a_rwnd outbound_streams = chunk.outbound_streams @@ -1821,16 +1877,19 @@ def _make_chunk_init_ack(self, code: 'Enum_Chunk', chunk: 'Optional[Data_INITACK init_tsn = chunk.init_tsn parameters = chunk.parameters + if len(flags) != 1: + raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid format') + if parameters is not None: parameters_value = self._make_sctp_parameters(parameters) else: parameters_value = [] - length = 20 + sum(len(param) if isinstance(param, bytes) else len(param.pack()) - for param in parameters_value) + length = self._make_nested_length(20, parameters_value, + None if chunk is None else chunk.length) return Schema_INITACKChunk( type=code, - flags=b'\x00', + flags=flags, length=length, init_tag=init_tag, a_rwnd=a_rwnd, @@ -1841,6 +1900,7 @@ def _make_chunk_init_ack(self, code: 'Enum_Chunk', chunk: 'Optional[Data_INITACK ) def _make_chunk_sack(self, code: 'Enum_Chunk', chunk: 'Optional[Data_SACKChunk]' = None, *, + flags: 'bytes' = b'\x00', cum_tsn_ack: 'int' = 0, a_rwnd: 'int' = 1500, # minimum permitted by [RFC 9260] gap_blocks: 'Optional[list[Schema_GapAckBlock | Data_GapAckBlock | tuple[int, int]]]' = None, # pylint: disable=line-too-long @@ -1851,6 +1911,7 @@ def _make_chunk_sack(self, code: 'Enum_Chunk', chunk: 'Optional[Data_SACKChunk]' Args: code: chunk type chunk: chunk data + flags: raw chunk flags, as a single byte cum_tsn_ack: cumulative TSN ack a_rwnd: advertised receiver window credit gap_blocks: gap ack blocks, each as a schema, a data model or a @@ -1861,6 +1922,9 @@ def _make_chunk_sack(self, code: 'Enum_Chunk', chunk: 'Optional[Data_SACKChunk]' Returns: Constructed chunk schema. + Raises: + ProtocolError: If ``flags`` is **NOT** exactly one byte. + Note: The counts of gap ack blocks and duplicate TSNs are derived from the supplied lists rather than taken as arguments, so that they cannot @@ -1868,11 +1932,15 @@ def _make_chunk_sack(self, code: 'Enum_Chunk', chunk: 'Optional[Data_SACKChunk]' """ if chunk is not None: + flags = chunk.flags cum_tsn_ack = chunk.cum_tsn_ack a_rwnd = chunk.a_rwnd gap_blocks = list(chunk.gap_blocks) dup_tsn = list(chunk.dup_tsn) + if len(flags) != 1: + raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid format') + blocks = [] # type: list[Schema_GapAckBlock] for block in gap_blocks or []: if isinstance(block, Schema_GapAckBlock): @@ -1886,7 +1954,7 @@ def _make_chunk_sack(self, code: 'Enum_Chunk', chunk: 'Optional[Data_SACKChunk]' return Schema_SACKChunk( type=code, - flags=b'\x00', + flags=flags, length=16 + len(blocks) * 4 + len(tsn_list) * 4, cum_tsn_ack=cum_tsn_ack, a_rwnd=a_rwnd, @@ -1897,6 +1965,7 @@ def _make_chunk_sack(self, code: 'Enum_Chunk', chunk: 'Optional[Data_SACKChunk]' ) def _make_chunk_heartbeat(self, code: 'Enum_Chunk', chunk: 'Optional[Data_HeartbeatChunk]' = None, *, + flags: 'bytes' = b'\x00', parameters: 'Optional[list[Schema_Parameter | tuple[Enum_Parameter, dict[str, Any]] | bytes] | Parameters]' = None, # pylint: disable=line-too-long **kwargs: 'Any') -> 'Schema_HeartbeatChunk': """Make SCTP HEARTBEAT chunk. @@ -1904,31 +1973,40 @@ def _make_chunk_heartbeat(self, code: 'Enum_Chunk', chunk: 'Optional[Data_Heartb Args: code: chunk type chunk: chunk data + flags: raw chunk flags, as a single byte parameters: heartbeat information parameters **kwargs: arbitrary keyword arguments Returns: Constructed chunk schema. + Raises: + ProtocolError: If ``flags`` is **NOT** exactly one byte. + """ if chunk is not None: + flags = chunk.flags parameters = chunk.parameters + if len(flags) != 1: + raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid format') + if parameters is not None: parameters_value = self._make_sctp_parameters(parameters) else: parameters_value = [] - length = 4 + sum(len(param) if isinstance(param, bytes) else len(param.pack()) - for param in parameters_value) + length = self._make_nested_length(4, parameters_value, + None if chunk is None else chunk.length) return Schema_HeartbeatChunk( type=code, - flags=b'\x00', + flags=flags, length=length, parameters=parameters_value, ) def _make_chunk_heartbeat_ack(self, code: 'Enum_Chunk', chunk: 'Optional[Data_HeartbeatACKChunk]' = None, *, + flags: 'bytes' = b'\x00', parameters: 'Optional[list[Schema_Parameter | tuple[Enum_Parameter, dict[str, Any]] | bytes] | Parameters]' = None, # pylint: disable=line-too-long **kwargs: 'Any') -> 'Schema_HeartbeatACKChunk': """Make SCTP HEARTBEAT ACK chunk. @@ -1936,31 +2014,40 @@ def _make_chunk_heartbeat_ack(self, code: 'Enum_Chunk', chunk: 'Optional[Data_He Args: code: chunk type chunk: chunk data + flags: raw chunk flags, as a single byte parameters: heartbeat information parameters **kwargs: arbitrary keyword arguments Returns: Constructed chunk schema. + Raises: + ProtocolError: If ``flags`` is **NOT** exactly one byte. + """ if chunk is not None: + flags = chunk.flags parameters = chunk.parameters + if len(flags) != 1: + raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid format') + if parameters is not None: parameters_value = self._make_sctp_parameters(parameters) else: parameters_value = [] - length = 4 + sum(len(param) if isinstance(param, bytes) else len(param.pack()) - for param in parameters_value) + length = self._make_nested_length(4, parameters_value, + None if chunk is None else chunk.length) return Schema_HeartbeatACKChunk( type=code, - flags=b'\x00', + flags=flags, length=length, parameters=parameters_value, ) def _make_chunk_abort(self, code: 'Enum_Chunk', chunk: 'Optional[Data_AbortChunk]' = None, *, + reserved: 'int' = 0, T: 'bool' = False, error: 'Optional[list[Schema_ErrorCause | tuple[Enum_CauseCode, dict[str, Any]] | bytes] | Causes]' = None, # pylint: disable=line-too-long **kwargs: 'Any') -> 'Schema_AbortChunk': @@ -1969,6 +2056,7 @@ def _make_chunk_abort(self, code: 'Enum_Chunk', chunk: 'Optional[Data_AbortChunk Args: code: chunk type chunk: chunk data + reserved: reserved flag bits T: whether the verification tag has been reflected error: zero or more error causes **kwargs: arbitrary keyword arguments @@ -1978,6 +2066,7 @@ def _make_chunk_abort(self, code: 'Enum_Chunk', chunk: 'Optional[Data_AbortChunk """ if chunk is not None: + reserved = chunk.flags.reserved T = chunk.flags.T error = chunk.error @@ -1985,12 +2074,13 @@ def _make_chunk_abort(self, code: 'Enum_Chunk', chunk: 'Optional[Data_AbortChunk error_value = self._make_sctp_causes(error) else: error_value = [] - length = 4 + sum(len(cause) if isinstance(cause, bytes) else len(cause.pack()) - for cause in error_value) + length = self._make_nested_length(4, error_value, + None if chunk is None else chunk.length) return Schema_AbortChunk( type=code, flags={ + 'reserved': reserved, 'T': int(T), }, length=length, @@ -1998,6 +2088,7 @@ def _make_chunk_abort(self, code: 'Enum_Chunk', chunk: 'Optional[Data_AbortChunk ) def _make_chunk_shutdown(self, code: 'Enum_Chunk', chunk: 'Optional[Data_ShutdownChunk]' = None, *, + flags: 'bytes' = b'\x00', cum_tsn_ack: 'int' = 0, **kwargs: 'Any') -> 'Schema_ShutdownChunk': """Make SCTP SHUTDOWN chunk. @@ -2005,43 +2096,63 @@ def _make_chunk_shutdown(self, code: 'Enum_Chunk', chunk: 'Optional[Data_Shutdow Args: code: chunk type chunk: chunk data + flags: raw chunk flags, as a single byte cum_tsn_ack: cumulative TSN ack **kwargs: arbitrary keyword arguments Returns: Constructed chunk schema. + Raises: + ProtocolError: If ``flags`` is **NOT** exactly one byte. + """ if chunk is not None: + flags = chunk.flags cum_tsn_ack = chunk.cum_tsn_ack + if len(flags) != 1: + raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid format') + return Schema_ShutdownChunk( type=code, - flags=b'\x00', + flags=flags, length=8, cum_tsn_ack=cum_tsn_ack, ) - def _make_chunk_shutdown_ack(self, code: 'Enum_Chunk', chunk: 'Optional[Data_ShutdownACKChunk]' = None, + def _make_chunk_shutdown_ack(self, code: 'Enum_Chunk', chunk: 'Optional[Data_ShutdownACKChunk]' = None, *, + flags: 'bytes' = b'\x00', **kwargs: 'Any') -> 'Schema_ShutdownACKChunk': """Make SCTP SHUTDOWN ACK chunk. Args: code: chunk type chunk: chunk data + flags: raw chunk flags, as a single byte **kwargs: arbitrary keyword arguments Returns: Constructed chunk schema. + Raises: + ProtocolError: If ``flags`` is **NOT** exactly one byte. + """ + if chunk is not None: + flags = chunk.flags + + if len(flags) != 1: + raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid format') + return Schema_ShutdownACKChunk( type=code, - flags=b'\x00', + flags=flags, length=4, ) def _make_chunk_error(self, code: 'Enum_Chunk', chunk: 'Optional[Data_ErrorChunk]' = None, *, + flags: 'bytes' = b'\x00', error: 'Optional[list[Schema_ErrorCause | tuple[Enum_CauseCode, dict[str, Any]] | bytes] | Causes]' = None, # pylint: disable=line-too-long **kwargs: 'Any') -> 'Schema_ErrorChunk': """Make SCTP ERROR chunk. @@ -2049,31 +2160,40 @@ def _make_chunk_error(self, code: 'Enum_Chunk', chunk: 'Optional[Data_ErrorChunk Args: code: chunk type chunk: chunk data + flags: raw chunk flags, as a single byte error: one or more error causes **kwargs: arbitrary keyword arguments Returns: Constructed chunk schema. + Raises: + ProtocolError: If ``flags`` is **NOT** exactly one byte. + """ if chunk is not None: + flags = chunk.flags error = chunk.error + if len(flags) != 1: + raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid format') + if error is not None: error_value = self._make_sctp_causes(error) else: error_value = [] - length = 4 + sum(len(cause) if isinstance(cause, bytes) else len(cause.pack()) - for cause in error_value) + length = self._make_nested_length(4, error_value, + None if chunk is None else chunk.length) return Schema_ErrorChunk( type=code, - flags=b'\x00', + flags=flags, length=length, error=error_value, ) def _make_chunk_cookie_echo(self, code: 'Enum_Chunk', chunk: 'Optional[Data_CookieEchoChunk]' = None, *, + flags: 'bytes' = b'\x00', cookie: 'bytes' = b'', **kwargs: 'Any') -> 'Schema_CookieEchoChunk': """Make SCTP COOKIE ECHO chunk. @@ -2081,6 +2201,7 @@ def _make_chunk_cookie_echo(self, code: 'Enum_Chunk', chunk: 'Optional[Data_Cook Args: code: chunk type chunk: chunk data + flags: raw chunk flags, as a single byte cookie: state cookie, as received in the INIT ACK chunk's state cookie parameter **kwargs: arbitrary keyword arguments @@ -2088,37 +2209,56 @@ def _make_chunk_cookie_echo(self, code: 'Enum_Chunk', chunk: 'Optional[Data_Cook Returns: Constructed chunk schema. + Raises: + ProtocolError: If ``flags`` is **NOT** exactly one byte. + """ if chunk is not None: + flags = chunk.flags cookie = chunk.cookie + if len(flags) != 1: + raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid format') + return Schema_CookieEchoChunk( type=code, - flags=b'\x00', + flags=flags, length=len(cookie) + 4, cookie=cookie, ) - def _make_chunk_cookie_ack(self, code: 'Enum_Chunk', chunk: 'Optional[Data_CookieACKChunk]' = None, + def _make_chunk_cookie_ack(self, code: 'Enum_Chunk', chunk: 'Optional[Data_CookieACKChunk]' = None, *, + flags: 'bytes' = b'\x00', **kwargs: 'Any') -> 'Schema_CookieACKChunk': """Make SCTP COOKIE ACK chunk. Args: code: chunk type chunk: chunk data + flags: raw chunk flags, as a single byte **kwargs: arbitrary keyword arguments Returns: Constructed chunk schema. + Raises: + ProtocolError: If ``flags`` is **NOT** exactly one byte. + """ + if chunk is not None: + flags = chunk.flags + + if len(flags) != 1: + raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid format') + return Schema_CookieACKChunk( type=code, - flags=b'\x00', + flags=flags, length=4, ) def _make_chunk_shutdown_complete(self, code: 'Enum_Chunk', chunk: 'Optional[Data_ShutdownCompleteChunk]' = None, *, + reserved: 'int' = 0, T: 'bool' = False, **kwargs: 'Any') -> 'Schema_ShutdownCompleteChunk': """Make SCTP SHUTDOWN COMPLETE chunk. @@ -2126,6 +2266,7 @@ def _make_chunk_shutdown_complete(self, code: 'Enum_Chunk', chunk: 'Optional[Dat Args: code: chunk type chunk: chunk data + reserved: reserved flag bits T: whether the verification tag has been reflected **kwargs: arbitrary keyword arguments @@ -2134,11 +2275,13 @@ def _make_chunk_shutdown_complete(self, code: 'Enum_Chunk', chunk: 'Optional[Dat """ if chunk is not None: + reserved = chunk.flags.reserved T = chunk.flags.T return Schema_ShutdownCompleteChunk( type=code, flags={ + 'reserved': reserved, 'T': int(T), }, length=4, @@ -2748,6 +2891,7 @@ def _read_cause_invalid_stream(self, schema: 'Schema_InvalidStreamIdentifierCaus code=schema.code, length=schema.length, stream_id=schema.stream_id, + reserved=schema.reserved, ) def _read_cause_missing_param(self, schema: 'Schema_MissingMandatoryParameterCause', *, causes: 'Causes') -> 'Data_MissingMandatoryParameterCause': # pylint: disable=unused-argument @@ -3197,6 +3341,7 @@ def _make_cause_donone(self, code: 'Enum_CauseCode', cause: 'Optional[Data_Unkno def _make_cause_invalid_stream(self, code: 'Enum_CauseCode', cause: 'Optional[Data_InvalidStreamIdentifierCause]' = None, *, stream_id: 'int' = 0, + reserved: 'bytes' = b'\x00\x00', **kwargs: 'Any') -> 'Schema_InvalidStreamIdentifierCause': """Make SCTP invalid stream identifier error cause. @@ -3204,19 +3349,28 @@ def _make_cause_invalid_stream(self, code: 'Enum_CauseCode', cause: 'Optional[Da code: error cause code cause: error cause data stream_id: stream identifier of the offending DATA chunk + reserved: reserved field, as two bytes **kwargs: arbitrary keyword arguments Returns: Constructed error cause schema. + Raises: + ProtocolError: If ``reserved`` is **NOT** exactly two bytes. + """ if cause is not None: stream_id = cause.stream_id + reserved = cause.reserved + + if len(reserved) != 2: + raise ProtocolError(f'{self.alias}: [Cause {code}] invalid format') return Schema_InvalidStreamIdentifierCause( code=code, length=8, stream_id=stream_id, + reserved=reserved, ) def _make_cause_missing_param(self, code: 'Enum_CauseCode', cause: 'Optional[Data_MissingMandatoryParameterCause]' = None, *, diff --git a/tests/protocols/transport/test_sctp_roundtrip_unit.py b/tests/protocols/transport/test_sctp_roundtrip_unit.py new file mode 100644 index 0000000000..455f9d0da8 --- /dev/null +++ b/tests/protocols/transport/test_sctp_roundtrip_unit.py @@ -0,0 +1,118 @@ +"""#1221 and #1222 -- SCTP chunks rebuild byte-exactly through ``from_data``. + +#1221: the chunk makers wrote the flags octet back as zero, the DATA and T-bit +flag schemas mapped only the defined bits, and the Invalid Stream Identifier +cause wrote its reserved field back as zero. The raw octets are now kept. + +#1222: :rfc:`9260#section-3.2` excludes the last parameter's (or cause's) +padding from the chunk length, but the makers counted it. The canonical length +is now computed, and a parsed chunk that counted the final padding keeps it. + +""" +from __future__ import annotations + +import importlib.util +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. +HEADER = '138813891122334400000000' + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class SCTPRoundTripUnitTests(unittest.TestCase): + def setUp(self) -> None: + reimport_once_per_class(self) + + def assertRoundTrip(self, hexdata: str) -> None: + from pcapkit.protocols.transport.sctp import SCTP + + raw = bytes.fromhex(hexdata) + self.assertEqual(SCTP.from_data(SCTP(raw).info).data.hex(), raw.hex()) + + def test_reserved_chunk_flags_round_trip(self) -> None: + """#1221: every chunk with reserved flags keeps the raw flags octet.""" + cases = { + 'INIT': '13881389112233445d2a322901a50014aabbccdd00010000000a000a00000001', + 'SACK': HEADER + '035a0010' + '00000001' + '00010000' + '00000000', + 'HEARTBEAT': HEADER + '04110008' + '00010004', + 'HEARTBEAT ACK': HEADER + '05220008' + '00010004', + 'SHUTDOWN': HEADER + '07010008' + '00000001', + 'SHUTDOWN ACK': HEADER + '08800004', + 'ERROR': HEADER + '09770004', + 'COOKIE ECHO': HEADER + '0a020008' + 'deadbeef', + 'COOKIE ACK': HEADER + '0b400004', + } + for name, hexdata in cases.items(): + with self.subTest(chunk=name): + self.assertRoundTrip(hexdata) + + def test_reserved_flag_bits_round_trip(self) -> None: + """#1221: DATA, ABORT and SHUTDOWN COMPLETE keep their reserved flag bits.""" + from pcapkit.protocols.transport.sctp import SCTP + + cases = { + 'DATA': (HEADER + '00f30011' + '00000001' + '00010002' + '00000000' + 'ab000000', 0xf), + 'ABORT': (HEADER + '06fe0004', 0x7f), + 'SHUTDOWN COMPLETE': (HEADER + '0eff0004', 0x7f), + } + for name, (hexdata, reserved) in cases.items(): + with self.subTest(chunk=name): + self.assertRoundTrip(hexdata) + chunk = next(iter(SCTP(bytes.fromhex(hexdata)).info.chunks.values())) + self.assertEqual(chunk.flags.reserved, reserved) + + def test_invalid_stream_cause_reserved_round_trip(self) -> None: + """#1221: the Invalid Stream Identifier cause keeps its reserved field.""" + from pcapkit.protocols.transport.sctp import SCTP + + hexdata = HEADER + '0600000c' + '00010008' + '0005beef' + self.assertRoundTrip(hexdata) + chunk = next(iter(SCTP(bytes.fromhex(hexdata)).info.chunks.values())) + self.assertEqual(next(iter(chunk.error.values())).reserved, b'\xbe\xef') + + def test_make_defaults_flags_to_zero(self) -> None: + """#1221: a chunk built from keywords still emits zero flags.""" + from pcapkit.const.sctp.chunk import Chunk + from pcapkit.protocols.transport.sctp import SCTP + + data = SCTP(srcport=1, dstport=2, chunks=[(Chunk.Initiation, {})]).data + self.assertEqual(data[13], 0) + + def test_chunk_length_excludes_final_padding(self) -> None: + """#1222: the final parameter's or cause's padding is not counted.""" + cases = { + 'INIT': '13881389112233445c42009a0100001baabbccdd00010000000a000a00000001000b000761620000', + 'INIT ACK': HEADER + '0200001b' + 'aabbccdd00010000000a000a00000001' + '000b000761620000', + 'HEARTBEAT': HEADER + '0400000b' + '00010007' + '61626300', + 'HEARTBEAT ACK': HEADER + '0500000b' + '00010007' + '61626300', + 'ABORT': HEADER + '0600000b' + '000c0007' + '61626300', + 'ERROR': HEADER + '0900000b' + '000c0007' + '61626300', + } + for name, hexdata in cases.items(): + with self.subTest(chunk=name): + self.assertRoundTrip(hexdata) + + def test_chunk_length_counting_final_padding_is_kept(self) -> None: + """#1222: a parsed length that counts the final padding, which RFC 9260 allows, is kept.""" + self.assertRoundTrip( + '13881389112233445c42009a0100001caabbccdd00010000000a000a00000001000b000761620000') + + def test_make_emits_canonical_length(self) -> None: + """#1222: ``make()`` declares the RFC 9260 length for an unpadded final parameter.""" + from pcapkit.const.sctp.chunk import Chunk + from pcapkit.const.sctp.parameter import Parameter + from pcapkit.protocols.transport.sctp import SCTP + + data = SCTP(srcport=1, dstport=2, chunks=[ + (Chunk.Heartbeat_Request, {'parameters': [(Parameter.Heartbeat_Info, {'info': b'abc'})]}), + ]).data + self.assertEqual(data[12:].hex(), '0400000b' + '00010007' + '61626300') + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/transport/test_sctp_unit.py b/tests/protocols/transport/test_sctp_unit.py index 7079c6de71..7f97baebfb 100644 --- a/tests/protocols/transport/test_sctp_unit.py +++ b/tests/protocols/transport/test_sctp_unit.py @@ -267,7 +267,8 @@ def test_every_chunk_type_round_trips(self) -> None: dict(init_tag=0x99AABBCC, a_rwnd=4660, outbound_streams=3, inbound_streams=4, init_tsn=7, parameters=[(Parameter.State_Cookie, {'cookie': b'\xde\xad\xbe\xef\x01'})]), - dict(length=32, init_tag=0x99AABBCC, a_rwnd=4660, outbound_streams=3, + # RFC 9260 section 3.2: the final parameter's padding is not counted. + dict(length=29, init_tag=0x99AABBCC, a_rwnd=4660, outbound_streams=3, inbound_streams=4, init_tsn=7)), (Chunk.Selective_Acknowledgement, dict(cum_tsn_ack=12, a_rwnd=4660, gap_blocks=[(2, 3), (5, 5)], @@ -282,7 +283,7 @@ def test_every_chunk_type_round_trips(self) -> None: dict(length=12)), (Chunk.Abort, dict(T=True, error=[(CauseCode.User_Initiated_Abort, {'info': b'bye'})]), - dict(length=12)), + dict(length=11)), (Chunk.Shutdown, dict(cum_tsn_ack=0x0A0B0C10), dict(length=8, cum_tsn_ack=0x0A0B0C10)),