Skip to content

fix(sctp): chunk flags, reserved fields and final padding round-trip #1253

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JarryShaw merged 1 commit into from
Oct 7, 2026
Merged

fix(sctp): chunk flags, reserved fields and final padding round-trip #1253

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
54 changes: 41 additions & 13 deletions pcapkit/protocols/data/transport/sctp.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -62,19 +64,21 @@ 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
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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand All @@ -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.
Expand All @@ -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.
Expand All @@ -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
Expand All @@ -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
Expand Down
19 changes: 14 additions & 5 deletions pcapkit/protocols/schema/transport/sctp.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -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

Expand Down Expand Up @@ -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':
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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),
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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.
Expand Down
Loading
Loading