# -*- 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, 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 BaseError, 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
if SPHINX_TYPE_CHECKING: # pragma: no cover
from typing_extensions import TypedDict
[docs]
class OffsetFlag(TypedDict):
"""TCP offset field flag."""
#: Data offset.
offset: int
#: ECN-nonce concealment protection.
ns: int
[docs]
class Flags(TypedDict):
"""TCP flags."""
#: Congestion window reduced.
cwr: int
#: ECN-Echo.
ece: int
#: Urgent pointer.
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
[docs]
class QuickStartFlags(TypedDict):
"""TCP quick start flags."""
#: Rate request.
rate: int
[docs]
class QuickStartNonce(TypedDict):
"""TCP quick start nonce."""
#: Nonce.
nonce: 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
#: Use of HMAC-SHA1.
hsa: int
[docs]
class MPTCPSubtypeJoin(TypedDict):
"""MPTCP Join subtype field."""
#: Subtype.
subtype: int
#: Backup flag.
backup: int
[docs]
class MPTCPSubtype(TypedDict):
"""MPTCP subtype field."""
#: Subtype.
subtype: int
[docs]
class MPTCPDSSFlags(TypedDict):
"""MPTCP-DSS flags."""
#: ``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
#: IP version.
version: int
[docs]
class MPTCPSubtypePriority(TypedDict):
"""MPTCP Priority subtype field."""
#: Subtype.
subtype: 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.
"""
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:
* If IP version is 4, a :class:`~pcapkit.corekit.fields.ipaddress.IPv4AddressField`
instance.
* If IP version is 6, a :class:`~pcapkit.corekit.fields.ipaddress.IPv6AddressField`
instance.
"""
if pkt['test']['version'] == 4:
return IPv4AddressField()
if pkt['test']['version'] == 6:
return IPv6AddressField()
raise FieldError(f'TCP: [OptNo {Enum_Option.Multipath_TCP}] {Enum_MPTCPOption.ADD_ADDR} invalid IP version')
[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:
* If ``A`` is clear, a :class:`~pcapkit.corekit.fields.misc.NoValueField`
instance -- the field is absent from the wire.
* If ``A`` is set and ``a`` is set, a
:class:`~pcapkit.corekit.fields.numbers.UInt64Field` instance.
* If ``A`` is set and ``a`` is clear, a
:class:`~pcapkit.corekit.fields.numbers.UInt32Field` instance.
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: ...)``, which is what it was
until #576.
The width lambda read ``8 if pkt['flags']['a'] else 0`` -- **0**, not 4 --
so an unextended Data ACK packed no octets at all while the ``length``
octet still counted 4 for it. That is the defect #576 records: the option
went onto the wire 4 (or 8, with ``dsn`` too) octets shorter than it
declared, and the ``ack`` value the caller supplied was simply not
present.
Correcting the lambda to ``8 if ... else 4`` would not have worked *at the
time*, because :class:`~pcapkit.corekit.fields.numbers.NumberField` could
not pack a callable length at all: it called ``build_template`` once at
``__init__`` with the placeholder length ``-1``, which latched
``_need_process = True``, and nothing cleared that flag when ``__call__``
later resolved the real length and rebuilt the template as ``>I``/``>Q``.
``pre_process`` then handed :func:`struct.pack` bytes for an integer
template and it raised ``struct.error: required argument is not an
integer``. Measured on the 8-octet form, which the old lambda did reach:
``_make_mptcp_dss(DSS, ack=1 << 40)`` raised exactly that.
That half is now history: **#598 fixed it**, in
:mod:`pcapkit.corekit.fields.numbers` where this note used to say the fix
belonged, by recomputing ``_need_process`` from the width actually in
force instead of once from the placeholder. A callable-length
``NumberField`` packs and unpacks both the 4- and the 8-octet form today,
so ``ConditionalField(NumberField(length=...), lambda pkt:
pkt['flags']['A'])`` would express this field correctly. Nor was wire
*absence* ever the obstacle: :attr:`MPTCPDSS.ssn`, :attr:`MPTCPDSS.dl_len`
and :attr:`MPTCPDSS.checksum` are each a ``ConditionalField`` on the
sibling ``M`` flag, so this very class already leans on that wrapper to
keep a field off the wire.
The ``SwitchField`` form is kept anyway, for a narrower reason about
composition rather than about absence. 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 name 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, and #603 does not make it.
"""
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:
* If ``M`` is clear, a :class:`~pcapkit.corekit.fields.misc.NoValueField`
instance -- the field is absent from the wire.
* If ``M`` is set and ``m`` is set, a
:class:`~pcapkit.corekit.fields.numbers.UInt64Field` instance.
* If ``M`` is set and ``m`` is clear, a
:class:`~pcapkit.corekit.fields.numbers.UInt32Field` instance.
Note:
Identical in shape to :func:`mptcp_dss_ack_selector`, and it replaces the
identical defect: ``NumberField(length=lambda pkt: 8 if pkt['flags']['m']
else 0, ...)``. See that function's note for why the ``0`` was wrong, why a
corrected lambda would not have packed either *at the time*, and why the
``SwitchField`` form is kept now that #598 has made a callable length work.
C.f. #576, #598.
"""
if not pkt['flags']['M']:
return NoValueField()
return UInt64Field() if pkt['flags']['m'] else UInt32Field()
class PortEnumField(EnumField):
"""Enumerated value for protocol fields.
Args:
length: Field size (in bytes); if a callable is given, it should return
an integer value and accept the current packet as its only argument.
default: Field default value, if any.
signed: Whether the field is signed.
byteorder: Field byte order.
bit_length: Field bit length.
callback: Callback function to be called upon
:meth:`self.__call__ <pcapkit.corekit.fields.field.FieldBase.__call__>`.
Important:
This class is specifically designed for :class:`~pcapkit.const.reg.apptype.AppType`
as it is actually a :class:`~enum.StrEnum` class.
"""
if TYPE_CHECKING:
_namespace: 'Enum_AppType'
def pre_process(self, value: 'int | Enum_AppType', packet: 'dict[str, Any]') -> 'int | bytes':
"""Process field value before construction (packing).
Arguments:
value: Field value.
packet: Packet data.
Returns:
Processed field value.
"""
if isinstance(value, Enum_AppType):
value = value.port
return super().pre_process(value, packet)
def post_process(self, value: 'int | bytes', packet: 'dict[str, Any]') -> 'Enum_AppType':
"""Process field value after parsing (unpacked).
Args:
value: Field value.
packet: Packet data.
Returns:
Processed field value -- the registry member declared for the
port, or an unregistered member of the same registry, carrying
the port itself, when the registry declares none. See GitHub
issue #575.
Notes:
:meth:`~pcapkit.const.reg.apptype.AppType.get` mints a fresh
member -- via :func:`aenum.extend_enum` -- for any port neither an
existing row nor one of :meth:`_missing_`'s documented IANA spans
accounts for, which in practice means the ephemeral/dynamic range.
Calling it unconditionally on every parsed port therefore grew the
registry without bound. This peeks at the registry
:meth:`~pcapkit.const.reg.apptype.AppType.get` itself would
consult -- its per-port rows via ``__registry__.getlist``, then
its documented spans via ``_missing_`` -- and only calls
:meth:`~pcapkit.const.reg.apptype.AppType.get` once one of those is
already known to hold, so a genuine miss gets
:meth:`EnumField._unregistered_member` instead of a mint.
A port outside this field's own width is rejected *before* any of
that, rather than being let through to :meth:`_missing_` and
caught alongside a genuine miss. Both are a bare :exc:`ValueError`
with nothing to tell them apart by type, and GitHub issue #764
gave the out-of-range case a deliberate, ``breaking``-tagged
rejection specifically so it would stop being minted over -- a
catch keyed on exception type alone cannot see the difference
between that and :mod:`aenum`'s own "no member has this value",
so it would absorb both and quietly revert #764 for these four
fields. Checking the width first needs no exception-based
distinction at all: it asks the same question :meth:`_missing_`
would eventually ask, and asks it in a way that never manufactures
the bare :exc:`ValueError` this method would otherwise have to
tell apart from a foreign one. ``self.length`` is the field's own
declared byte width (``2`` for every caller of this class, hence
``0``-``65535``) rather than a hard-coded ``65535`` borrowed from
:meth:`~pcapkit.const.reg.apptype.AppType._missing_`'s own guard,
so the two stay in lockstep by construction: whatever width this
field is ever given, the range checked here is exactly the range a
value of that width can hold, no more permissive and no more
restrictive.
"""
value = super(EnumField, self).post_process(value, packet)
proto = Enum_TransportProtocol.tcp
if not (isinstance(value, int) and 0 <= value < (1 << (8 * self.length))):
# NOTE: lets AppType.get() -- unmodified -- raise #764's rejection
# for a port this field's own width cannot represent, rather than
# risking it being absorbed below as a foreign miss.
return self._namespace.get(value, proto=proto)
owner = self._namespace._dispatch(value, proto) # pylint: disable=protected-access
if not owner.__registry__.getlist(value): # type: ignore[union-attr]
try:
declared = owner._missing_(value) # pylint: disable=protected-access
except ValueError as error:
# NOTE: value is already known to be in-width here, so this
# ValueError is aenum's own "no member has this value" for an
# in-range but unassigned port -- a foreign miss, absorbed --
# never #764's out-of-range rejection, which never reaches
# this branch. A pcapkit.utilities.exceptions error is still a
# deliberate registry decision and propagates unchanged.
if isinstance(error, BaseError):
raise
declared = None
if declared is None:
# NOTE: an unregistered member of ``owner`` itself, per GitHub
# issue #575's owner ruling -- see EnumField._unregistered_member
# -- rather than a foreign pseudo-enum. ``.port``, ``.svc`` and
# ``.proto`` are what a real AppType member carries -- read
# unconditionally by e.g. Transport._decode_next_layer's
# ``srcport.port`` -- and ``svc='unknown'`` matches what the
# mint this replaces used to name it. They are passed in the
# order AppType.__new__ sets them, because
# DictDumper.object_hook renders a member's addon keys straight
# out of its ``__dict__`` in insertion order -- so any other
# order here would make an unassigned port dump ``port`` before
# ``svc`` while every declared one dumps ``svc`` first.
return self._unregistered_member(
owner, f'unknown [{value:d} - {proto.name}]',
svc='unknown', port=value, proto=proto)
return self._namespace.get(value, proto=proto)
[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': ...
@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."""
#: 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."""
#: Profile data.
profile: 'POCProfile' = BitField(length=1, namespace={
'start': (0, 1),
'end': (1, 1),
})
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={
'rate': (4, 4),
})
#: TTL difference.
diff: 'int' = UInt8Field()
#: QS nonce.
nonce: 'QuickStartNonce' = BitField(length=4, namespace={
'nonce': (0, 30),
})
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': ...
@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 :data:`Option.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. It read from bit 1 until #553, straddling the low
#: seven bits of ``kind`` and the high bit of ``length``, so a 12-octet
#: MP_CAPABLE (``1e 0c 01``) decoded its length as 60.
test: 'MPTCPSubtypeTest' = ForwardMatchField(BitField(length=3, namespace={
'length': (8, 8),
'subtype': (16, 4),
}))
#: Subtype-specific data.
data: 'MPTCP' = SwitchField(
selector=mptcp_data_selector,
)
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."""
__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.
#
# Before this, both directions were broken: packing rejected ``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 silently misread the ``kind`` octet as the
# subtype/flags octet, since nothing had consumed it first. C.f. #541.
#: 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 third real
# field alongside ``kind``/``length`` above -- deliberately, and the
# difference from those two 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. C.f. #566,
# the third and last ``TYPE_CHECKING``-only attribute this class had; the
# other two (``kind``, ``length``) were fixed in #541.
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'] - 2)
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),
'hsa': (7, 1),
})
#: Option sender's key.
skey: 'int' = UInt64Field()
#: Option receiver's key.
#:
#: :rfc:`8684` section 3.1 gives MP_CAPABLE as 12 octets without this key
#: and 20 octets with it, so the field is present only for the latter --
#: not, as it read until #567, for every length *except* 32, which is not
#: an MP_CAPABLE length either RFC form uses.
rkey: 'int' = ConditionalField(
UInt64Field(),
lambda pkt: pkt['length'] == 20,
)
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int', test: 'MPTCPSubtypeCapable', flags: 'MPTCPCapableFlags', skey: 'int', rkey: 'Optional[int]') -> '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),
'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),
'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.
test: 'MPTCPSubtype' = BitField(length=1, namespace={
'subtype': (0, 4),
})
#: Reserved.
reserved: 'bytes' = PaddingField(length=1)
#: 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 flags.
test: 'MPTCPSubtype' = BitField(length=1, namespace={
'subtype': (0, 4),
})
#: Flags.
flags: 'MPTCPDSSFlags' = BitField(length=1, namespace={
'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 records what this field
#: declared until #576 and why the switch form is kept.
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.
checksum: 'bytes' = ConditionalField(
BytesField(length=2),
lambda pkt: pkt['flags']['M'],
)
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 IP version.
test: 'MPTCPSubtypeAddAddress' = BitField(length=1, namespace={
'subtype': (0, 4),
'version': (4, 4),
})
#: 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, 22),
)
if TYPE_CHECKING:
def __init__(self, kind: 'Enum_Option', length: 'int', test: 'MPTCPSubtypeAddAddress', addr_id: 'int', address: 'IPv4Address | IPv6Address', port: 'Optional[int]') -> 'None': ...
[docs]
@schema_final
class MPTCPRemoveAddress(MPTCP, code=Enum_MPTCPOption.REMOVE_ADDR):
"""Header schema for Multipath TCP remove address option."""
#: Subtype.
test: 'MPTCPSubtype' = BitField(length=1, namespace={
'subtype': (0, 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),
'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.
test: 'MPTCPSubtype' = BitField(length=1, namespace={
'subtype': (0, 4),
})
#: 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.
test: 'MPTCPSubtype' = BitField(length=1, namespace={
'subtype': (0, 4),
})
#: 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. Until #576 this
#: field did not exist and ``test`` was the only octet between ``length`` and
#: ``key``, so the schema packed **11** octets against a ``length`` of 12.
#: Declared the same way :class:`MPTCPJoinACK` declares its own reserved
#: octet, for the same reason: a wider ``test`` would make the reserved bits
#: look like part of the subtype namespace.
reserved: 'bytes' = PaddingField(length=1)
#: 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)
#: Destination port.
dstport: 'Enum_AppType' = PortEnumField(length=2, namespace=Enum_AppType)
#: Sequence number.
seq: 'int' = UInt32Field()
#: Acknowledgement number.
ack: 'int' = UInt32Field()
#: Data offset.
offset: 'OffsetFlag' = BitField(length=1, namespace={
'offset': (0, 4),
'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 - 20,
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': ...