# -*- coding: utf-8 -*-
# mypy: disable-error-code=assignment
"""header schema for transmission control protocol"""
import collections
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.tcp.checksum import Checksum as Enum_Checksum
from pcapkit.const.tcp.mp_tcp_option import MPTCPOption as Enum_MPTCPOption
from pcapkit.const.tcp.option import Option as Enum_Option
from pcapkit.corekit.fields.collections import ListField, OptionField
from pcapkit.corekit.fields.ipaddress import IPv4AddressField, IPv6AddressField
from pcapkit.corekit.fields.misc import (ConditionalField, ForwardMatchField, NoValueField,
PayloadField, SchemaField, SwitchField)
from pcapkit.corekit.fields.numbers import (EnumField, PortEnumField, UInt8Field, UInt16Field,
UInt32Field, UInt64Field)
from pcapkit.corekit.fields.strings import BitField, BytesField, PaddingField
from pcapkit.protocols.schema.schema import EnumSchema, Schema, schema_final
from pcapkit.utilities.exceptions import FieldError
from pcapkit.utilities.logging import SPHINX_TYPE_CHECKING
__all__ = [
'TCP',
'Option',
'UnassignedOption', 'EndOfOptionList', 'NoOperation', 'MaximumSegmentSize', 'WindowScale',
'SACKPermitted', 'SACK', 'Echo', 'EchoReply', 'Timestamps', 'PartialOrderConnectionPermitted',
'PartialOrderServiceProfile', 'CC', 'CCNew', 'CCEcho', 'AlternateChecksumRequest',
'AlternateChecksumData', 'MD5Signature', 'QuickStartResponse', 'UserTimeout',
'Authentication', 'FastOpenCookie',
'MPTCP',
'MPTCPUnknown', 'MPTCPCapable', 'MPTCPDSS', 'MPTCPAddAddress', 'MPTCPRemoveAddress',
'MPTCPPriority', 'MPTCPFallback', 'MPTCPFastclose',
'MPTCPJoin',
'MPTCPJoinSYN', 'MPTCPJoinSYNACK', 'MPTCPJoinACK',
]
if TYPE_CHECKING:
from ipaddress import IPv4Address, IPv6Address
from typing import Any, DefaultDict, Optional, Type
from pcapkit.corekit.fields.field import FieldBase as Field
from pcapkit.protocols.protocol import ProtocolBase
#: Minimum length of a TCP header, i.e. the fixed part with no options,
#: :rfc:`9293#section-3.1`. The one home of this value (:issue:`1739`), here so
#: that the schema's option length and the protocol modules read the same one;
#: it is fixed by the RFC, so it is never configurable.
#:
#: .. versionadded:: 1.5.0
TCP_MIN_HEADER_LEN = 20
#: Offset of the octet whose high four bits are the Data Offset field -- the
#: header length in 32-bit words -- within a TCP header, :rfc:`9293#section-3.1`.
#: The one home of this value (:issue:`1739`), beside :data:`TCP_MIN_HEADER_LEN`;
#: it is fixed by the RFC, so it is never configurable.
#:
#: .. versionadded:: 1.5.0
TCP_DATA_OFFSET_OCTET = 12
if SPHINX_TYPE_CHECKING: # pragma: no cover
from typing_extensions import TypedDict
[docs]
class OffsetFlag(TypedDict):
"""TCP offset field flag."""
#: Data offset.
offset: int
#: Reserved bits.
reserved: int
#: ECN-nonce concealment protection.
ns: int
[docs]
class Flags(TypedDict):
"""TCP flags."""
#: Congestion window reduced.
cwr: int
#: ECN-Echo.
ece: int
#: Urgent.
urg: int
#: Acknowledgment.
ack: int
#: Push function.
psh: int
#: Reset connection.
rst: int
#: Synchronize sequence numbers.
syn: int
#: Last packet from sender.
fin: int
[docs]
class POCProfile(TypedDict):
"""TCP partial order connection service profile."""
#: Start flag.
start: int
#: End flag.
end: int
#: Reserved bits (``Filler``).
reserved: int
[docs]
class QuickStartFlags(TypedDict):
"""TCP quick start flags."""
#: Reserved bits.
reserved: int
#: Rate request.
rate: int
[docs]
class QuickStartNonce(TypedDict):
"""TCP quick start nonce."""
#: Nonce.
nonce: int
#: Reserved bits (``R``).
reserved: int
[docs]
class TimeoutInfo(TypedDict):
"""User timeout information."""
#: Granularity.
granularity: int
#: Timeout value.
timeout: int
[docs]
class MPTCPSubtypeTest(TypedDict):
"""TCP MPTCP subtype."""
#: Length.
length: int
#: Subtype.
subtype: int
[docs]
class MPTCPSubtypeUnknown(TypedDict):
"""TCP unknown MPTCP subtype field."""
#: Subtype.
subtype: int
#: Data.
data: int
[docs]
class MPTCPSubtypeCapable(TypedDict):
"""MPTCP Capable subtype field."""
#: Subtype.
subtype: int
#: Version.
version: int
[docs]
class MPTCPCapableFlags(TypedDict):
"""MPTCP Capable flags."""
#: Checksum required.
req: int
#: Extensibility flag.
ext: int
#: Do not establish new subflows to the source address.
deny_join: int
#: Reserved flags (``D`` to ``G``).
reserved: int
#: Use of HMAC-SHA1 under version 0, HMAC-SHA256 under version 1.
hsa: int
[docs]
class MPTCPSubtypeJoin(TypedDict):
"""MPTCP Join subtype field."""
#: Subtype.
subtype: int
#: Reserved.
reserved: int
#: Backup flag.
backup: int
[docs]
class MPTCPSubtype(TypedDict):
"""MPTCP subtype field."""
#: Subtype.
subtype: int
#: Reserved.
reserved: int
[docs]
class MPTCPDSSFlags(TypedDict):
"""MPTCP-DSS flags."""
#: Reserved.
reserved: int
#: ``DATA_FIN`` flag.
F: int
#: Data sequence number is 8 octets (if not set, DSN is 4 octets).
m: int
#: Data Sequence Number (DSN), Subflow Sequence Number (SSN), Data-Level
#: Length, and Checksum present.
M: int
#: Data ACK is 8 octets (if not set, Data ACK is 4 octets).
a: int
#: Data ACK present.
A: int
[docs]
class MPTCPSubtypeAddAddress(TypedDict):
"""MPTCP Add Address subtype field."""
#: Subtype.
subtype: int
#: Reserved.
reserved: int
#: Echo flag.
echo: int
[docs]
class MPTCPSubtypePriority(TypedDict):
"""MPTCP Priority subtype field."""
#: Subtype.
subtype: int
#: Reserved.
reserved: int
#: Backup flag.
backup: int
[docs]
def mptcp_data_selector(pkt: 'dict[str, Any]') -> 'Field':
"""Selector function for :attr:`_MPTCP.data` field.
Args:
pkt: Packet data.
Returns:
A :class:`~pcapkit.corekit.fields.misc.SchemaField` wrapped
:class:`~pcapkit.protocols.schema.transport.tcp.MPTCP` subclass
instance.
Raises:
FieldError: If an MP_JOIN option is carried by a segment with
neither SYN nor ACK set.
"""
subtype = Enum_MPTCPOption.get(pkt['test']['subtype'])
pkt['test']['subtype'] = subtype
schema = MPTCP.registry[subtype]
if subtype == Enum_MPTCPOption.MP_JOIN and schema is MPTCPJoin: # placeholder
if pkt['flags']['syn'] == 1 and pkt['flags']['ack'] == 0:
schema = MPTCPJoinSYN
elif pkt['flags']['syn'] == 1 and pkt['flags']['ack'] == 1:
schema = MPTCPJoinSYNACK
elif pkt['flags']['syn'] == 0 and pkt['flags']['ack'] == 1:
schema = MPTCPJoinACK
else:
raise FieldError(f'TCP: [OptNo {Enum_Option.Multipath_TCP}] {Enum_MPTCPOption.MP_JOIN} invalid flags')
return SchemaField(length=pkt['test']['length'], schema=schema)
[docs]
def mptcp_add_address_selector(pkt: 'dict[str, Any]') -> 'Field':
"""Selector function for :attr:`MPTCPAddAddress.address` field.
Args:
pkt: Packet data.
Returns:
An :class:`~pcapkit.corekit.fields.ipaddress.IPv4AddressField` instance
if the option length is one of the IPv4 lengths, or an
:class:`~pcapkit.corekit.fields.ipaddress.IPv6AddressField` instance if
it is one of the IPv6 lengths.
Raises:
FieldError: If the option length is neither.
Note:
:rfc:`8684` section 3.4.1 figure 12 has no IP version field: the
address family follows from the option length, which is 8 or 20 for
an IPv4 or IPv6 address, plus 2 with the port, plus 8 with the
truncated HMAC. The :rfc:`6824` ``IPVer`` nibble occupies the bits
:rfc:`8684` reserves and spends on ``E``, and its lengths are the
HMAC-less ones, so both layouts are read the same way.
"""
if pkt['length'] in (8, 10, 16, 18):
return IPv4AddressField()
if pkt['length'] in (20, 22, 28, 30):
return IPv6AddressField()
raise FieldError(f'TCP: [OptNo {Enum_Option.Multipath_TCP}] {Enum_MPTCPOption.ADD_ADDR} invalid length')
[docs]
def mptcp_dss_checksum(pkt: 'dict[str, Any]') -> 'bool':
"""Condition function for :attr:`MPTCPDSS.checksum` field.
:rfc:`8684` section 3.3 makes the checksum part of the mapping only when
checksums were negotiated, so a mapping without one is 2 octets shorter
and only the option length tells the two forms apart.
Args:
pkt: Packet data.
Returns:
Whether the option length leaves room for the checksum after the
Data ACK and the mapping.
"""
if not pkt['flags']['M']:
return False
size = 4 + ((8 if pkt['flags']['a'] else 4) if pkt['flags']['A'] else 0) \
+ (8 if pkt['flags']['m'] else 4) + 6
return pkt['length'] >= size + 2
[docs]
def mptcp_dss_ack_selector(pkt: 'dict[str, Any]') -> 'Field':
"""Selector function for :attr:`MPTCPDSS.ack` field.
:rfc:`8684` section 3.3 figure 9 gives the Data ACK as "4 or 8 octets,
depending on flags": present only when ``A`` is set, and 8 octets wide only
when ``a`` is *also* set -- "a = Data ACK is 8 octets (if not set, Data ACK
is 4 octets)".
Args:
pkt: Packet data.
Returns:
A :class:`~pcapkit.corekit.fields.misc.NoValueField` instance if ``A``
is clear, since the field is then absent from the wire; otherwise a
:class:`~pcapkit.corekit.fields.numbers.UInt64Field` instance if ``a``
is set, or a :class:`~pcapkit.corekit.fields.numbers.UInt32Field`
instance if it is clear.
Note:
This is a :class:`~pcapkit.corekit.fields.misc.SwitchField` selector
rather than a :class:`~pcapkit.corekit.fields.misc.ConditionalField`
wrapping ``NumberField(length=lambda pkt: ...)``. An unextended Data ACK
is 4 octets, so a width lambda returning ``0`` for it would leave the
``length`` octet counting 4 octets that were never packed.
The ``ConditionalField`` form could express this field correctly, since a
callable-length ``NumberField`` packs and unpacks both the 4- and the
8-octet form, and wire *absence* is no obstacle:
:attr:`MPTCPDSS.ssn`, :attr:`MPTCPDSS.dl_len` and
:attr:`MPTCPDSS.checksum` are each a ``ConditionalField`` on the sibling
``M`` flag (the checksum on the option length as well). The
``SwitchField`` form is kept for a narrower reason about
composition. A
:class:`~pcapkit.corekit.fields.misc.ConditionalField`'s ``length``
forwards to the wrapped field unconditionally, never consulting the
condition, so reading it while the condition is false -- the wrapped field
then still unresolved, at its ``-1`` placeholder -- raises
``struct.error: bad char in struct format``. Nothing here meets that only
because :class:`Schema
<pcapkit.protocols.schema.schema.Schema>`'s ``pack`` and ``unpack``
special-case ``ConditionalField`` by type and skip the wrapped field
outright before any ``length`` is read. A ``SwitchField`` needs no such
special case: its selector always hands back an already-concrete field,
:class:`~pcapkit.corekit.fields.misc.NoValueField` included, so its
``length`` is safe wherever it is read. Swapping the two would be a
behaviour change, not a tidy-up.
"""
if not pkt['flags']['A']:
return NoValueField()
return UInt64Field() if pkt['flags']['a'] else UInt32Field()
[docs]
def mptcp_dss_dsn_selector(pkt: 'dict[str, Any]') -> 'Field':
"""Selector function for :attr:`MPTCPDSS.dsn` field.
:rfc:`8684` section 3.3 figure 9 gives the Data Sequence Number as "4 or 8
octets, depending on flags": present only when ``M`` is set, and 8 octets
wide only when ``m`` is *also* set -- "m = Data Sequence Number is 8 octets
(if not set, DSN is 4 octets)".
Args:
pkt: Packet data.
Returns:
A :class:`~pcapkit.corekit.fields.misc.NoValueField` instance if ``M``
is clear, since the field is then absent from the wire; otherwise a
:class:`~pcapkit.corekit.fields.numbers.UInt64Field` instance if ``m``
is set, or a :class:`~pcapkit.corekit.fields.numbers.UInt32Field`
instance if it is clear.
Note:
Identical in shape to :func:`mptcp_dss_ack_selector`; see that function's
note for why a ``SwitchField`` is used in place of a
``ConditionalField`` around a callable-length ``NumberField``.
"""
if not pkt['flags']['M']:
return NoValueField()
return UInt64Field() if pkt['flags']['m'] else UInt32Field()
[docs]
class Option(EnumSchema[Enum_Option]):
"""Header schema for TCP options."""
__default__ = lambda: UnassignedOption
#: Option kind.
kind: 'Enum_Option' = EnumField(length=1, namespace=Enum_Option)
#: Option length.
length: 'int' = ConditionalField(
UInt8Field(),
lambda pkt: pkt['kind'] not in (Enum_Option.End_of_Option_List, Enum_Option.No_Operation),
)
[docs]
def post_process(self, packet: 'dict[str, Any]') -> 'Schema':
"""Revise ``schema`` data after unpacking process.
Args:
packet: Unpacked data.
Returns:
Revised schema.
"""
# for EOOL/NOP option, length is always 1
if self.kind in (Enum_Option.End_of_Option_List, Enum_Option.No_Operation):
self.length = 1
return self
[docs]
@schema_final
class UnassignedOption(Option):
"""Header schema for TCP unassigned options."""
#: Option data.
data: 'bytes' = BytesField(length=lambda pkt: pkt['length'] - 2)
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int', data: 'bytes') -> 'None': ...
[docs]
@schema_final
class EndOfOptionList(Option, code=Enum_Option.End_of_Option_List):
"""Header schema for TCP end of option list."""
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int') -> 'None': ...
[docs]
@schema_final
class NoOperation(Option, code=Enum_Option.No_Operation):
"""Header schema for TCP no operation."""
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int') -> 'None': ...
[docs]
@schema_final
class MaximumSegmentSize(Option, code=Enum_Option.Maximum_Segment_Size):
"""Header schema for TCP max segment size option."""
#: Maximum segment size.
mss: 'int' = UInt16Field()
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int', mss: 'int') -> 'None': ...
[docs]
@schema_final
class WindowScale(Option, code=Enum_Option.Window_Scale):
"""Header schema for TCP window scale option."""
#: Window scale (shift count).
shift: 'int' = UInt8Field()
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int', shift: 'int') -> 'None': ...
[docs]
@schema_final
class SACKPermitted(Option, code=Enum_Option.SACK_Permitted):
"""Header schema for TCP SACK permitted option."""
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int') -> 'None': ...
[docs]
@schema_final
class SACKBlock(Schema):
"""Header schema for TCP SACK option data."""
#: Left edge of the block.
left: 'int' = UInt32Field()
#: Right edge of the block.
right: 'int' = UInt32Field()
if TYPE_CHECKING:
def __init__(self, left: 'int', right: 'int') -> 'None': ...
[docs]
@schema_final
class SACK(Option, code=Enum_Option.SACK):
"""Header schema for TCP SACK option."""
#: Selected ACK data.
sack: 'list[SACKBlock]' = ListField(
length=lambda pkt: pkt['length'] - 2,
item_type=SchemaField(length=8, schema=SACKBlock),
)
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int', sack: 'list[SACKBlock]') -> 'None': ...
[docs]
@schema_final
class Echo(Option, code=Enum_Option.Echo):
"""Header schema for TCP echo option."""
#: Info to be echoed.
data: 'bytes' = BytesField(length=4)
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int', data: 'bytes') -> 'None': ...
[docs]
@schema_final
class EchoReply(Option, code=Enum_Option.Echo_Reply):
"""Header schema for TCP echo reply option."""
#: Echoed info.
data: 'bytes' = BytesField(length=4)
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int', data: 'bytes') -> 'None': ...
[docs]
@schema_final
class Timestamps(Option, code=Enum_Option.Timestamps):
"""Header schema for TCP timestamps option.
.. versionchanged:: 1.1.0
Renamed from ``Timestamp``, which :mod:`pcapkit.protocols.schema`
exported as ``TCP_Timestamp``; the export is ``TCP_Timestamps``.
"""
#: Timestamp value.
value: 'int' = UInt32Field()
#: Timestamp echo reply.
reply: 'int' = UInt32Field()
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int', value: 'int', reply: 'int') -> 'None': ...
[docs]
@schema_final
class PartialOrderConnectionPermitted(Option, code=Enum_Option.Partial_Order_Connection_Permitted):
"""Header schema for TCP partial order connection permitted option."""
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int') -> 'None': ...
[docs]
@schema_final
class PartialOrderServiceProfile(Option, code=Enum_Option.Partial_Order_Service_Profile):
"""Header schema for TCP partial order connection service profile option.
.. versionchanged:: 1.1.0
Renamed from ``PartialOrderConnectionProfile``, which
:mod:`pcapkit.protocols.schema` exported as
``TCP_PartialOrderConnectionProfile``; the export is
``TCP_PartialOrderServiceProfile``.
"""
#: Profile data.
profile: 'POCProfile' = BitField(length=1, namespace={
'start': (0, 1),
'end': (1, 1),
'reserved': (2, 6),
})
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int', profile: 'POCProfile') -> 'None': ...
[docs]
@schema_final
class CC(Option, code=Enum_Option.CC):
"""Header schema for TCP CC option."""
#: Connection count.
count: 'int' = UInt32Field()
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int', count: 'int') -> 'None': ...
[docs]
@schema_final
class CCNew(Option, code=Enum_Option.CC_NEW):
"""Header schema for TCP connection count (new) option."""
#: Connection count.
count: 'int' = UInt32Field()
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int', count: 'int') -> 'None': ...
[docs]
@schema_final
class CCEcho(Option, code=Enum_Option.CC_ECHO):
"""Header schema for TCP connection count (echo) option."""
#: Connection count.
count: 'int' = UInt32Field()
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int', count: 'int') -> 'None': ...
[docs]
@schema_final
class AlternateChecksumRequest(Option, code=Enum_Option.TCP_Alternate_Checksum_Request):
"""Header schema for TCP alternate checksum request option."""
#: Checksum algorithm.
algorithm: 'Enum_Checksum' = EnumField(length=1, namespace=Enum_Checksum)
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int', algorithm: 'Enum_Checksum') -> 'None': ...
[docs]
@schema_final
class AlternateChecksumData(Option, code=Enum_Option.TCP_Alternate_Checksum_Data):
"""Header schema for TCP alternate checksum data option."""
#: Checksum data.
data: 'bytes' = BytesField(length=lambda pkt: pkt['length'] - 2)
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int', data: 'bytes') -> 'None': ...
[docs]
@schema_final
class MD5Signature(Option, code=Enum_Option.MD5_Signature_Option):
"""Header schema for TCP MD5 signature option."""
#: MD5 digest.
digest: 'bytes' = BytesField(length=16)
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int', digest: 'bytes') -> 'None': ...
[docs]
@schema_final
class QuickStartResponse(Option, code=Enum_Option.Quick_Start_Response):
"""Header schema for TCP quick start response option."""
#: Flags.
flags: 'QuickStartFlags' = BitField(length=1, namespace={
'reserved': (0, 4),
'rate': (4, 4),
})
#: TTL difference.
diff: 'int' = UInt8Field()
#: QS nonce.
nonce: 'QuickStartNonce' = BitField(length=4, namespace={
'nonce': (0, 30),
'reserved': (30, 2),
})
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int', flags: 'QuickStartFlags', diff: 'int', nonce: 'QuickStartNonce') -> 'None': ...
[docs]
@schema_final
class UserTimeout(Option, code=Enum_Option.User_Timeout_Option):
"""Header schema for TCP user timeout option."""
#: Granularity and user timeout.
info: 'TimeoutInfo' = BitField(length=2, namespace={
'granularity': (0, 1),
'timeout': (1, 15),
})
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int', info: 'TimeoutInfo') -> 'None': ...
[docs]
@schema_final
class Authentication(Option, code=Enum_Option.TCP_Authentication_Option):
"""Header schema for TCP authentication option."""
#: Key ID.
key_id: 'int' = UInt8Field()
#: Next key ID.
next_key_id: 'int' = UInt8Field()
#: MAC value.
mac: 'bytes' = BytesField(length=lambda pkt: pkt['length'] - 4)
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int', key_id: 'int', next_key_id: 'int', mac: 'bytes') -> 'None': ...
[docs]
@schema_final
class _MPTCP(Schema):
"""Header schema for Multipath TCP options in a generic representation."""
#: Subtype and flags.
#:
#: The forward match is anchored at the option's first octet, not after the
#: option header: this schema is what :attr:`Option.registry
#: <pcapkit.protocols.schema.schema.EnumSchema.registry>` dispatches to
#: for ``kind == Multipath_TCP`` and, unlike the other option schemas, it
#: does not inherit :class:`Option` and so consumes ``kind`` and ``length``
#: itself. :rfc:`8684` section 3 therefore puts ``kind`` in bits 0-7,
#: ``length`` in bits 8-15, and the subtype in bits 16-19 of this 3-octet
#: window -- which is why ``subtype`` reads from bit 16 and ``length`` has
#: to read from bit 8.
test: 'MPTCPSubtypeTest' = ForwardMatchField(BitField(length=3, namespace={
'length': (8, 8),
'subtype': (16, 4),
}))
#: Subtype-specific data.
data: 'MPTCP' = SwitchField(
selector=mptcp_data_selector,
)
[docs]
def post_process(self, packet: 'dict[str, Any]') -> 'MPTCP':
"""Revise ``schema`` data after unpacking process.
Args:
packet: Unpacked data.
Returns:
Revised schema.
"""
ret = self.data
ret.option = Enum_Option.Multipath_TCP
ret.length = self.test['length']
ret.subtype = Enum_MPTCPOption.get(packet['test']['subtype'])
return ret
# register ``_MPTCP`` as ``Multipath_TCP`` option
Option.register(Enum_Option.Multipath_TCP, _MPTCP)
[docs]
class MPTCP(EnumSchema[Enum_MPTCPOption]):
"""Header schema for Multipath TCP options.
.. versionchanged:: 1.1.0
An :class:`~pcapkit.protocols.schema.schema.EnumSchema` keyed on
:class:`~pcapkit.const.tcp.mp_tcp_option.MPTCPOption` rather than a
subclass of :class:`~pcapkit.protocols.schema.transport.tcp.Option`,
which registers :class:`~pcapkit.protocols.schema.transport.tcp._MPTCP`
under ``Multipath_TCP`` instead.
"""
__enum__: 'DefaultDict[Enum_MPTCPOption, Type[MPTCP]]' = collections.defaultdict(lambda: MPTCPUnknown)
# NOTE: ``_MPTCP.data`` (a :class:`~pcapkit.corekit.fields.misc.SwitchField`,
# via :func:`mptcp_data_selector`) hands each subtype schema below the
# *whole* option -- ``kind``, ``length`` and all -- starting at the same
# first octet ``_MPTCP.test`` peeked and rewound past, rather than the
# bytes left over after some outer field already consumed a header. So
# unlike :class:`MPTCP`'s siblings that inherit :class:`Option`, which
# declares ``kind``/``length`` for exactly this reason, this base class
# has to declare its own -- unconditionally, since every Multipath TCP
# subtype carries an explicit length octet (no EOOL/NOP-style exception
# applies here). Every subclass's own leading field starts at the third
# octet as a result, which is where its ``test`` (subtype/flags) field
# expects to read from.
#
# Without them, packing would reject ``kind=``/``length=`` from the
# ``_make_mptcp_*`` makers with ``UnknownFieldWarning`` and then
# ``KeyError: 'length'`` the moment a sibling field's condition (e.g.
# ``MPTCPAddAddress.port``, ``MPTCPCapable.rkey``) read ``pkt['length']``;
# unpacking would misread the ``kind`` octet as the subtype/flags octet,
# since nothing had consumed it first.
#: Option kind.
kind: 'Enum_Option' = EnumField(length=1, namespace=Enum_Option)
#: MPTCP length.
length: 'int' = UInt8Field()
# NOTE: ``subtype`` stays an annotation rather than becoming a real field
# alongside ``kind``/``length`` above -- deliberately, and the difference
# from those fields is why. ``kind`` and ``length`` are each the
# *sole* source of their own octet: nothing else in the schema packs them,
# so declaring them as real fields was the only way to get them onto the
# wire at all. ``subtype`` is not like that: every concrete subclass
# already encodes it as 4 bits of its own ``test`` :class:`BitField` (e.g.
# ``MPTCPCapable.test['subtype']``), which is what actually gets packed.
# A real ``Field`` for ``subtype`` on top of that would either pack the
# same 4 bits twice under two names, or need a "derive, don't pack" kind
# of field that this library's :mod:`~pcapkit.corekit.fields` does not
# have. So this attribute is populated by the *construction* path instead
# -- :meth:`~pcapkit.protocols.transport.tcp.TCP._make_mode_mp` sets it
# right after building the subtype-specific schema, mirroring exactly what
# :meth:`_MPTCP.post_process` already does for real unpacking.
if TYPE_CHECKING:
#: MPTCP subtype.
subtype: 'Enum_MPTCPOption'
[docs]
@schema_final
class MPTCPUnknown(MPTCP):
"""Header schema for unknown Multipath TCP option."""
#: Subtype and data.
test: 'MPTCPSubtypeUnknown' = BitField(length=1, namespace={
'subtype': (0, 4),
'data': (4, 4),
})
#: Data.
data: 'bytes' = BytesField(length=lambda pkt: pkt['length'] - 3)
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int', test: 'MPTCPSubtypeUnknown', data: 'bytes') -> 'None': ...
[docs]
@schema_final
class MPTCPCapable(MPTCP, code=Enum_MPTCPOption.MP_CAPABLE):
"""Header schema for Multipath TCP capable option."""
#: Subtype and version.
test: 'MPTCPSubtypeCapable' = BitField(length=1, namespace={
'subtype': (0, 4),
'version': (4, 4),
})
#: Flags.
flags: 'MPTCPCapableFlags' = BitField(length=1, namespace={
'req': (0, 1),
'ext': (1, 1),
'deny_join': (2, 1),
'reserved': (3, 4),
'hsa': (7, 1),
})
# NOTE: :rfc:`8684` section 3.1 figure 4 gives MP_CAPABLE five lengths --
# 4 (SYN, header only), 12 (sender's key), 20 (both keys), 22 (both keys
# and Data-Level Length) and 24 (that plus the Checksum) -- so each field
# below is present from the length that first carries it.
#: Option sender's key (if option length > 4).
skey: 'int' = ConditionalField(
UInt64Field(),
lambda pkt: pkt['length'] > 4,
)
#: Option receiver's key (if option length > 12).
rkey: 'int' = ConditionalField(
UInt64Field(),
lambda pkt: pkt['length'] > 12,
)
#: Data-level length (if option length > 20).
dl_len: 'int' = ConditionalField(
UInt16Field(),
lambda pkt: pkt['length'] > 20,
)
#: Checksum (if option length > 22).
checksum: 'bytes' = ConditionalField(
BytesField(length=2),
lambda pkt: pkt['length'] > 22,
)
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int', test: 'MPTCPSubtypeCapable', flags: 'MPTCPCapableFlags', skey: 'Optional[int]', rkey: 'Optional[int]', dl_len: 'Optional[int]', checksum: 'Optional[bytes]') -> 'None': ...
[docs]
class MPTCPJoin(MPTCP, code=Enum_MPTCPOption.MP_JOIN): # register as a placeholder
"""Header schema for Multipath TCP join option."""
[docs]
@schema_final
class MPTCPJoinSYN(MPTCPJoin):
"""Header schema for Multipath TCP join option for ``SYN`` connection."""
#: Subtype and flags.
test: 'MPTCPSubtypeJoin' = BitField(length=1, namespace={
'subtype': (0, 4),
'reserved': (4, 3),
'backup': (7, 1),
})
#: Address ID.
addr_id: 'int' = UInt8Field()
#: Receiver's token.
token: 'int' = UInt32Field()
#: Sender's random number.
nonce: 'int' = UInt32Field()
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int', test: 'MPTCPSubtypeJoin', addr_id: 'int', token: 'int', nonce: 'int') -> 'None': ...
[docs]
@schema_final
class MPTCPJoinSYNACK(MPTCPJoin):
"""Header schema for Multipath TCP join option for ``SYN/ACK`` connection."""
#: Subtype and flags.
test: 'MPTCPSubtypeJoin' = BitField(length=1, namespace={
'subtype': (0, 4),
'reserved': (4, 3),
'backup': (7, 1),
})
#: Address ID.
addr_id: 'int' = UInt8Field()
#: Sender's truncated HMAC
hmac: 'bytes' = BytesField(length=8)
#: Sender's random number.
nonce: 'int' = UInt32Field()
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int', test: 'MPTCPSubtypeJoin', addr_id: 'int', hmac: 'bytes', nonce: 'int') -> 'None': ...
[docs]
@schema_final
class MPTCPJoinACK(MPTCPJoin):
"""Header schema for Multipath TCP join option for ``ACK`` connection."""
#: Subtype and reserved.
test: 'MPTCPSubtype' = BitField(length=2, namespace={
'subtype': (0, 4),
'reserved': (4, 12),
})
#: Sender's HMAC.
hmac: 'bytes' = BytesField(length=20)
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int', test: 'MPTCPSubtype', hmac: 'bytes') -> 'None': ...
[docs]
@schema_final
class MPTCPDSS(MPTCP, code=Enum_MPTCPOption.DSS):
"""Header schema for Multipath TCP DSS option."""
#: Subtype and reserved.
test: 'MPTCPSubtype' = BitField(length=1, namespace={
'subtype': (0, 4),
'reserved': (4, 4),
})
#: Flags.
flags: 'MPTCPDSSFlags' = BitField(length=1, namespace={
'reserved': (0, 3),
'F': (3, 1),
'm': (4, 1),
'M': (5, 1),
'a': (6, 1),
'A': (7, 1),
})
#: Data ACK.
#:
#: 4 octets when ``A`` is set, 8 when ``a`` is set as well, absent otherwise --
#: :rfc:`8684` section 3.3 figure 9. Both the presence test and the width live
#: in :func:`mptcp_dss_ack_selector`, whose note explains why the switch form
#: is used.
ack: 'int' = SwitchField(
selector=mptcp_dss_ack_selector,
)
#: Data sequence number.
#:
#: 4 octets when ``M`` is set, 8 when ``m`` is set as well, absent otherwise --
#: :rfc:`8684` section 3.3 figure 9. C.f. :func:`mptcp_dss_dsn_selector`.
dsn: 'int' = SwitchField(
selector=mptcp_dss_dsn_selector,
)
#: Subflow sequence number.
ssn: 'int' = ConditionalField(
UInt32Field(),
lambda pkt: pkt['flags']['M'],
)
#: Data level length.
dl_len: 'int' = ConditionalField(
UInt16Field(),
lambda pkt: pkt['flags']['M'],
)
#: Checksum (if negotiated, c.f. :func:`mptcp_dss_checksum`).
checksum: 'bytes' = ConditionalField(
BytesField(length=2),
mptcp_dss_checksum,
)
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int', test: 'MPTCPSubtype', flags: 'MPTCPDSSFlags', ack: 'Optional[int]', dsn: 'Optional[int]', ssn: 'Optional[int]', dl_len: 'Optional[int]', checksum: 'Optional[bytes]') -> 'None': ...
[docs]
@schema_final
class MPTCPAddAddress(MPTCP, code=Enum_MPTCPOption.ADD_ADDR):
"""Header schema for Multipath TCP add address option."""
#: Subtype and flags.
test: 'MPTCPSubtypeAddAddress' = BitField(length=1, namespace={
'subtype': (0, 4),
'reserved': (4, 3),
'echo': (7, 1),
})
#: Address ID.
addr_id: 'int' = UInt8Field()
#: Address.
address: 'IPv4Address | IPv6Address' = SwitchField(
selector=mptcp_add_address_selector,
)
#: Port.
port: 'int' = ConditionalField(
UInt16Field(),
lambda pkt: pkt['length'] in (10, 18, 22, 30),
)
#: Truncated HMAC.
hmac: 'bytes' = ConditionalField(
BytesField(length=8),
lambda pkt: pkt['length'] in (16, 18, 28, 30),
)
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int', test: 'MPTCPSubtypeAddAddress', addr_id: 'int', address: 'IPv4Address | IPv6Address', port: 'Optional[int]', hmac: 'Optional[bytes]') -> 'None': ...
[docs]
@schema_final
class MPTCPRemoveAddress(MPTCP, code=Enum_MPTCPOption.REMOVE_ADDR):
"""Header schema for Multipath TCP remove address option."""
#: Subtype and reserved.
test: 'MPTCPSubtype' = BitField(length=1, namespace={
'subtype': (0, 4),
'reserved': (4, 4),
})
#: Address ID.
addr_id: 'list[int]' = ListField(
length=lambda pkt: pkt['length'] - 3,
item_type=UInt8Field(),
)
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int', test: 'MPTCPSubtype', addr_id: 'list[int]') -> 'None': ...
[docs]
@schema_final
class MPTCPPriority(MPTCP, code=Enum_MPTCPOption.MP_PRIO):
"""Header schema for Multipath TCP priority option."""
#: Subtype.
test: 'MPTCPSubtypePriority' = BitField(length=1, namespace={
'subtype': (0, 4),
'reserved': (4, 3),
'backup': (7, 1),
})
#: Address ID.
addr_id: 'int' = ConditionalField(
UInt8Field(),
lambda pkt: pkt['length'] == 4,
)
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int', test: 'MPTCPSubtypePriority', addr_id: 'Optional[int]') -> 'None': ...
[docs]
@schema_final
class MPTCPFallback(MPTCP, code=Enum_MPTCPOption.MP_FAIL):
"""Header schema for Multipath TCP fallback option."""
#: Subtype and reserved.
#:
#: :rfc:`8684` section 3.7 figure 16 follows the subtype's 4 bits with
#: **12** reserved bits, so the Data Sequence Number starts at the fifth
#: octet of the 12-octet option.
test: 'MPTCPSubtype' = BitField(length=2, namespace={
'subtype': (0, 4),
'reserved': (4, 12),
})
#: Data sequence number.
dsn: 'int' = UInt64Field()
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int', test: 'MPTCPSubtype', dsn: 'int') -> 'None': ...
[docs]
@schema_final
class MPTCPFastclose(MPTCP, code=Enum_MPTCPOption.MP_FASTCLOSE):
"""Header schema for Multipath TCP fastclose option."""
#: Subtype and reserved.
#:
#: :rfc:`8684` section 3.5 figure 14 spends a whole 32-bit row on
#: ``Kind``/``Length``/``Subtype``/``(reserved)``, i.e. the subtype's 4 bits
#: are followed by **12** reserved bits, not 4 -- so the subtype-and-reserved
#: part is 2 octets and the option is 12 octets in total.
test: 'MPTCPSubtype' = BitField(length=2, namespace={
'subtype': (0, 4),
'reserved': (4, 12),
})
#: Option receiver's key.
key: 'int' = UInt64Field()
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int', test: 'MPTCPSubtype', key: 'int') -> 'None': ...
[docs]
@schema_final
class FastOpenCookie(Option, code=Enum_Option.TCP_Fast_Open_Cookie):
"""Header schema for TCP Fast Open option."""
#: Cookie.
cookie: 'bytes' = ConditionalField(
BytesField(length=lambda pkt: pkt['length'] - 2),
lambda pkt: pkt['length'] >= 6,
)
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int', cookie: 'Optional[bytes]') -> 'None': ...
[docs]
@schema_final
class TCP(Schema):
"""Header schema for TCP packet."""
#: Source port.
srcport: 'Enum_AppType' = PortEnumField(length=2, namespace=Enum_AppType,
proto=Enum_TransportProtocol.tcp)
#: Destination port.
dstport: 'Enum_AppType' = PortEnumField(length=2, namespace=Enum_AppType,
proto=Enum_TransportProtocol.tcp)
#: Sequence number.
seq: 'int' = UInt32Field()
#: Acknowledgement number.
ack: 'int' = UInt32Field()
#: Data offset.
offset: 'OffsetFlag' = BitField(length=1, namespace={
'offset': (0, 4),
'reserved': (4, 3),
'ns': (7, 1),
})
#: TCP flags.
flags: 'Flags' = BitField(length=1, namespace={
'cwr': (0, 1),
'ece': (1, 1),
'urg': (2, 1),
'ack': (3, 1),
'psh': (4, 1),
'rst': (5, 1),
'syn': (6, 1),
'fin': (7, 1),
})
#: Window size.
window: 'int' = UInt16Field()
#: Checksum.
checksum: 'bytes' = BytesField(length=2)
#: Urgent pointer.
urgent: 'int' = UInt16Field()
#: Options.
options: 'list[Option]' = OptionField(
length=lambda pkt: pkt['offset']['offset'] * 4 - TCP_MIN_HEADER_LEN,
base_schema=Option,
type_name='kind',
registry=Option.registry,
eool=Enum_Option.End_of_Option_List,
)
#: Padding.
padding: 'bytes' = PaddingField(length=lambda pkt: pkt.get('__option_padding__', 0)) # key generated by OptionField
#: Payload.
payload: 'bytes' = PayloadField()
if TYPE_CHECKING:
def __init__(self, srcport: 'Enum_AppType | int', dstport: 'Enum_AppType | int', seq: 'int', ack: 'int',
offset: 'OffsetFlag', flags: 'Flags', window: 'int', checksum: 'bytes',
urgent: 'int', options: 'list[Option | bytes] | bytes', payload: 'bytes | ProtocolBase | Schema') -> 'None': ...