# -*- coding: utf-8 -*-
# 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
from pcapkit.const.reg.apptype import TransportProtocol as Enum_TransportProtocol
from pcapkit.const.sctp.cause_code import CauseCode as Enum_CauseCode
from pcapkit.const.sctp.chunk import Chunk as Enum_Chunk
from pcapkit.const.sctp.parameter import Parameter as Enum_Parameter
from pcapkit.const.sctp.payload_protocol_identifier import \
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 PayloadField, SchemaField
from pcapkit.corekit.fields.numbers import EnumField, PortEnumField, UInt16Field, UInt32Field
from pcapkit.corekit.fields.strings import BitField, BytesField, PaddingField
from pcapkit.protocols.schema.schema import EnumSchema, Schema, schema_final
from pcapkit.utilities.logging import SPHINX_TYPE_CHECKING
__all__ = [
'SCTP',
'Chunk',
'UnknownChunk', 'DATAChunk', 'INITChunk', 'INITACKChunk', 'SACKChunk',
'HeartbeatChunk', 'HeartbeatACKChunk', 'AbortChunk', 'ShutdownChunk',
'ShutdownACKChunk', 'ErrorChunk', 'CookieEchoChunk', 'CookieACKChunk',
'ShutdownCompleteChunk',
'GapAckBlock',
'Parameter',
'UnknownParameter', 'HeartbeatInfoParameter', 'IPv4AddressParameter',
'IPv6AddressParameter', 'StateCookieParameter', 'UnrecognizedParameter',
'CookiePreservativeParameter', 'HostNameAddressParameter',
'SupportedAddressTypesParameter',
'ErrorCause',
'UnknownCause', 'InvalidStreamIdentifierCause', 'MissingMandatoryParameterCause',
'StaleCookieCause', 'OutOfResourceCause', 'UnresolvableAddressCause',
'UnrecognizedChunkTypeCause', 'InvalidMandatoryParameterCause',
'UnrecognizedParametersCause', 'NoUserDataCause',
'CookieReceivedWhileShuttingDownCause',
'RestartOfAnAssociationWithNewAddressesCause', 'UserInitiatedAbortCause',
'ProtocolViolationCause',
]
if TYPE_CHECKING:
from ipaddress import IPv4Address, IPv6Address
from typing import IO, Any, Callable
from pcapkit.protocols.protocol import ProtocolBase
if SPHINX_TYPE_CHECKING: # pragma: no cover
from typing_extensions import TypedDict
[docs]
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.
U: int
#: (B)eginning fragment bit.
B: int
#: (E)nding fragment bit.
E: int
[docs]
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
[docs]
def padding_length(pkt: 'dict[str, Any]') -> 'int':
"""Length of the trailing padding of an SCTP type-length-value structure.
Chunks, chunk parameters and error causes are all padded with all-zero
bytes to a multiple of four bytes, and per :rfc:`9260#section-3.2` that
padding is **not** counted in the ``length`` field. The padding is still
on the wire, though, so it has to be consumed for the enclosing list to
stay aligned.
Args:
pkt: Packet data.
Returns:
Number of padding bytes, clamped to the number of bytes left in the
enclosing structure, so that a final chunk captured without (all of)
its padding still parses, although :rfc:`9260#section-3.2` has the
sender pad every chunk. 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)
[docs]
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 ``_restore_padding`` in
:mod:`pcapkit.protocols.transport.sctp` 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 <pcapkit.protocols.schema.schema.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__')
[docs]
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.
Args:
length: Callback computing the field's nominal length.
Returns:
A callback returning that length, never exceeding the bytes left in the
enclosing structure.
A count field and the list it counts can disagree on a malformed packet, and
:meth:`ListField.unpack <pcapkit.corekit.fields.collections.ListField.unpack>`
subtracts the *parsed* size of each item from its budget -- so a
:class:`~pcapkit.corekit.fields.misc.SchemaField` item that runs out of bytes
parses to nothing, subtracts nothing, and the loop never terminates. Clamping
to the bytes on hand turns that hang into a short list, which the read
handlers then reject against the declared length.
Only list fields are wrapped: :meth:`ListField.pack
<pcapkit.corekit.fields.collections.ListField.pack>` ignores the length
entirely, so clamping cannot truncate anything on construction.
"""
def callback(pkt: 'dict[str, Any]') -> 'int':
value = length(pkt)
remaining = pkt.get('__length__')
if isinstance(remaining, int) and remaining >= 0:
return min(value, remaining)
return value
return callback
[docs]
def nested_length(base: 'int') -> 'Callable[[dict[str, Any]], int]':
"""Build a length callback for a chunk's nested type-length-value list.
Args:
base: Size of the chunk's fixed-length fields, including the four-byte
chunk header.
Returns:
A callback returning the size of the chunk's nested list *including* the
chunk's own trailing padding.
The chunks that carry a nested list -- INIT, INIT ACK, HEARTBEAT,
HEARTBEAT ACK, ABORT and ERROR -- deliberately have no separate
:class:`~pcapkit.corekit.fields.strings.PaddingField`, and fold the chunk's
trailing padding into the nested list's own span instead, which parsing and
construction each need for a reason of their own:
* On parsing, a sender is allowed by :rfc:`9260#section-3.2` to leave the
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__``. 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 <pcapkit.protocols.schema.schema.Schema.pack>` passes
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: 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':
length = pkt.get('length') or 0
return max(length - base + (-length % 4), 0)
return callback
[docs]
class ErrorCause(EnumSchema[Enum_CauseCode]):
"""Header schema for SCTP error causes."""
__default__ = lambda: UnknownCause
#: Cause code.
code: 'Enum_CauseCode' = EnumField(length=2, namespace=Enum_CauseCode)
#: Cause length.
length: 'int' = UInt16Field()
[docs]
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)
[docs]
@schema_final
class UnknownCause(ErrorCause):
"""Header schema for SCTP error causes with unknown cause codes."""
#: Cause-specific information.
value: 'bytes' = BytesField(length=lambda pkt: max(pkt['length'] - 4, 0))
#: Padding.
padding: 'bytes' = PaddingField(length=padding_length)
if TYPE_CHECKING:
def __init__(self, code: 'Enum_CauseCode', length: 'int', value: 'bytes') -> 'None': ...
[docs]
@schema_final
class InvalidStreamIdentifierCause(ErrorCause, code=Enum_CauseCode.Invalid_Stream_Identifier):
"""Header schema for SCTP invalid stream identifier error cause."""
#: Stream identifier of the offending DATA chunk.
stream_id: 'int' = UInt16Field()
#: Reserved.
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',
reserved: 'bytes') -> 'None': ...
[docs]
@schema_final
class MissingMandatoryParameterCause(ErrorCause, code=Enum_CauseCode.Missing_Mandatory_Parameter):
"""Header schema for SCTP missing mandatory parameter error cause."""
#: Number of missing parameters.
num: 'int' = UInt32Field()
#: Missing parameter types.
types: 'list[Enum_Parameter]' = ListField(
length=bounded(lambda pkt: min(max(pkt['num'], 0) * 2, max(pkt['length'] - 8, 0))),
item_type=EnumField(length=2, namespace=Enum_Parameter),
)
#: Padding.
padding: 'bytes' = PaddingField(length=padding_length)
if TYPE_CHECKING:
def __init__(self, code: 'Enum_CauseCode', length: 'int', num: 'int',
types: 'list[Enum_Parameter]') -> 'None': ...
[docs]
@schema_final
class StaleCookieCause(ErrorCause, code=Enum_CauseCode.Stale_Cookie):
"""Header schema for SCTP stale cookie error cause."""
#: Measure of staleness, in microseconds.
staleness: 'int' = UInt32Field()
#: Padding.
padding: 'bytes' = PaddingField(length=padding_length)
if TYPE_CHECKING:
def __init__(self, code: 'Enum_CauseCode', length: 'int', staleness: 'int') -> 'None': ...
[docs]
@schema_final
class OutOfResourceCause(ErrorCause, code=Enum_CauseCode.Out_of_Resource):
"""Header schema for SCTP out of resource error cause."""
#: Padding.
padding: 'bytes' = PaddingField(length=padding_length)
if TYPE_CHECKING:
def __init__(self, code: 'Enum_CauseCode', length: 'int') -> 'None': ...
[docs]
@schema_final
class UnresolvableAddressCause(ErrorCause, code=Enum_CauseCode.Unresolvable_Address):
"""Header schema for SCTP unresolvable address error cause."""
#: The offending address parameter, complete with its type and length.
value: 'bytes' = BytesField(length=lambda pkt: max(pkt['length'] - 4, 0))
#: Padding.
padding: 'bytes' = PaddingField(length=padding_length)
if TYPE_CHECKING:
def __init__(self, code: 'Enum_CauseCode', length: 'int', value: 'bytes') -> 'None': ...
[docs]
@schema_final
class UnrecognizedChunkTypeCause(ErrorCause, code=Enum_CauseCode.Unrecognized_Chunk_Type):
"""Header schema for SCTP unrecognized chunk type error cause."""
#: The unrecognized chunk, complete with its type, flags and length.
value: 'bytes' = BytesField(length=lambda pkt: max(pkt['length'] - 4, 0))
#: Padding.
padding: 'bytes' = PaddingField(length=padding_length)
if TYPE_CHECKING:
def __init__(self, code: 'Enum_CauseCode', length: 'int', value: 'bytes') -> 'None': ...
[docs]
@schema_final
class InvalidMandatoryParameterCause(ErrorCause, code=Enum_CauseCode.Invalid_Mandatory_Parameter):
"""Header schema for SCTP invalid mandatory parameter error cause."""
#: Padding.
padding: 'bytes' = PaddingField(length=padding_length)
if TYPE_CHECKING:
def __init__(self, code: 'Enum_CauseCode', length: 'int') -> 'None': ...
[docs]
@schema_final
class UnrecognizedParametersCause(ErrorCause, code=Enum_CauseCode.Unrecognized_Parameters):
"""Header schema for SCTP unrecognized parameters error cause."""
#: The unrecognized parameters, complete with their types and lengths.
value: 'bytes' = BytesField(length=lambda pkt: max(pkt['length'] - 4, 0))
#: Padding.
padding: 'bytes' = PaddingField(length=padding_length)
if TYPE_CHECKING:
def __init__(self, code: 'Enum_CauseCode', length: 'int', value: 'bytes') -> 'None': ...
[docs]
@schema_final
class NoUserDataCause(ErrorCause, code=Enum_CauseCode.No_User_Data):
"""Header schema for SCTP no user data error cause."""
#: TSN of the offending DATA chunk.
tsn: 'int' = UInt32Field()
#: Padding.
padding: 'bytes' = PaddingField(length=padding_length)
if TYPE_CHECKING:
def __init__(self, code: 'Enum_CauseCode', length: 'int', tsn: 'int') -> 'None': ...
[docs]
@schema_final
class CookieReceivedWhileShuttingDownCause(ErrorCause, code=Enum_CauseCode.Cookie_Received_While_Shutting_Down): # pylint: disable=line-too-long
"""Header schema for SCTP cookie received while shutting down error cause."""
#: Padding.
padding: 'bytes' = PaddingField(length=padding_length)
if TYPE_CHECKING:
def __init__(self, code: 'Enum_CauseCode', length: 'int') -> 'None': ...
[docs]
@schema_final
class RestartOfAnAssociationWithNewAddressesCause(ErrorCause, code=Enum_CauseCode.Restart_of_an_Association_with_New_Addresses): # pylint: disable=line-too-long
"""Header schema for SCTP restart of an association with new addresses error cause."""
#: The new address parameters, complete with their types and lengths.
value: 'bytes' = BytesField(length=lambda pkt: max(pkt['length'] - 4, 0))
#: Padding.
padding: 'bytes' = PaddingField(length=padding_length)
if TYPE_CHECKING:
def __init__(self, code: 'Enum_CauseCode', length: 'int', value: 'bytes') -> 'None': ...
[docs]
@schema_final
class UserInitiatedAbortCause(ErrorCause, code=Enum_CauseCode.User_Initiated_Abort):
"""Header schema for SCTP user-initiated abort error cause."""
#: Upper layer abort reason.
info: 'bytes' = BytesField(length=lambda pkt: max(pkt['length'] - 4, 0))
#: Padding.
padding: 'bytes' = PaddingField(length=padding_length)
if TYPE_CHECKING:
def __init__(self, code: 'Enum_CauseCode', length: 'int', info: 'bytes') -> 'None': ...
[docs]
@schema_final
class ProtocolViolationCause(ErrorCause, code=Enum_CauseCode.Protocol_Violation):
"""Header schema for SCTP protocol violation error cause."""
#: Additional information.
info: 'bytes' = BytesField(length=lambda pkt: max(pkt['length'] - 4, 0))
#: Padding.
padding: 'bytes' = PaddingField(length=padding_length)
if TYPE_CHECKING:
def __init__(self, code: 'Enum_CauseCode', length: 'int', info: 'bytes') -> 'None': ...
[docs]
class Parameter(EnumSchema[Enum_Parameter]):
"""Header schema for SCTP chunk parameters."""
__default__ = lambda: UnknownParameter
#: Parameter type.
type: 'Enum_Parameter' = EnumField(length=2, namespace=Enum_Parameter)
#: Parameter length.
length: 'int' = UInt16Field()
[docs]
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)
[docs]
@schema_final
class UnknownParameter(Parameter):
"""Header schema for SCTP chunk parameters with unknown types."""
#: Parameter value.
value: 'bytes' = BytesField(length=lambda pkt: max(pkt['length'] - 4, 0))
#: Padding.
padding: 'bytes' = PaddingField(length=padding_length)
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Parameter', length: 'int', value: 'bytes') -> 'None': ...
[docs]
@schema_final
class HeartbeatInfoParameter(Parameter, code=Enum_Parameter.Heartbeat_Info):
"""Header schema for SCTP heartbeat info parameter."""
#: Sender-specific heartbeat info.
info: 'bytes' = BytesField(length=lambda pkt: max(pkt['length'] - 4, 0))
#: Padding.
padding: 'bytes' = PaddingField(length=padding_length)
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Parameter', length: 'int', info: 'bytes') -> 'None': ...
[docs]
@schema_final
class IPv4AddressParameter(Parameter, code=Enum_Parameter.IPv4_Address):
"""Header schema for SCTP IPv4 address parameter."""
#: IPv4 address of the sending endpoint.
address: 'IPv4Address' = IPv4AddressField()
#: Padding.
padding: 'bytes' = PaddingField(length=padding_length)
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Parameter', length: 'int',
address: 'IPv4Address | int | str | bytes') -> 'None': ...
[docs]
@schema_final
class IPv6AddressParameter(Parameter, code=Enum_Parameter.IPv6_Address):
"""Header schema for SCTP IPv6 address parameter."""
#: IPv6 address of the sending endpoint.
address: 'IPv6Address' = IPv6AddressField()
#: Padding.
padding: 'bytes' = PaddingField(length=padding_length)
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Parameter', length: 'int',
address: 'IPv6Address | int | str | bytes') -> 'None': ...
[docs]
@schema_final
class StateCookieParameter(Parameter, code=Enum_Parameter.State_Cookie):
"""Header schema for SCTP state cookie parameter."""
#: State cookie.
cookie: 'bytes' = BytesField(length=lambda pkt: max(pkt['length'] - 4, 0))
#: Padding.
padding: 'bytes' = PaddingField(length=padding_length)
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Parameter', length: 'int', cookie: 'bytes') -> 'None': ...
[docs]
@schema_final
class UnrecognizedParameter(Parameter, code=Enum_Parameter.Unrecognized_Parameter):
"""Header schema for SCTP unrecognized parameter parameter."""
#: The unrecognized parameter, complete with its type and length.
value: 'bytes' = BytesField(length=lambda pkt: max(pkt['length'] - 4, 0))
#: Padding.
padding: 'bytes' = PaddingField(length=padding_length)
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Parameter', length: 'int', value: 'bytes') -> 'None': ...
[docs]
@schema_final
class CookiePreservativeParameter(Parameter, code=Enum_Parameter.Cookie_Preservative):
"""Header schema for SCTP cookie preservative parameter."""
#: Suggested cookie life-span increment, in milliseconds.
increment: 'int' = UInt32Field()
#: Padding.
padding: 'bytes' = PaddingField(length=padding_length)
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Parameter', length: 'int', increment: 'int') -> 'None': ...
[docs]
@schema_final
class HostNameAddressParameter(Parameter, code=Enum_Parameter.Host_Name_Address):
"""Header schema for SCTP host name address parameter."""
#: Host name as captured. :rfc:`9260#section-3.3.2.1.4` asks for at least one
#: null terminator, but a parsed name may lack it.
name: 'bytes' = BytesField(length=lambda pkt: max(pkt['length'] - 4, 0))
#: Padding.
padding: 'bytes' = PaddingField(length=padding_length)
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Parameter', length: 'int', name: 'bytes') -> 'None': ...
[docs]
@schema_final
class SupportedAddressTypesParameter(Parameter, code=Enum_Parameter.Supported_Address_Types):
"""Header schema for SCTP supported address types parameter."""
#: Supported address types, given as address parameter types.
types: 'list[Enum_Parameter]' = ListField(
length=bounded(lambda pkt: max(pkt['length'] - 4, 0)),
item_type=EnumField(length=2, namespace=Enum_Parameter),
)
#: Padding.
padding: 'bytes' = PaddingField(length=padding_length)
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Parameter', length: 'int',
types: 'list[Enum_Parameter]') -> 'None': ...
[docs]
@schema_final
class GapAckBlock(Schema):
"""Header schema for SCTP SACK chunk gap ack blocks."""
#: Start offset TSN of the gap ack block.
start: 'int' = UInt16Field()
#: End offset TSN of the gap ack block.
end: 'int' = UInt16Field()
if TYPE_CHECKING:
def __init__(self, start: 'int', end: 'int') -> 'None': ...
[docs]
class Chunk(EnumSchema[Enum_Chunk]):
"""Header schema for SCTP chunks."""
__default__ = lambda: UnknownChunk
#: Chunk type.
type: 'Enum_Chunk' = EnumField(length=1, namespace=Enum_Chunk)
#: Chunk flags, whose meaning depends on the chunk type.
flags: 'bytes' = BytesField(length=1)
#: Chunk length, excluding any trailing padding.
length: 'int' = UInt16Field()
[docs]
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)
[docs]
@schema_final
class UnknownChunk(Chunk):
"""Header schema for SCTP chunks with unknown types."""
#: Chunk value.
value: 'bytes' = BytesField(length=lambda pkt: max(pkt['length'] - 4, 0))
#: Padding.
padding: 'bytes' = PaddingField(length=padding_length)
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Chunk', flags: 'bytes', length: 'int',
value: 'bytes') -> 'None': ...
[docs]
@schema_final
class DATAChunk(Chunk, code=Enum_Chunk.Payload_Data):
"""Header schema for SCTP DATA chunks."""
#: Chunk flags.
flags: 'DATAChunkFlags' = BitField(length=1, namespace={
'reserved': (0, 4),
'I': (4, 1),
'U': (5, 1),
'B': (6, 1),
'E': (7, 1),
})
#: Transmission sequence number.
tsn: 'int' = UInt32Field()
#: Stream identifier.
stream_id: 'int' = UInt16Field()
#: Stream sequence number.
stream_seq: 'int' = UInt16Field()
#: Payload protocol identifier.
ppid: 'Enum_PayloadProtocolIdentifier' = EnumField(
length=4, namespace=Enum_PayloadProtocolIdentifier)
#: User data.
data: 'bytes' = BytesField(length=lambda pkt: max(pkt['length'] - 16, 0))
#: Padding.
padding: 'bytes' = PaddingField(length=padding_length)
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Chunk', flags: 'DATAChunkFlags', length: 'int', tsn: 'int',
stream_id: 'int', stream_seq: 'int',
ppid: 'Enum_PayloadProtocolIdentifier | int', data: 'bytes') -> 'None': ...
[docs]
@schema_final
class INITChunk(Chunk, code=Enum_Chunk.Initiation):
"""Header schema for SCTP INIT chunks."""
#: Initiate tag.
init_tag: 'int' = UInt32Field()
#: Advertised receiver window credit.
a_rwnd: 'int' = UInt32Field()
#: Number of outbound streams.
outbound_streams: 'int' = UInt16Field()
#: Number of inbound streams.
inbound_streams: 'int' = UInt16Field()
#: Initial transmission sequence number.
init_tsn: 'int' = UInt32Field()
#: Optional and variable-length parameters, including the chunk's own
#: trailing padding; see :func:`nested_length`.
parameters: 'list[Parameter]' = OptionField(
length=bounded(nested_length(20)),
base_schema=Parameter,
type_name='type',
registry=Parameter.registry,
)
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Chunk', flags: 'bytes', length: 'int', init_tag: 'int',
a_rwnd: 'int', outbound_streams: 'int', inbound_streams: 'int',
init_tsn: 'int',
parameters: 'list[Parameter | bytes] | bytes') -> 'None': ...
[docs]
@schema_final
class INITACKChunk(Chunk, code=Enum_Chunk.Initiation_Acknowledgement):
"""Header schema for SCTP INIT ACK chunks."""
#: Initiate tag.
init_tag: 'int' = UInt32Field()
#: Advertised receiver window credit.
a_rwnd: 'int' = UInt32Field()
#: Number of outbound streams.
outbound_streams: 'int' = UInt16Field()
#: Number of inbound streams.
inbound_streams: 'int' = UInt16Field()
#: Initial transmission sequence number.
init_tsn: 'int' = UInt32Field()
#: Optional and variable-length parameters, including the chunk's own
#: trailing padding; see :func:`nested_length`.
parameters: 'list[Parameter]' = OptionField(
length=bounded(nested_length(20)),
base_schema=Parameter,
type_name='type',
registry=Parameter.registry,
)
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Chunk', flags: 'bytes', length: 'int', init_tag: 'int',
a_rwnd: 'int', outbound_streams: 'int', inbound_streams: 'int',
init_tsn: 'int',
parameters: 'list[Parameter | bytes] | bytes') -> 'None': ...
[docs]
@schema_final
class SACKChunk(Chunk, code=Enum_Chunk.Selective_Acknowledgement):
"""Header schema for SCTP SACK chunks."""
#: Cumulative TSN ack.
cum_tsn_ack: 'int' = UInt32Field()
#: Advertised receiver window credit.
a_rwnd: 'int' = UInt32Field()
#: Number of gap ack blocks.
num_gap_blocks: 'int' = UInt16Field()
#: Number of duplicate TSNs.
num_dup_tsn: 'int' = UInt16Field()
#: Gap ack blocks.
gap_blocks: 'list[GapAckBlock]' = ListField(
length=bounded(
lambda pkt: min(max(pkt['num_gap_blocks'], 0) * 4, max(pkt['length'] - 16, 0))),
item_type=SchemaField(length=4, schema=GapAckBlock),
)
#: Duplicate TSNs.
dup_tsn: 'list[int]' = ListField(
length=bounded(
lambda pkt: min(max(pkt['num_dup_tsn'], 0) * 4,
max(pkt['length'] - 16 - max(pkt['num_gap_blocks'], 0) * 4, 0))),
item_type=UInt32Field(),
)
#: Padding.
padding: 'bytes' = PaddingField(length=padding_length)
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Chunk', flags: 'bytes', length: 'int', cum_tsn_ack: 'int',
a_rwnd: 'int', num_gap_blocks: 'int', num_dup_tsn: 'int',
gap_blocks: 'list[GapAckBlock] | bytes',
dup_tsn: 'list[int] | bytes') -> 'None': ...
[docs]
@schema_final
class HeartbeatChunk(Chunk, code=Enum_Chunk.Heartbeat_Request):
"""Header schema for SCTP HEARTBEAT chunks."""
#: Heartbeat information parameters, including the chunk's own trailing
#: padding; see :func:`nested_length`.
parameters: 'list[Parameter]' = OptionField(
length=bounded(nested_length(4)),
base_schema=Parameter,
type_name='type',
registry=Parameter.registry,
)
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Chunk', flags: 'bytes', length: 'int',
parameters: 'list[Parameter | bytes] | bytes') -> 'None': ...
[docs]
@schema_final
class HeartbeatACKChunk(Chunk, code=Enum_Chunk.Heartbeat_Acknowledgement):
"""Header schema for SCTP HEARTBEAT ACK chunks."""
#: Heartbeat information parameters, including the chunk's own trailing
#: padding; see :func:`nested_length`.
parameters: 'list[Parameter]' = OptionField(
length=bounded(nested_length(4)),
base_schema=Parameter,
type_name='type',
registry=Parameter.registry,
)
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Chunk', flags: 'bytes', length: 'int',
parameters: 'list[Parameter | bytes] | bytes') -> 'None': ...
[docs]
@schema_final
class AbortChunk(Chunk, code=Enum_Chunk.Abort):
"""Header schema for SCTP ABORT chunks."""
#: 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;
#: see :func:`nested_length`.
error: 'list[ErrorCause]' = OptionField(
length=bounded(nested_length(4)),
base_schema=ErrorCause,
type_name='code',
registry=ErrorCause.registry,
)
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Chunk', flags: 'TBitFlags', length: 'int',
error: 'list[ErrorCause | bytes] | bytes') -> 'None': ...
[docs]
@schema_final
class ShutdownChunk(Chunk, code=Enum_Chunk.Shutdown):
"""Header schema for SCTP SHUTDOWN chunks."""
#: Cumulative TSN ack.
cum_tsn_ack: 'int' = UInt32Field()
#: Padding.
padding: 'bytes' = PaddingField(length=padding_length)
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Chunk', flags: 'bytes', length: 'int',
cum_tsn_ack: 'int') -> 'None': ...
[docs]
@schema_final
class ShutdownACKChunk(Chunk, code=Enum_Chunk.Shutdown_Acknowledgement):
"""Header schema for SCTP SHUTDOWN ACK chunks."""
#: Padding.
padding: 'bytes' = PaddingField(length=padding_length)
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Chunk', flags: 'bytes', length: 'int') -> 'None': ...
[docs]
@schema_final
class ErrorChunk(Chunk, code=Enum_Chunk.Operation_Error):
"""Header schema for SCTP ERROR chunks."""
#: One or more error causes, including the chunk's own trailing padding;
#: see :func:`nested_length`.
error: 'list[ErrorCause]' = OptionField(
length=bounded(nested_length(4)),
base_schema=ErrorCause,
type_name='code',
registry=ErrorCause.registry,
)
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Chunk', flags: 'bytes', length: 'int',
error: 'list[ErrorCause | bytes] | bytes') -> 'None': ...
[docs]
@schema_final
class CookieEchoChunk(Chunk, code=Enum_Chunk.State_Cookie):
"""Header schema for SCTP COOKIE ECHO chunks."""
#: State cookie, as received in the INIT ACK chunk's state cookie parameter.
cookie: 'bytes' = BytesField(length=lambda pkt: max(pkt['length'] - 4, 0))
#: Padding.
padding: 'bytes' = PaddingField(length=padding_length)
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Chunk', flags: 'bytes', length: 'int',
cookie: 'bytes') -> 'None': ...
[docs]
@schema_final
class CookieACKChunk(Chunk, code=Enum_Chunk.Cookie_Acknowledgement):
"""Header schema for SCTP COOKIE ACK chunks."""
#: Padding.
padding: 'bytes' = PaddingField(length=padding_length)
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Chunk', flags: 'bytes', length: 'int') -> 'None': ...
[docs]
@schema_final
class ShutdownCompleteChunk(Chunk, code=Enum_Chunk.Shutdown_Complete):
"""Header schema for SCTP SHUTDOWN COMPLETE chunks."""
#: Chunk flags.
flags: 'TBitFlags' = BitField(length=1, namespace={
'reserved': (0, 7),
'T': (7, 1),
})
#: Padding.
padding: 'bytes' = PaddingField(length=padding_length)
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Chunk', flags: 'TBitFlags', length: 'int') -> 'None': ...
[docs]
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, see :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
[docs]
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
<pcapkit.protocols.schema.schema.Schema.unpack>` hands back to the stream for
:attr:`SCTP.trailer` to keep as captured (:issue:`1468`).
"""
[docs]
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
[docs]
@schema_final
class SCTP(Schema):
"""Header schema for SCTP packets.
Note:
Unlike :class:`~pcapkit.protocols.schema.transport.tcp.TCP` and
:class:`~pcapkit.protocols.schema.transport.udp.UDP`, there is **no**
payload field, since SCTP carries its user data inside DATA chunks
rather than after the common header. See
:meth:`SCTP._get_payload <pcapkit.protocols.transport.sctp.SCTP._get_payload>`
for how the next layer is located.
"""
#: Source port.
srcport: 'Enum_AppType' = PortEnumField(length=2, namespace=Enum_AppType,
proto=Enum_TransportProtocol.sctp)
#: Destination port.
dstport: 'Enum_AppType' = PortEnumField(length=2, namespace=Enum_AppType,
proto=Enum_TransportProtocol.sctp)
#: Verification tag.
vtag: 'int' = UInt32Field()
#: Checksum, as a CRC32c over the whole packet with this field zeroed.
chksum: 'bytes' = BytesField(length=4)
#: Chunks.
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, see :class:`ChunkListField`
#: and :attr:`IPv4.trailer <pcapkit.protocols.schema.internet.ipv4.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',
trailer: 'bytes' = b'') -> 'None': ...