# -*- coding: utf-8 -*-
# mypy: disable-error-code=dict-item
"""TCP - Transmission Control Protocol
=========================================
.. module:: pcapkit.protocols.transport.tcp.protocol
:mod:`pcapkit.protocols.transport.tcp.protocol` holds
:class:`~pcapkit.protocols.transport.tcp.protocol.TCP`, the extractor for the
Transmission Control Protocol (TCP) [*]_, whose header is:
======= ========= ========================= =======================================
Octets Bits Name Description
======= ========= ========================= =======================================
0 0 ``tcp.srcport`` Source Port
2 16 ``tcp.dstport`` Destination Port
4 32 ``tcp.seq`` Sequence Number
8 64 ``tcp.ack`` Acknowledgement Number (if ACK set)
12 96 ``tcp.hdr_len`` Data Offset
12 100 ``tcp.flags.reserved`` Reserved (must be ``\\x00``)
12 103 ``tcp.flags.ns`` ECN Concealment Protection (NS)
13 104 ``tcp.flags.cwr`` Congestion Window Reduced (CWR)
13 105 ``tcp.flags.ece`` ECN-Echo (ECE)
13 106 ``tcp.flags.urg`` Urgent (URG)
13 107 ``tcp.flags.ack`` Acknowledgement (ACK)
13 108 ``tcp.flags.psh`` Push Function (PSH)
13 109 ``tcp.flags.rst`` Reset Connection (RST)
13 110 ``tcp.flags.syn`` Synchronize Sequence Numbers (SYN)
13 111 ``tcp.flags.fin`` Last Packet from Sender (FIN)
14 112 ``tcp.window_size`` Size of Receive Window
16 128 ``tcp.checksum`` Checksum
18 144 ``tcp.urgent_pointer`` Urgent Pointer (if URG set)
20 160 ``tcp.options`` TCP Options (if data offset > 5)
======= ========= ========================= =======================================
.. [*] https://en.wikipedia.org/wiki/Transmission_Control_Protocol
"""
import collections
import datetime
import math
from typing import TYPE_CHECKING, cast
from pcapkit.const.reg.apptype import AppType as Enum_AppType
from pcapkit.const.reg.apptype import TransportProtocol as Enum_TransportProtocol
from pcapkit.const.reg.transtype import TransType
from pcapkit.const.tcp.checksum import Checksum as Enum_Checksum
from pcapkit.const.tcp.flags import Flags as Enum_Flags
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.module import ModuleDescriptor
from pcapkit.corekit.multidict import OrderedMultiDict
from pcapkit.corekit.schema.ipaddress import parse_ip_address
from pcapkit.corekit.schema.padding import pad_length
from pcapkit.protocols.schema import Schema
from pcapkit.protocols.transport.tcp.data import CC as Data_CC
from pcapkit.protocols.transport.tcp.data import MPTCPDSS as Data_MPTCPDSS
from pcapkit.protocols.transport.tcp.data import SACK as Data_SACK
from pcapkit.protocols.transport.tcp.data import TCP as Data_TCP
from pcapkit.protocols.transport.tcp.data import AlternateChecksumData as Data_AlternateChecksumData
from pcapkit.protocols.transport.tcp.data import \
AlternateChecksumRequest as Data_AlternateChecksumRequest
from pcapkit.protocols.transport.tcp.data import Authentication as Data_Authentication
from pcapkit.protocols.transport.tcp.data import CCEcho as Data_CCEcho
from pcapkit.protocols.transport.tcp.data import CCNew as Data_CCNew
from pcapkit.protocols.transport.tcp.data import Echo as Data_Echo
from pcapkit.protocols.transport.tcp.data import EchoReply as Data_EchoReply
from pcapkit.protocols.transport.tcp.data import EndOfOptionList as Data_EndOfOptionList
from pcapkit.protocols.transport.tcp.data import FastOpenCookie as Data_FastOpenCookie
from pcapkit.protocols.transport.tcp.data import Flags as Data_Flags
from pcapkit.protocols.transport.tcp.data import MaximumSegmentSize as Data_MaximumSegmentSize
from pcapkit.protocols.transport.tcp.data import MD5Signature as Data_MD5Signature
from pcapkit.protocols.transport.tcp.data import MPTCPAddAddress as Data_MPTCPAddAddress
from pcapkit.protocols.transport.tcp.data import MPTCPCapable as Data_MPTCPCapable
from pcapkit.protocols.transport.tcp.data import MPTCPCapableFlag as Data_MPTCPCapableFlag
from pcapkit.protocols.transport.tcp.data import MPTCPFallback as Data_MPTCPFallback
from pcapkit.protocols.transport.tcp.data import MPTCPFastclose as Data_MPTCPFastclose
from pcapkit.protocols.transport.tcp.data import MPTCPJoinACK as Data_MPTCPJoinACK
from pcapkit.protocols.transport.tcp.data import MPTCPJoinSYN as Data_MPTCPJoinSYN
from pcapkit.protocols.transport.tcp.data import MPTCPJoinSYNACK as Data_MPTCPJoinSYNACK
from pcapkit.protocols.transport.tcp.data import MPTCPPriority as Data_MPTCPPriority
from pcapkit.protocols.transport.tcp.data import MPTCPRemoveAddress as Data_MPTCPRemoveAddress
from pcapkit.protocols.transport.tcp.data import MPTCPUnknown as Data_MPTCPUnknown
from pcapkit.protocols.transport.tcp.data import NoOperation as Data_NoOperation
from pcapkit.protocols.transport.tcp.data import \
PartialOrderConnectionPermitted as Data_PartialOrderConnectionPermitted
from pcapkit.protocols.transport.tcp.data import \
PartialOrderServiceProfile as Data_PartialOrderServiceProfile
from pcapkit.protocols.transport.tcp.data import QuickStartResponse as Data_QuickStartResponse
from pcapkit.protocols.transport.tcp.data import SACKBlock as Data_SACKBlock
from pcapkit.protocols.transport.tcp.data import SACKPermitted as Data_SACKPermitted
from pcapkit.protocols.transport.tcp.data import Timestamps as Data_Timestamps
from pcapkit.protocols.transport.tcp.data import UnassignedOption as Data_UnassignedOption
from pcapkit.protocols.transport.tcp.data import UserTimeout as Data_UserTimeout
from pcapkit.protocols.transport.tcp.data import WindowScale as Data_WindowScale
from pcapkit.protocols.transport.tcp.schema import CC as Schema_CC
from pcapkit.protocols.transport.tcp.schema import MPTCPDSS as Schema_MPTCPDSS
from pcapkit.protocols.transport.tcp.schema import SACK as Schema_SACK
from pcapkit.protocols.transport.tcp.schema import TCP as Schema_TCP
from pcapkit.protocols.transport.tcp.schema import TCP_MIN_HEADER_LEN
from pcapkit.protocols.transport.tcp.schema import \
AlternateChecksumData as Schema_AlternateChecksumData
from pcapkit.protocols.transport.tcp.schema import \
AlternateChecksumRequest as Schema_AlternateChecksumRequest
from pcapkit.protocols.transport.tcp.schema import Authentication as Schema_Authentication
from pcapkit.protocols.transport.tcp.schema import CCEcho as Schema_CCEcho
from pcapkit.protocols.transport.tcp.schema import CCNew as Schema_CCNew
from pcapkit.protocols.transport.tcp.schema import Echo as Schema_Echo
from pcapkit.protocols.transport.tcp.schema import EchoReply as Schema_EchoReply
from pcapkit.protocols.transport.tcp.schema import EndOfOptionList as Schema_EndOfOptionList
from pcapkit.protocols.transport.tcp.schema import FastOpenCookie as Schema_FastOpenCookie
from pcapkit.protocols.transport.tcp.schema import MaximumSegmentSize as Schema_MaximumSegmentSize
from pcapkit.protocols.transport.tcp.schema import MD5Signature as Schema_MD5Signature
from pcapkit.protocols.transport.tcp.schema import MPTCPAddAddress as Schema_MPTCPAddAddress
from pcapkit.protocols.transport.tcp.schema import MPTCPCapable as Schema_MPTCPCapable
from pcapkit.protocols.transport.tcp.schema import MPTCPFallback as Schema_MPTCPFallback
from pcapkit.protocols.transport.tcp.schema import MPTCPFastclose as Schema_MPTCPFastclose
from pcapkit.protocols.transport.tcp.schema import MPTCPJoinACK as Schema_MPTCPJoinACK
from pcapkit.protocols.transport.tcp.schema import MPTCPJoinSYN as Schema_MPTCPJoinSYN
from pcapkit.protocols.transport.tcp.schema import MPTCPJoinSYNACK as Schema_MPTCPJoinSYNACK
from pcapkit.protocols.transport.tcp.schema import MPTCPPriority as Schema_MPTCPPriority
from pcapkit.protocols.transport.tcp.schema import MPTCPRemoveAddress as Schema_MPTCPRemoveAddress
from pcapkit.protocols.transport.tcp.schema import MPTCPUnknown as Schema_MPTCPUnknown
from pcapkit.protocols.transport.tcp.schema import NoOperation as Schema_NoOperation
from pcapkit.protocols.transport.tcp.schema import \
PartialOrderConnectionPermitted as Schema_PartialOrderConnectionPermitted
from pcapkit.protocols.transport.tcp.schema import \
PartialOrderServiceProfile as Schema_PartialOrderServiceProfile
from pcapkit.protocols.transport.tcp.schema import QuickStartResponse as Schema_QuickStartResponse
from pcapkit.protocols.transport.tcp.schema import SACKBlock as Schema_SACKBlock
from pcapkit.protocols.transport.tcp.schema import SACKPermitted as Schema_SACKPermitted
from pcapkit.protocols.transport.tcp.schema import Timestamps as Schema_Timestamps
from pcapkit.protocols.transport.tcp.schema import UnassignedOption as Schema_UnassignedOption
from pcapkit.protocols.transport.tcp.schema import UserTimeout as Schema_UserTimeout
from pcapkit.protocols.transport.tcp.schema import WindowScale as Schema_WindowScale
from pcapkit.protocols.transport.transport import Transport
from pcapkit.utilities import config
from pcapkit.utilities.exceptions import ProtocolError
from pcapkit.utilities.warnings import RegistryWarning, warn
if TYPE_CHECKING:
from datetime import timedelta
from ipaddress import IPv4Address, IPv6Address
from typing import Any, Callable, ClassVar, DefaultDict, Optional, Type
from mypy_extensions import DefaultArg, KwArg, NamedArg
from typing_extensions import Literal
from pcapkit.corekit.enum import EnumRegistry
from pcapkit.protocols.misc.checksum.ip_pseudo.protocol import IP_Pseudo
from pcapkit.protocols.protocol import ProtocolBase
from pcapkit.protocols.transport.tcp.data import MPTCP as Data_MPTCP
from pcapkit.protocols.transport.tcp.data import MPTCPJoin as Data_MPTCPJoin
from pcapkit.protocols.transport.tcp.data import Option as Data_Option
from pcapkit.protocols.transport.tcp.schema import MPTCP as Schema_MPTCP
from pcapkit.protocols.transport.tcp.schema import Flags as Schema_Flags
from pcapkit.protocols.transport.tcp.schema import MPTCPJoin as Schema_MPTCPJoin
from pcapkit.protocols.transport.tcp.schema import Option as Schema_Option
Option = OrderedMultiDict[Enum_Option, Data_Option]
OptionParser = Callable[[Schema_Option, NamedArg(Option, 'options')], Data_Option]
MPOptionParser = Callable[[Schema_MPTCP, NamedArg(Option, 'options')], Data_MPTCP]
OptionConstructor = Callable[[Enum_Option, DefaultArg(Optional[Data_Option]),
KwArg(Any)], Schema_Option]
MPOptionConstructor = Callable[[Enum_MPTCPOption, DefaultArg(Optional[Data_MPTCP]),
KwArg(Any)], Schema_MPTCP]
__all__ = ['TCP']
#: The pseudo-header a payload's checksum is computed over in :meth:`TCP.make`,
#: resolved there rather than imported, so its first import holds
#: :data:`~pcapkit.utilities.importlib.import_lock` (:issue:`2771`).
_PSEUDO_HEADER: 'ModuleDescriptor[Any]' = \
ModuleDescriptor('pcapkit.protocols.misc.checksum.ip_pseudo.protocol', 'pseudo_header')
[docs]
class TCP(Transport[Data_TCP, Schema_TCP],
schema=Schema_TCP, data=Data_TCP):
"""This class implements Transmission Control Protocol.
The :attr:`self.__proto__ <pcapkit.protocols.transport.tcp.protocol.TCP.__proto__>`
registry binds these ports:
.. list-table::
:header-rows: 1
* - Port Number
- Protocol
* - 20
- :class:`pcapkit.protocols.application.ftp_data.protocol.FTP_DATA`
* - 21
- :class:`pcapkit.protocols.application.ftp.protocol.FTP`
* - 80
- :class:`pcapkit.protocols.application.http.protocol.HTTP`
* - 443
- :class:`pcapkit.protocols.application.tls.protocol.TLS`
* - 8080
- :class:`pcapkit.protocols.application.http.protocol.HTTP`
The TCP options it parses, by
:class:`pcapkit.const.tcp.option.Option` code:
.. list-table::
:header-rows: 1
* - Option Code
- Option Parser
- Option Constructor
* - :attr:`~pcapkit.const.tcp.option.Option.End_of_Option_List`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._read_mode_eool`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._make_mode_eool`
* - :attr:`~pcapkit.const.tcp.option.Option.No_Operation`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._read_mode_nop`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._make_mode_nop`
* - :attr:`~pcapkit.const.tcp.option.Option.Maximum_Segment_Size`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._read_mode_mss`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._make_mode_mss`
* - :attr:`~pcapkit.const.tcp.option.Option.Window_Scale`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._read_mode_ws`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._make_mode_ws`
* - :attr:`~pcapkit.const.tcp.option.Option.SACK_Permitted`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._read_mode_sackpmt`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._make_mode_sackpmt`
* - :attr:`~pcapkit.const.tcp.option.Option.SACK`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._read_mode_sack`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._make_mode_sack`
* - :attr:`~pcapkit.const.tcp.option.Option.Echo`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._read_mode_echo`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._make_mode_echo`
* - :attr:`~pcapkit.const.tcp.option.Option.Echo_Reply`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._read_mode_echore`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._make_mode_echore`
* - :attr:`~pcapkit.const.tcp.option.Option.Timestamps`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._read_mode_ts`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._make_mode_ts`
* - :attr:`~pcapkit.const.tcp.option.Option.Partial_Order_Connection_Permitted`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._read_mode_poc`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._make_mode_poc`
* - :attr:`~pcapkit.const.tcp.option.Option.Partial_Order_Service_Profile`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._read_mode_pocsp`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._make_mode_pocsp`
* - :attr:`~pcapkit.const.tcp.option.Option.CC`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._read_mode_cc`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._make_mode_cc`
* - :attr:`~pcapkit.const.tcp.option.Option.CC_NEW`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._read_mode_ccnew`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._make_mode_ccnew`
* - :attr:`~pcapkit.const.tcp.option.Option.CC_ECHO`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._read_mode_ccecho`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._make_mode_ccecho`
* - :attr:`~pcapkit.const.tcp.option.Option.TCP_Alternate_Checksum_Request`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._read_mode_chkreq`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._make_mode_chkreq`
* - :attr:`~pcapkit.const.tcp.option.Option.TCP_Alternate_Checksum_Data`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._read_mode_chksum`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._make_mode_chksum`
* - :attr:`~pcapkit.const.tcp.option.Option.MD5_Signature_Option`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._read_mode_sig`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._make_mode_sig`
* - :attr:`~pcapkit.const.tcp.option.Option.Quick_Start_Response`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._read_mode_qs`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._make_mode_qs`
* - :attr:`~pcapkit.const.tcp.option.Option.User_Timeout_Option`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._read_mode_timeout`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._make_mode_timeout`
* - :attr:`~pcapkit.const.tcp.option.Option.TCP_Authentication_Option`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._read_mode_ao`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._make_mode_ao`
* - :attr:`~pcapkit.const.tcp.option.Option.Multipath_TCP`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._read_mode_mp`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._make_mode_mp`
* - :attr:`~pcapkit.const.tcp.option.Option.TCP_Fast_Open_Cookie`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._read_mode_fastopen`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._make_mode_fastopen`
The Multipath TCP options it parses, by
:class:`pcapkit.const.tcp.mp_tcp_option.MPTCPOption` subtype:
.. list-table::
:header-rows: 1
* - Option Code
- Option Parser
- Option Constructor
* - :attr:`~pcapkit.const.tcp.mp_tcp_option.MPTCPOption.MP_CAPABLE`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._read_mptcp_capable`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._make_mptcp_capable`
* - :attr:`~pcapkit.const.tcp.mp_tcp_option.MPTCPOption.MP_JOIN`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._read_mptcp_join`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._make_mptcp_join`
* - :attr:`~pcapkit.const.tcp.mp_tcp_option.MPTCPOption.DSS`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._read_mptcp_dss`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._make_mptcp_dss`
* - :attr:`~pcapkit.const.tcp.mp_tcp_option.MPTCPOption.ADD_ADDR`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._read_mptcp_addaddr`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._make_mptcp_addaddr`
* - :attr:`~pcapkit.const.tcp.mp_tcp_option.MPTCPOption.REMOVE_ADDR`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._read_mptcp_remove`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._make_mptcp_remove`
* - :attr:`~pcapkit.const.tcp.mp_tcp_option.MPTCPOption.MP_PRIO`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._read_mptcp_prio`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._make_mptcp_prio`
* - :attr:`~pcapkit.const.tcp.mp_tcp_option.MPTCPOption.MP_FAIL`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._read_mptcp_fail`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._make_mptcp_fail`
* - :attr:`~pcapkit.const.tcp.mp_tcp_option.MPTCPOption.MP_FASTCLOSE`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._read_mptcp_fastclose`
- :meth:`~pcapkit.protocols.transport.tcp.protocol.TCP._make_mptcp_fastclose`
.. versionchanged:: 1.5.0
* :meth:`read` reports the :attr:`connection` flags of a flagless
segment, and ``info.connection``, as an empty
:class:`~pcapkit.const.tcp.flags.Flags` rather than a plain
:obj:`int` (:issue:`616`).
* A constructed MPTCP option sets its ``subtype`` (:issue:`566`), and
MP_CAPABLE is written at the :rfc:`8684` lengths (:issue:`567`) and
read at each of them (:issue:`1042`).
* A rebuild keeps what was parsed: the option order and padding
(:issue:`1151`), the reserved bits of the header and of the POCSP,
Quick-Start and MPTCP options (:issue:`1211`, :issue:`1212`), the
octets after EOOL (:issue:`1218`), the User Timeout granularity
(:issue:`1219`), and the MP_CAPABLE ``C`` flag and ``D``-``G`` bits
(:issue:`1220`).
* :meth:`read` refuses a Data Offset that declares more header than the
data holds, rather than reading the missing octets as zeros
(:issue:`1404`).
* MPTCP parsing follows the flags and the option length: MP_FAIL reads
its DSN at the right offset (:issue:`1213`), an unknown subtype stays
within its option (:issue:`1214`), DSS sizes its Data ACK and DSN from
the ``a`` and ``m`` flags and reads a checksum only when one is
present (:issue:`1215`, :issue:`1216`), and ADD_ADDR parses the
:rfc:`8684` layout (:issue:`1217`).
.. versionchanged:: 1.6.0
* Port 443 binds
:class:`~pcapkit.protocols.application.tls.protocol.TLS`
(:issue:`2437`).
* The checksum is computed when :meth:`make` is given none, over the
``pseudo`` it is given or by the enclosing IP layer, and verified by
that layer under :data:`~pcapkit.utilities.config.CHECKSUM_VERIFY`
(:issue:`2473`).
* A Quick-Start rate the four-bit rate code cannot carry, or a nonce or
reserved value outside its field, raises
:exc:`~pcapkit.utilities.exceptions.ProtocolError`, rather than a bare
:exc:`ValueError`, an :exc:`OverflowError` or a
:exc:`~pcapkit.utilities.exceptions.FieldValueError`, or a rate of
zero written in its place (:issue:`2617`).
* :meth:`make` and the option makers raise
:exc:`~pcapkit.utilities.exceptions.ProtocolError` for an integer
field given a value its width cannot hold, rather than writing its low
bits or raising a :exc:`~pcapkit.utilities.exceptions.FieldValueError`
or a bare :exc:`ValueError`, and for a fixed-size octet field given
the wrong number of octets, rather than padding or cutting them. A
User Timeout that is neither an :class:`int` nor a
:class:`~datetime.timedelta` raises the same, rather than an
:exc:`AttributeError`. The Quick-Start TTL Diff wraps, as
:rfc:`4782` defines it modulo 256 (:issue:`2614`).
* Each enumeration field of the option makers, the alternate checksum
``algorithm`` and the Multipath TCP ``subtype``, and the kind of each
option given to :meth:`make`, takes a member or an :class:`int` code,
resolved with the enumeration's ``get``; a name, a :obj:`bool`, or a
code the field cannot carry raises
:exc:`~pcapkit.utilities.exceptions.ProtocolError`, rather than
resolving against an enumeration inferred from the caller's frame,
writing a different value or raising another exception
(:issue:`2594`).
* An option given to :meth:`make` as raw octets, of a kind no member of
:class:`~pcapkit.const.tcp.option.Option` holds, such as 255, which
parses as an unassigned option, builds as its ``(code, arguments)``
pair does, rather than raising
:exc:`~pcapkit.utilities.exceptions.EnumValueError`; and empty raw
octets, which hold no kind, raise
:exc:`~pcapkit.utilities.exceptions.ProtocolError` rather than an
:exc:`IndexError` (:issue:`2708`).
* A header :meth:`make` builds, such as the one
:meth:`~pcapkit.protocols.protocol.Protocol.from_data` rebuilds, reads
the octets after its End of Option List as ``padding``, as a parse of
its octets does, rather than as empty (:issue:`2769`).
.. versionremoved:: 1.6.0
The ``algorithm_default``, ``algorithm_namespace`` and
``algorithm_reversed`` arguments of the alternate checksum request
option maker, and the ``subtype_default``, ``subtype_namespace`` and
``subtype_reversed`` arguments of the Multipath TCP option maker;
resolve a name with
:meth:`~pcapkit.protocols.protocol.Protocol.resolve_index` first
(:issue:`2594`).
"""
##########################################################################
# Defaults.
##########################################################################
#: Protocol index mapping for decoding next layer, see
#: :meth:`self._decode_next_layer
#: <pcapkit.protocols.transport.transport.Transport._decode_next_layer>` &
#: :meth:`self._import_next_layer
#: <pcapkit.protocols.protocol.Protocol._import_next_layer>`.
__proto__ = collections.defaultdict(
lambda: ModuleDescriptor('pcapkit.protocols.misc.raw.protocol', 'Raw'),
{
# Ports are IANA service-name registry assignments, quoting that
# registry's own service name and description:
#
# 20 ftp-data File Transfer [Default Data]
# 21 ftp File Transfer Protocol [Control]
# 80 http World Wide Web HTTP
# 443 https http protocol over TLS/SSL
# 8080 http-alt HTTP Alternate (see port 80)
#
# 8443 is deliberately absent: IANA registers it as ``pcsync-https``,
# not as an HTTP or TLS alternate.
#
# The HTTP ports bind the version-dispatching HTTP base, since
# HTTP/1 and cleartext HTTP/2 share them (RFC 9113): the payload,
# not the port, decides the version, through
# ``HTTP._dispatch_version`` (#2524). UDP's table does the same.
20: ModuleDescriptor('pcapkit.protocols.application.ftp_data.protocol', 'FTP_DATA'),
21: ModuleDescriptor('pcapkit.protocols.application.ftp.protocol', 'FTP'),
80: ModuleDescriptor('pcapkit.protocols.application.http.protocol', 'HTTP'),
443: ModuleDescriptor('pcapkit.protocols.application.tls.protocol', 'TLS'),
8080: ModuleDescriptor('pcapkit.protocols.application.http.protocol', 'HTTP'),
},
)
#: What :attr:`__proto__` is keyed by -- the enum a key must be a member of,
#: and whether a bare :class:`int` is valid too. Read by
#: :func:`~pcapkit.foundation.registry.protocols.protocol.register_protocol_code`
#: to check a ``code=`` key as
#: :func:`~pcapkit.foundation.registry.protocols.transport.register_tcp` checks
#: its own; a bare :class:`int` is a port number, which an
#: :class:`~pcapkit.const.reg.apptype.apptype.AppType` member carries
#: (:issue:`2268`).
__proto_key__: 'ClassVar[tuple[Type[EnumRegistry], bool]]' = (Enum_AppType, True)
#: Option code to method mapping, see :meth:`_read_tcp_options` and
#: :meth:`_make_tcp_options`. A :class:`str` value names the methods
#: ``_read_mode_${name}`` and ``_make_mode_${name}``, falling back to
#: :meth:`_read_mode_donone` and :meth:`_make_mode_donone` when the class
#: has no such method; any other value is a ``(parser, constructor)`` pair
#: of callables.
__option__ = collections.defaultdict(
lambda: 'donone',
{
Enum_Option.End_of_Option_List: 'eool', # [RFC 793] End of Option List
Enum_Option.No_Operation: 'nop', # [RFC 793] No-Operation
Enum_Option.Maximum_Segment_Size: 'mss', # [RFC 793] Maximum Segment Size
Enum_Option.Window_Scale: 'ws', # [RFC 7323] Window Scale
Enum_Option.SACK_Permitted: 'sackpmt', # [RFC 2018] SACK Permitted
Enum_Option.SACK: 'sack', # [RFC 2018] SACK
Enum_Option.Echo: 'echo', # [RFC 1072] Echo
Enum_Option.Echo_Reply: 'echore', # [RFC 1072] Echo Reply
Enum_Option.Timestamps: 'ts', # [RFC 7323] Timestamps
Enum_Option.Partial_Order_Connection_Permitted: 'poc', # [RFC 1693] POC Permitted
Enum_Option.Partial_Order_Service_Profile: 'pocsp', # [RFC 1693] POC-Serv Profile
Enum_Option.CC: 'cc', # [RFC 1644] Connection Count
Enum_Option.CC_NEW: 'ccnew', # [RFC 1644] CC.NEW
Enum_Option.CC_ECHO: 'ccecho', # [RFC 1644] CC.ECHO
Enum_Option.TCP_Alternate_Checksum_Request: 'chkreq', # [RFC 1146] Alt-Chksum Request
Enum_Option.TCP_Alternate_Checksum_Data: 'chksum', # [RFC 1146] Alt-Chksum Data
Enum_Option.MD5_Signature_Option: 'sig', # [RFC 2385] MD5 Signature Option
Enum_Option.Quick_Start_Response: 'qs', # [RFC 4782] Quick-Start Response
Enum_Option.User_Timeout_Option: 'timeout', # [RFC 5482] User Timeout Option
Enum_Option.TCP_Authentication_Option: 'ao', # [RFC 5925] TCP Authentication Option
Enum_Option.Multipath_TCP: 'mp', # [RFC 8684] Multipath TCP
Enum_Option.TCP_Fast_Open_Cookie: 'fastopen', # [RFC 7413] Fast Open
},
) # type: DefaultDict[int, str | tuple[OptionParser, OptionConstructor]]
#: Option code to method mapping, see :meth:`_read_mode_mp` and
#: :meth:`_make_mode_mp`. A :class:`str` value names the methods
#: ``_read_mptcp_${name}`` and ``_make_mptcp_${name}``, falling back to
#: :meth:`_read_mptcp_unknown` and :meth:`_make_mptcp_unknown` when the
#: class has no such method; any other value is a ``(parser, constructor)``
#: pair of callables.
__mp_option__ = collections.defaultdict(
lambda: 'unknown',
{
Enum_MPTCPOption.MP_CAPABLE: 'capable',
Enum_MPTCPOption.MP_JOIN: 'join',
Enum_MPTCPOption.DSS: 'dss',
Enum_MPTCPOption.ADD_ADDR: 'addaddr',
Enum_MPTCPOption.REMOVE_ADDR: 'remove',
Enum_MPTCPOption.MP_PRIO: 'prio',
Enum_MPTCPOption.MP_FAIL: 'fail',
Enum_MPTCPOption.MP_FASTCLOSE: 'fastclose',
},
) # type: DefaultDict[int, str | tuple[MPOptionParser, MPOptionConstructor]]
#: Whether :meth:`make` left the checksum for the enclosing IP layer to
#: compute, see :mod:`pcapkit.protocols.misc.checksum.ip_pseudo.protocol`.
_checksum_pending = False
##########################################################################
# Properties.
##########################################################################
@property
def name(self) -> 'Literal["Transmission Control Protocol"]':
"""Name of current protocol."""
return 'Transmission Control Protocol'
@property
def length(self) -> 'int':
"""Header length of current protocol."""
return self._info.hdr_len
@property
def src(self) -> 'Enum_AppType':
"""Source port."""
return self._info.srcport
@property
def dst(self) -> 'Enum_AppType':
"""Destination port."""
return self._info.dstport
@property
def connection(self) -> 'Enum_Flags':
"""Connection flags."""
return self._flags
##########################################################################
# Methods.
##########################################################################
[docs]
def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_TCP': # pylint: disable=unused-argument
"""Read Transmission Control Protocol (TCP).
Structure of TCP header [:rfc:`793`]::
0 1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Source Port | Destination Port |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Sequence Number |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Acknowledgement Number |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Data | |U|A|P|R|S|F| |
| Offset| Reserved |R|C|S|S|Y|I| Window |
| | |G|K|H|T|N|N| |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Checksum | Urgent Pointer |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Options | Padding |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| data |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Args:
length: Length of packet data.
**kwargs: Arbitrary keyword arguments.
Returns:
Parsed packet data.
Raises:
ProtocolError: If the Data Offset declares more header than the
data holds.
"""
if length is None:
length = len(self)
schema = self.__header__
tcp = Data_TCP(
srcport=schema.srcport,
dstport=schema.dstport,
seq=schema.seq,
ack=schema.ack,
hdr_len=schema.offset['offset'] * 4,
flags=Data_Flags(
ns=bool(schema.offset['ns']),
cwr=bool(schema.flags['cwr']),
ece=bool(schema.flags['ece']),
urg=bool(schema.flags['urg']),
ack=bool(schema.flags['ack']),
psh=bool(schema.flags['psh']),
rst=bool(schema.flags['rst']),
syn=bool(schema.flags['syn']),
fin=bool(schema.flags['fin']),
reserved=schema.offset['reserved'],
),
window_size=schema.window,
checksum=schema.checksum,
urgent_pointer=schema.urgent,
)
# connection control flags
#
# NOTE: ``Enum_Flags(0)``, not ``cast('Enum_Flags', 0)``: a flagless
# segment would keep a plain ``int``, and ``in`` on it raises TypeError.
_flag = Enum_Flags(0)
for key, val in schema.flags.items():
if val == 1:
_flag |= Enum_Flags.get(key.upper())
self._flags = _flag
tcp.__update__({
'connection': self._flags,
})
# NOTE: Rejected rather than read as zeros, which ``from_data`` would
# rebuild, so the parent keeps the octets as captured (#1404).
if len(self) < tcp.hdr_len:
raise ProtocolError(f'{self.alias}: header length {tcp.hdr_len} runs past the end of '
f'the data ({len(self)} octets)')
_optl = tcp.hdr_len - TCP_MIN_HEADER_LEN
if _optl:
# NOTE: Post-EOOL octets are kept so ``make`` rebuilds the parsed Data Offset;
# sliced as IPv4 does, since a built schema keeps them in ``options`` (#2769).
options = self._read_tcp_options(_optl)
_used = sum(opt.length for _, opt in options.items(multi=True))
tcp.__update__({
'options': options,
'padding': self._data[TCP_MIN_HEADER_LEN + _used:tcp.hdr_len],
})
return self._decode_next_layer(tcp, (tcp.srcport.port, tcp.dstport.port), length - tcp.hdr_len)
[docs]
def make(self,
srcport: 'Enum_AppType | int' = 0,
dstport: 'Enum_AppType | int' = 0,
seq_no: 'int' = 0,
ack_no: 'int' = 0,
ns: 'bool' = False,
cwr: 'bool' = False,
ece: 'bool' = False,
urg: 'bool' = False,
ack: 'bool' = False,
psh: 'bool' = False,
rst: 'bool' = False,
syn: 'bool' = False,
fin: 'bool' = False,
window: 'int' = 65535, # reasonable default value
checksum: 'Optional[bytes]' = None,
urgent: 'int' = 0,
options: 'Optional[list[Schema_Option | tuple[Enum_Option | int, dict[str, Any]] | bytes] | Option]' = None, # pylint: disable=line-too-long
payload: 'bytes | ProtocolBase | Schema' = b'',
padding: 'bytes' = b'',
reserved: 'int' = 0,
pseudo: 'Optional[IP_Pseudo]' = None,
**kwargs: 'Any') -> 'Schema_TCP':
"""Make (construct) packet data.
Args:
srcport: Source port.
dstport: Destination port.
seq_no: Sequence number.
ack_no: Acknowledgement number.
ns: ECN-nonce concealment protection.
cwr: Congestion window reduced.
ece: ECN-Echo.
urg: Urgent.
ack: Acknowledgement.
psh: Push function.
rst: Reset connection.
syn: Synchronize sequence numbers.
fin: Last packet from sender.
window: Window size.
checksum: Checksum; if :data:`None` and
:data:`~pcapkit.utilities.config.CHECKSUM_COMPUTE` is on,
computed over ``pseudo`` (:rfc:`793#section-3.1`), or left to
an IP layer this segment is the payload of; zero otherwise.
urgent: Urgent pointer.
options: TCP options.
payload: Payload of the packet.
padding: Octets after the End of Option List option; dropped
unless ``options`` end with one.
reserved: Reserved bits 4-6 of octet 12.
pseudo: Pseudo-header of the enclosing IP layer; its length is
replaced by this segment's own.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed packet data.
Raises:
ProtocolError: If ``srcport`` or ``dstport`` is a :obj:`bool`
(:issue:`2287`), a numeric field does not fit its width, or
``checksum`` is not two octets.
.. versionchanged:: 1.5.0
An out-of-range ``srcport`` or ``dstport`` raises
:exc:`ValueError` rather than being accepted (:issue:`758`).
"""
# NOTE: Flags come first: ``_make_mptcp_join`` picks the MP_JOIN layout from
# ``self._flags``, which a fresh instance lacks and a parsed one holds stale.
flags = {
'cwr': int(cwr),
'ece': int(ece),
'urg': int(urg),
'ack': int(ack),
'psh': int(psh),
'rst': int(rst),
'syn': int(syn),
'fin': int(fin),
} # type: Schema_Flags
# NOTE: ``Enum_Flags(0)``, not ``cast('Enum_Flags', 0)``: with no flag
# set a plain ``int`` stays, and ``in`` on it raises TypeError.
_flag = Enum_Flags(0)
for key, val in flags.items():
if val == 1:
_flag |= Enum_Flags.get(key.upper())
self._flags = _flag
if options is not None:
options_value, total_length = self._make_tcp_options(options, padding)
else:
options_value, total_length = [], 0
offset = math.ceil((TCP_MIN_HEADER_LEN + total_length) / 4)
schema = Schema_TCP(
srcport=self._make_port(srcport, Enum_TransportProtocol.tcp),
dstport=self._make_port(dstport, Enum_TransportProtocol.tcp),
seq=self._make_uint(seq_no, bits=32, name='sequence number'),
ack=self._make_uint(ack_no, bits=32, name='acknowledgement number'),
offset={
'offset': offset,
'reserved': self._make_uint(reserved, bits=3, name='reserved'),
'ns': int(ns),
},
flags=flags,
window=self._make_uint(window, bits=16, name='window'),
checksum=b'\x00\x00' if checksum is None else self._make_octets(checksum, size=2, name='checksum'),
urgent=self._make_uint(urgent, bits=16, name='urgent pointer'),
options=options_value,
payload=payload,
)
self._checksum_pending = False
if checksum is None and config.CHECKSUM_COMPUTE:
if pseudo is None:
self._checksum_pending = True
else:
from pcapkit.corekit.checksum.internet import compute_checksum # isort: skip # pylint: disable=import-outside-toplevel
pseudo_header = _PSEUDO_HEADER.klass
octets = schema.pack()
schema.checksum = compute_checksum(octets, prefix=pseudo_header(pseudo, len(octets)))
return schema
[docs]
@classmethod
def register_option(cls, code: 'Enum_Option', meth: 'str | tuple[OptionParser, OptionConstructor]') -> 'None':
"""Register an option parser.
Args:
code: TCP option code.
meth: Method name, or a ``(parser, constructor)`` pair of
callables, to parse and construct the option; see
:attr:`__option__`.
Warns:
RegistryWarning: If a different entry is already registered for
``code``; it is overwritten. Handing back the entry already
stored, or one equal to it, is a silent no-op. Re-registering
the entry ``code`` shipped with is accepted like any other,
so an override can be undone (:issue:`1362`, :issue:`1363`).
"""
incumbent = cls.__option__.get(code)
if incumbent is not None and incumbent == meth:
return
if incumbent is not None:
warn(f'option {code} already registered, overwriting', RegistryWarning)
cls.__option__[code] = meth
[docs]
@classmethod
def register_mp_option(cls, code: 'Enum_MPTCPOption', meth: 'str | tuple[MPOptionParser, MPOptionConstructor]') -> 'None':
"""Register an MPTCP option parser.
Args:
code: MPTCP option code.
meth: Method name, or a ``(parser, constructor)`` pair of
callables, to parse and construct the option; see
:attr:`__mp_option__`.
Warns:
RegistryWarning: If a different entry is already registered for
``code``; it is overwritten. Handing back the entry already
stored, or one equal to it, is a silent no-op. Re-registering
the entry ``code`` shipped with is accepted like any other,
so an override can be undone (:issue:`1362`, :issue:`1363`).
"""
incumbent = cls.__mp_option__.get(code)
if incumbent is not None and incumbent == meth:
return
if incumbent is not None:
warn(f'option {code} already registered, overwriting', RegistryWarning)
cls.__mp_option__[code] = meth
##########################################################################
# Data models.
##########################################################################
def __length_hint__(self) -> 'Literal[20]':
"""Return an estimated length for the object."""
return TCP_MIN_HEADER_LEN # type: ignore[return-value]
[docs]
@classmethod
def __index__(cls) -> 'TransType': # pylint: disable=invalid-index-returned
"""Numeral registry index of the protocol.
Returns:
Numeral registry index of the protocol in `IANA`_.
.. _IANA: https://www.iana.org/assignments/protocol-numbers/protocol-numbers.xhtml
"""
return TransType.TCP # type: ignore[return-value]
##########################################################################
# Utilities.
##########################################################################
[docs]
@classmethod
def _make_data(cls, data: 'Data_TCP') -> 'dict[str, Any]': # type: ignore[override]
"""Create key-value pairs from ``data`` for protocol construction.
Args:
data: protocol data
Returns:
Key-value pairs for protocol construction.
"""
return {
'srcport': data.srcport,
'dstport': data.dstport,
'seq_no': data.seq,
'ack_no': data.ack,
'ns': data.flags.ns,
'cwr': data.flags.cwr,
'ece': data.flags.ece,
'urg': data.flags.urg,
'ack': data.flags.ack,
'psh': data.flags.psh,
'rst': data.flags.rst,
'syn': data.flags.syn,
'fin': data.flags.fin,
'window': data.window_size,
'checksum': data.checksum,
'urgent': data.urgent_pointer,
'options': getattr(data, 'options', None),
'padding': getattr(data, 'padding', b''),
'reserved': data.flags.reserved,
'payload': cls._make_payload(data),
}
[docs]
def _read_tcp_options(self, size: 'int') -> 'Option':
"""Read TCP option list.
Arguments:
size: length of option list
Returns:
Extracted TCP options.
Raises:
ProtocolError: If the options run past ``size``.
"""
counter = 0 # length of read option list
options = OrderedMultiDict() # type: Option
for schema in self.__header__.options:
kind = schema.kind
name = self._lookup_registry(self.__option__, kind)
if isinstance(name, str):
meth_name = f'_read_mode_{name}'
meth = cast('OptionParser',
getattr(self, meth_name, self._read_mode_donone))
else:
meth = name[0]
data = meth(schema, options=options)
# record option data
options.add(kind, data)
counter += len(schema)
# break when End of Option List (EOOL) triggered
if kind == Enum_Option.End_of_Option_List:
break
# check threshold
if counter > size:
raise ProtocolError('TCP: invalid format')
return options
[docs]
def _make_tcp_options(self,
options: 'list[Schema_Option | tuple[Enum_Option | int, dict[str, Any]] | bytes] | Option',
padding: 'bytes' = b'') -> 'tuple[list[Schema_Option | bytes], int]':
"""Make options for TCP.
Args:
options: TCP options
padding: octets after the End of Option List option
Returns:
Tuple of options and total length of options.
Raises:
ProtocolError: If an option's kind is a name, a :obj:`bool`, or a
code the 8-bit kind octet cannot carry (:issue:`2594`), if an
option given as raw octets is empty (:issue:`2708`), or if
an option maker rejects its option, e.g. a user timeout too
large for 15 bits of minutes, an MPTCP ``subtype`` or an
alternate checksum ``algorithm`` given by name, or an MPTCP
option whose fields, or the segment's flags, it cannot encode.
FieldValueError: If an ADD_ADDR ``addr`` is a :obj:`bool` or not a
valid IP address.
Note:
The options are emitted in the given order, ``NOP`` and ``EOOL``
included, so that a parsed option list re-emits the same octets.
``padding`` follows them as given if they end with ``EOOL``, and is
dropped otherwise. Alignment is added only once, after that, when
the list does not end on a 32-bit boundary: zero octets if the list
already ends with ``EOOL``, otherwise ``NOP`` options followed by an
``EOOL``.
"""
total_length = 0
options_list = [] # type: list[Schema_Option | bytes]
code = None # type: Optional[Enum_Option]
if isinstance(options, list):
for schema in options:
if isinstance(schema, bytes):
if not schema:
raise ProtocolError(f'{self.alias}: invalid option: {schema!r} holds no kind octet')
# NOTE: Resolved like a ``(code, arguments)`` pair's kind, so a kind
# no ``Option`` member holds, such as 255, stays its code (#2708).
code = self._make_index(schema[0], Enum_Option, bits=8, name='option kind')
data = schema # type: Schema_Option | bytes
data_len = len(data)
elif isinstance(schema, Schema):
code = schema.kind
data = schema
data_len = len(schema.pack())
else:
kind, args = cast('tuple[Enum_Option | int, dict[str, Any]]', schema)
code = self._make_index(kind, Enum_Option, bits=8, name='option kind')
name = self._lookup_registry(self.__option__, code)
if isinstance(name, str):
meth_name = f'_make_mode_{name}'
meth = cast('OptionConstructor',
getattr(self, meth_name, self._make_mode_donone))
else:
meth = name[1]
data = meth(code, **args)
data_len = len(data.pack())
options_list.append(data)
total_length += data_len
else:
for key, option in options.items(multi=True):
code = self._make_index(key, Enum_Option, bits=8, name='option kind')
name = self._lookup_registry(self.__option__, code)
if isinstance(name, str):
meth_name = f'_make_mode_{name}'
meth = cast('OptionConstructor',
getattr(self, meth_name, self._make_mode_donone))
else:
meth = name[1]
data = meth(code, option)
data_len = len(data.pack())
options_list.append(data)
total_length += data_len
# NOTE: ``padding`` belongs after EOOL; with that gone, e.g. options edited after
# parsing, it is dropped and the Data Offset counts the options alone.
if padding and code == Enum_Option.End_of_Option_List:
options_list.append(padding)
total_length += len(padding)
# force alignment to 32-bit boundary
pad_len = pad_length(total_length, 4)
if pad_len:
total_length += pad_len
if code == Enum_Option.End_of_Option_List:
options_list.append(bytes(pad_len))
else:
pad_opt = self._make_mode_nop(Enum_Option.No_Operation) # type: ignore[arg-type]
for _ in range(pad_len - 1):
options_list.append(pad_opt)
options_list.append(self._make_mode_eool(Enum_Option.End_of_Option_List)) # type: ignore[arg-type]
return options_list, total_length
[docs]
def _read_mode_donone(self, schema: 'Schema_UnassignedOption', *, options: 'Option') -> 'Data_UnassignedOption': # pylint: disable=unused-argument
"""Read TCP unassigned option.
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
"""
option = Data_UnassignedOption(
kind=schema.kind,
length=schema.length,
data=schema.data,
)
return option
[docs]
def _read_mode_eool(self, schema: 'Schema_EndOfOptionList', *, options: 'Option') -> 'Data_EndOfOptionList': # pylint: disable=unused-argument
"""Read TCP End of Option List option.
Structure of TCP end of option list option [:rfc:`793`]:
.. code-block:: text
+--------+
|00000000|
+--------+
Kind=0
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
"""
return Data_EndOfOptionList(
kind=schema.kind,
length=1,
)
[docs]
def _read_mode_nop(self, schema: 'Schema_NoOperation', *, options: 'Option') -> 'Data_NoOperation': # pylint: disable=unused-argument
"""Read TCP No Operation option.
Structure of TCP no-operation option [:rfc:`793`]:
.. code-block:: text
+--------+
|00000001|
+--------+
Kind=1
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
"""
return Data_NoOperation(
kind=schema.kind,
length=1,
)
[docs]
def _read_mode_mss(self, schema: 'Schema_MaximumSegmentSize', *, options: 'Option') -> 'Data_MaximumSegmentSize': # pylint: disable=unused-argument
"""Read TCP max segment size option.
Structure of TCP maximum segment size option [:rfc:`793`]:
.. code-block:: text
+--------+--------+---------+--------+
|00000010|00000100| max seg size |
+--------+--------+---------+--------+
Kind=2 Length=4
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
Raises:
ProtocolError: If length is **NOT** ``4``.
"""
if schema.length != 4:
raise ProtocolError(f'{self.alias}: [OptNo {schema.kind}] invalid format')
data = Data_MaximumSegmentSize(
kind=schema.kind,
length=schema.length,
mss=schema.mss,
)
return data
[docs]
def _read_mode_ws(self, schema: 'Schema_WindowScale', *, options: 'Option') -> 'Data_WindowScale': # pylint: disable=unused-argument
"""Read TCP window scale option.
Structure of TCP window scale option [:rfc:`7323`]:
.. code-block:: text
+---------+---------+---------+
| Kind=3 |Length=3 |shift.cnt|
+---------+---------+---------+
1 1 1
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
Raises:
ProtocolError: If length is **NOT** ``3``.
"""
if schema.length != 3:
raise ProtocolError(f'{self.alias}: [OptNo {schema.kind}] invalid format')
data = Data_WindowScale(
kind=schema.kind,
length=schema.length,
shift=schema.shift,
)
return data
[docs]
def _read_mode_sackpmt(self, schema: 'Schema_SACKPermitted', *, options: 'Option') -> 'Data_SACKPermitted': # pylint: disable=unused-argument
"""Read TCP SACK permitted option.
Structure of TCP SACK permitted option [:rfc:`2018`]:
.. code-block:: text
+---------+---------+
| Kind=4 | Length=2|
+---------+---------+
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
Raises:
ProtocolError: If length is **NOT** ``2``.
"""
if schema.length != 2:
raise ProtocolError(f'{self.alias}: [OptNo {schema.kind}] invalid format')
return Data_SACKPermitted(
kind=schema.kind,
length=schema.length,
)
[docs]
def _read_mode_sack(self, schema: 'Schema_SACK', *, options: 'Option') -> 'Data_SACK': # pylint: disable=unused-argument
"""Read TCP SACK option.
Structure of TCP SACK option [:rfc:`2018`]:
.. code-block:: text
+--------+--------+
| Kind=5 | Length |
+--------+--------+--------+--------+
| Left Edge of 1st Block |
+--------+--------+--------+--------+
| Right Edge of 1st Block |
+--------+--------+--------+--------+
| |
/ . . . /
| |
+--------+--------+--------+--------+
| Left Edge of nth Block |
+--------+--------+--------+--------+
| Right Edge of nth Block |
+--------+--------+--------+--------+
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
Raises:
ProtocolError: If length is **NOT** a multiple of ``8`` plus ``2``.
"""
if (schema.length - 2) % 8 != 0:
raise ProtocolError(f'{self.alias}: [OptNo {schema.kind}] invalid format')
data = Data_SACK(
kind=schema.kind,
length=schema.length,
sack=tuple(
Data_SACKBlock(
left=block.left,
right=block.right,
) for block in schema.sack
),
)
return data
[docs]
def _read_mode_echo(self, schema: 'Schema_Echo', *, options: 'Option') -> 'Data_Echo': # pylint: disable=unused-argument
"""Read TCP echo option.
Structure of TCP echo option [:rfc:`1072`]:
.. code-block:: text
+--------+--------+--------+--------+--------+--------+
| Kind=6 | Length | 4 bytes of info to be echoed |
+--------+--------+--------+--------+--------+--------+
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
Raises:
ProtocolError: If length is **NOT** ``6``.
"""
if schema.length != 6:
raise ProtocolError(f'{self.alias}: [OptNo {schema.kind}] invalid format')
data = Data_Echo(
kind=schema.kind,
length=schema.length,
data=schema.data,
)
return data
[docs]
def _read_mode_echore(self, schema: 'Schema_EchoReply', *, options: 'Option') -> 'Data_EchoReply': # pylint: disable=unused-argument
"""Read TCP echo reply option.
Structure of TCP echo reply option [:rfc:`1072`]:
.. code-block:: text
+--------+--------+--------+--------+--------+--------+
| Kind=7 | Length | 4 bytes of echoed info |
+--------+--------+--------+--------+--------+--------+
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
Raises:
ProtocolError: If length is **NOT** ``6``.
"""
if schema.length != 6:
raise ProtocolError(f'{self.alias}: [OptNo {schema.kind}] invalid format')
data = Data_EchoReply(
kind=schema.kind,
length=schema.length,
data=schema.data,
)
return data
[docs]
def _read_mode_ts(self, schema: 'Schema_Timestamps', *, options: 'Option') -> 'Data_Timestamps': # pylint: disable=unused-argument
"""Read TCP timestamps option.
Structure of TCP timestamp option [:rfc:`7323`]:
.. code-block:: text
+-------+-------+---------------------+---------------------+
|Kind=8 | 10 | TS Value (TSval) |TS Echo Reply (TSecr)|
+-------+-------+---------------------+---------------------+
1 1 4 4
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
Raises:
ProtocolError: If length is **NOT** ``10``.
"""
if schema.length != 10:
raise ProtocolError(f'{self.alias}: [OptNo {schema.kind}] invalid format')
data = Data_Timestamps(
kind=schema.kind,
length=schema.length,
timestamp=schema.value,
echo=schema.reply,
)
return data
[docs]
def _read_mode_poc(self, schema: 'Schema_PartialOrderConnectionPermitted', *, options: 'Option') -> 'Data_PartialOrderConnectionPermitted': # pylint: disable=unused-argument
"""Read TCP partial order connection permitted option.
Structure of TCP ``POC-Permitted`` option [:rfc:`1693`][:rfc:`6247`]:
.. code-block:: text
+-----------+-------------+
| Kind=9 | Length=2 |
+-----------+-------------+
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
Raises:
ProtocolError: If length is **NOT** ``2``.
"""
if schema.length != 2:
raise ProtocolError(f'{self.alias}: [OptNo {schema.kind}] invalid format')
return Data_PartialOrderConnectionPermitted(
kind=schema.kind,
length=schema.length,
)
[docs]
def _read_mode_pocsp(self, schema: 'Schema_PartialOrderServiceProfile', *, options: 'Option') -> 'Data_PartialOrderServiceProfile': # pylint: disable=unused-argument
"""Read TCP partial order connection service profile option.
Structure of TCP ``POC-SP`` option [:rfc:`1693`][:rfc:`6247`]:
.. code-block:: text
1 bit 1 bit 6 bits
+----------+----------+------------+----------+--------+
| Kind=10 | Length=3 | Start_flag | End_flag | Filler |
+----------+----------+------------+----------+--------+
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
Raises:
ProtocolError: If length is **NOT** ``3``.
"""
if schema.length != 3:
raise ProtocolError(f'{self.alias}: [OptNo {schema.kind}] invalid format')
data = Data_PartialOrderServiceProfile(
kind=schema.kind,
length=schema.length,
start=bool(schema.profile['start']),
end=bool(schema.profile['end']),
reserved=schema.profile['reserved'],
)
return data
[docs]
def _read_mode_cc(self, schema: 'Schema_CC', *, options: 'Option') -> 'Data_CC': # pylint: disable=unused-argument
"""Read TCP connection count option.
Structure of TCP ``CC`` option [:rfc:`1644`]:
.. code-block:: text
+--------+--------+--------+--------+--------+--------+
|00001011|00000110| Connection Count: SEG.CC |
+--------+--------+--------+--------+--------+--------+
Kind=11 Length=6
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
Raises:
ProtocolError: If length is **NOT** ``6``.
"""
if schema.length != 6:
raise ProtocolError(f'{self.alias}: [OptNo {schema.kind}] invalid format')
data = Data_CC(
kind=schema.kind,
length=schema.length,
cc=schema.count,
)
return data
[docs]
def _read_mode_ccnew(self, schema: 'Schema_CCNew', *, options: 'Option') -> 'Data_CCNew': # pylint: disable=unused-argument
"""Read TCP connection count (new) option.
Structure of TCP ``CC.NEW`` option [:rfc:`1644`]:
.. code-block:: text
+--------+--------+--------+--------+--------+--------+
|00001100|00000110| Connection Count: SEG.CC |
+--------+--------+--------+--------+--------+--------+
Kind=12 Length=6
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
Raises:
ProtocolError: If length is **NOT** ``6``.
"""
if schema.length != 6:
raise ProtocolError(f'{self.alias}: [OptNo {schema.kind}] invalid format')
data = Data_CCNew(
kind=schema.kind,
length=schema.length,
cc=schema.count,
)
return data
[docs]
def _read_mode_ccecho(self, schema: 'Schema_CCEcho', *, options: 'Option') -> 'Data_CCEcho': # pylint: disable=unused-argument
"""Read TCP connection count (echo) option.
Structure of TCP ``CC.ECHO`` option [:rfc:`1644`]:
.. code-block:: text
+--------+--------+--------+--------+--------+--------+
|00001101|00000110| Connection Count: SEG.CC |
+--------+--------+--------+--------+--------+--------+
Kind=13 Length=6
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
Raises:
ProtocolError: If length is **NOT** ``6``.
"""
if schema.length != 6:
raise ProtocolError(f'{self.alias}: [OptNo {schema.kind}] invalid format')
data = Data_CCEcho(
kind=schema.kind,
length=schema.length,
cc=schema.count,
)
return data
[docs]
def _read_mode_chkreq(self, schema: 'Schema_AlternateChecksumRequest', *, options: 'Option') -> 'Data_AlternateChecksumRequest': # pylint: disable=unused-argument
"""Read TCP Alternate Checksum Request option.
Structure of TCP ``CHKSUM-REQ`` [:rfc:`1146`][:rfc:`6247`]:
.. code-block:: text
+----------+----------+----------+
| Kind=14 | Length=3 | chksum |
+----------+----------+----------+
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
Raises:
ProtocolError: If length is **NOT** ``3``.
"""
if schema.length != 3:
raise ProtocolError(f'{self.alias}: [OptNo {schema.kind}] invalid format')
data = Data_AlternateChecksumRequest(
kind=schema.kind,
length=schema.length,
chksum=schema.algorithm,
)
return data
[docs]
def _read_mode_chksum(self, schema: 'Schema_AlternateChecksumData', *, options: 'Option') -> 'Data_AlternateChecksumData': # pylint: disable=unused-argument
"""Read Alternate Checksum Data option.
Structure of TCP ``CHKSUM`` [:rfc:`1146`][:rfc:`6247`]:
.. code-block:: text
+---------+---------+---------+ +---------+
| Kind=15 |Length=N | data | ... | data |
+---------+---------+---------+ +---------+
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
"""
data = Data_AlternateChecksumData(
kind=schema.kind,
length=schema.length,
data=schema.data,
)
return data
[docs]
def _read_mode_sig(self, schema: 'Schema_MD5Signature', *, options: 'Option') -> 'Data_MD5Signature': # pylint: disable=unused-argument
"""Read MD5 Signature option.
Structure of TCP ``SIG`` option [:rfc:`2385`]:
.. code-block:: text
+---------+---------+-------------------+
| Kind=19 |Length=18| MD5 digest... |
+---------+---------+-------------------+
| |
+---------------------------------------+
| |
+---------------------------------------+
| |
+-------------------+-------------------+
| |
+-------------------+
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
Raises:
ProtocolError: If length is **NOT** ``18``.
"""
if schema.length != 18:
raise ProtocolError(f'{self.alias}: [OptNo {schema.kind}] invalid format')
data = Data_MD5Signature(
kind=schema.kind,
length=schema.length,
digest=schema.digest,
)
return data
[docs]
def _read_mode_qs(self, schema: 'Schema_QuickStartResponse', *, options: 'Option') -> 'Data_QuickStartResponse': # pylint: disable=unused-argument
"""Read Quick-Start Response option.
Structure of TCP ``QSopt`` [:rfc:`4782`]:
.. code-block:: text
0 1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Kind | Length=8 | Resv. | Rate | TTL Diff |
| | | |Request| |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| QS Nonce | R |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
Raises:
ProtocolError: If length is **NOT** ``8``.
"""
if schema.length != 8:
raise ProtocolError(f'{self.alias}: [OptNo {schema.kind}] invalid format')
data = Data_QuickStartResponse(
kind=schema.kind,
length=schema.length,
req_rate=self._read_qs_rate(schema.flags['rate']),
ttl_diff=schema.diff,
nonce=schema.nonce['nonce'],
reserved=schema.flags['reserved'],
nonce_reserved=schema.nonce['reserved'],
)
return data
[docs]
def _read_mode_timeout(self, schema: 'Schema_UserTimeout', *, options: 'Option') -> 'Data_UserTimeout': # pylint: disable=unused-argument
"""Read User Timeout option.
Structure of TCP ``TIMEOUT`` [:rfc:`5482`]:
.. code-block:: text
0 1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Kind = 28 | Length = 4 |G| User Timeout |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
Raises:
ProtocolError: If length is **NOT** ``4``.
"""
if schema.length != 4:
raise ProtocolError(f'{self.alias}: [OptNo {schema.kind}] invalid format')
if schema.info['granularity'] == 1:
time = datetime.timedelta(minutes=schema.info['timeout'])
else:
time = datetime.timedelta(seconds=schema.info['timeout'])
data = Data_UserTimeout(
kind=schema.kind,
length=schema.length,
timeout=time,
granularity=bool(schema.info['granularity']),
)
return data
[docs]
def _read_mode_ao(self, schema: 'Schema_Authentication', *, options: 'Option') -> 'Data_Authentication': # pylint: disable=unused-argument
"""Read Authentication option.
Structure of TCP ``AOopt`` [:rfc:`5925`]:
.. code-block:: text
+------------+------------+------------+------------+
| Kind=29 | Length | KeyID | RNextKeyID |
+------------+------------+------------+------------+
| MAC ...
+-----------------------------------...
...-----------------+
... MAC (con't) |
...-----------------+
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
Raises:
ProtocolError: If length is **NOT** larger than or equal to ``4``.
"""
if schema.length < 4:
raise ProtocolError(f'{self.alias}: [OptNo {schema.kind}] invalid format')
data = Data_Authentication(
kind=schema.kind,
length=schema.length,
key_id=schema.key_id,
next_key_id=schema.next_key_id,
mac=schema.mac,
)
return data
[docs]
def _read_mode_mp(self, schema: 'Schema_MPTCP', *, options: 'Option') -> 'Data_MPTCP': # pylint: disable=unused-argument
"""Read Multipath TCP option.
Structure of ``MP-TCP`` [:rfc:`8684`]:
.. code-block:: text
1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+---------------+---------------+-------+-----------------------+
| Kind | Length |Subtype| |
+---------------+---------------+-------+ |
| Subtype-specific data |
| (variable length) |
+---------------------------------------------------------------+
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
"""
subtype = schema.subtype
name = self._lookup_registry(self.__mp_option__, subtype)
if isinstance(name, str):
meth_name = f'_read_mptcp_{name}'
meth = cast('MPOptionParser',
getattr(self, meth_name, self._read_mptcp_unknown))
else:
meth = name[0]
data = meth(schema, options=options)
return data
[docs]
def _read_mptcp_unknown(self, schema: 'Schema_MPTCPUnknown', *, options: 'Option') -> 'Data_MPTCPUnknown': # pylint: disable=unused-argument
"""Read unknown MPTCP subtype.
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
"""
data = Data_MPTCPUnknown(
kind=Enum_Option.Multipath_TCP, # type: ignore[arg-type]
length=schema.length,
subtype=schema.subtype,
data=schema.test['data'].to_bytes(1, 'big', signed=False) + schema.data,
)
return data
[docs]
def _read_mptcp_capable(self, schema: 'Schema_MPTCPCapable', *, options: 'Option') -> 'Data_MPTCPCapable': # pylint: disable=unused-argument
"""Read Multipath Capable option.
Structure of ``MP_CAPABLE`` [:rfc:`8684`]:
.. code-block:: text
1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+---------------+---------------+-------+-------+---------------+
| Kind | Length |Subtype|Version|A|B|C|D|E|F|G|H|
+---------------+---------------+-------+-------+---------------+
| Option Sender's Key (64 bits) |
| (if option Length > 4) |
| |
+---------------------------------------------------------------+
| Option Receiver's Key (64 bits) |
| (if option Length > 12) |
| |
+-------------------------------+-------------------------------+
| Data-Level Length (16 bits) | Checksum (16 bits, optional) |
+-------------------------------+-------------------------------+
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
Raises:
ProtocolError: If length is **NOT** ``4``, ``12``, ``20``, ``22``
or ``24``.
"""
# NOTE: :rfc:`8684` section 3.1 defines exactly these five lengths;
# the schema's field conditions follow from them.
if schema.length not in (4, 12, 20, 22, 24):
raise ProtocolError(f'{self.alias}: [OptNo {schema.kind}] invalid format')
data = Data_MPTCPCapable(
kind=Enum_Option.Multipath_TCP, # type: ignore[arg-type]
length=schema.length,
subtype=schema.subtype,
version=schema.test['version'],
flags=Data_MPTCPCapableFlag(
req=bool(schema.flags['req']),
ext=bool(schema.flags['ext']),
deny_join=bool(schema.flags['deny_join']),
reserved=schema.flags['reserved'],
hsa=bool(schema.flags['hsa']),
),
skey=schema.skey if schema.length > 4 else None,
rkey=schema.rkey if schema.length > 12 else None,
dl_len=schema.dl_len if schema.length > 20 else None,
checksum=schema.checksum if schema.length > 22 else None,
)
return data
[docs]
def _read_mptcp_join(self, schema: 'Schema_MPTCPJoin', *, options: 'Option') -> 'Data_MPTCPJoin':
"""Read Join Connection option.
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
Raises:
ProtocolError: If the segment carries neither SYN nor ACK.
"""
if Enum_Flags.SYN in self._flags and Enum_Flags.ACK not in self._flags: # MP_JOIN-SYN
return self._read_join_syn(schema, options=options) # type: ignore[arg-type]
if Enum_Flags.SYN in self._flags and Enum_Flags.ACK in self._flags: # MP_JOIN-SYN/ACK
return self._read_join_synack(schema, options=options) # type: ignore[arg-type]
if Enum_Flags.SYN not in self._flags and Enum_Flags.ACK in self._flags: # MP_JOIN-ACK
return self._read_join_ack(schema, options=options) # type: ignore[arg-type]
raise ProtocolError(f'{self.alias}: [OptNo {schema.kind}] {schema.subtype}: invalid flags combination')
[docs]
def _read_mptcp_dss(self, schema: 'Schema_MPTCPDSS', *, options: 'Option') -> 'Data_MPTCPDSS': # pylint: disable=unused-argument
"""Read Data Sequence Signal (Data ACK and Data Sequence Mapping) option.
Structure of ``DSS`` [:rfc:`8684`, section 3.3, figure 9]:
.. code-block:: text
1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+---------------+---------------+-------+----------------------+
| Kind | Length |Subtype| (reserved) |F|m|M|a|A|
+---------------+---------------+-------+----------------------+
| |
| Data ACK (4 or 8 octets, depending on flags) |
| |
+--------------------------------------------------------------+
| |
| Data sequence number (4 or 8 octets, depending on flags) |
| |
+--------------------------------------------------------------+
| Subflow Sequence Number (4 octets) |
+-------------------------------+------------------------------+
| Data-Level Length (2 octets) | Checksum (2 octets) |
+-------------------------------+------------------------------+
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
Raises:
ProtocolError: If the length is not the one the ``A``, ``a``, ``M`` and
``m`` flags give, or, with ``M`` set, that plus the 2-octet checksum.
Note:
The reserved field spans the 4 low bits of the subtype octet and the
3 high bits of the flags octet; it is kept as one 7-bit value. The
checksum is absent when the option length leaves no room for it,
i.e., when checksums were not negotiated.
.. versionchanged:: 1.6.0
A length the flags do not give is refused, rather than read with the
missing fields as zeros or the extra octets dropped (:issue:`2669`).
"""
flags = schema.flags
size = 4 + ((8 if flags['a'] else 4) if flags['A'] else 0) \
+ ((8 if flags['m'] else 4) + 6 if flags['M'] else 0)
if schema.length != size and not (flags['M'] and schema.length == size + 2):
raise ProtocolError(f'{self.alias}: [OptNo {schema.kind}] invalid format')
data = Data_MPTCPDSS(
kind=Enum_Option.Multipath_TCP, # type: ignore[arg-type]
length=schema.length,
subtype=schema.subtype,
reserved=(schema.test['reserved'] << 3) | schema.flags['reserved'],
data_fin=bool(schema.flags['F']),
ack_wide=bool(schema.flags['a']),
ack=schema.ack,
dsn_wide=bool(schema.flags['m']),
dsn=schema.dsn,
ssn=schema.ssn,
dl_len=schema.dl_len,
checksum=schema.checksum,
)
return data
[docs]
def _read_mptcp_addaddr(self, schema: 'Schema_MPTCPAddAddress', *, options: 'Option') -> 'Data_MPTCPAddAddress': # pylint: disable=unused-argument
"""Read Add Address option.
Structure of ``ADD_ADDR`` [:rfc:`8684`, section 3.4.1, figure 12]:
.. code-block:: text
1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+---------------+---------------+-------+-------+---------------+
| Kind | Length |Subtype|(rsv)|E| Address ID |
+---------------+---------------+-------+-------+---------------+
| Address (IPv4: 4 octets / IPv6: 16 octets) |
+-------------------------------+-------------------------------+
| Port (2 octets, optional) | |
+-------------------------------+ |
| Truncated HMAC (8 octets, if E=0) |
| +-------------------------------+
| |
+-------------------------------+
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
Raises:
ProtocolError: If the length is not one of the :rfc:`8684` lengths.
Note:
The address family, the port and the HMAC all follow from the
option length, never from the ``E`` flag, so an :rfc:`6824`
option, whose ``IPVer`` nibble overlaps the reserved bits and
``E``, parses the same way and round-trips unchanged.
"""
if schema.length not in (8, 10, 16, 18, 20, 22, 28, 30):
raise ProtocolError(f'{self.alias}: [OptNo {schema.kind}] invalid format')
data = Data_MPTCPAddAddress(
kind=Enum_Option.Multipath_TCP, # type: ignore[arg-type]
length=schema.length,
subtype=schema.subtype,
version=schema.address.version,
reserved=schema.test['reserved'],
echo=bool(schema.test['echo']),
addr_id=schema.addr_id,
addr=schema.address,
port=schema.port,
hmac=schema.hmac,
)
return data
[docs]
def _read_mptcp_remove(self, schema: 'Schema_MPTCPRemoveAddress', *, options: 'Option') -> 'Data_MPTCPRemoveAddress': # pylint: disable=unused-argument
"""Read Remove Address option.
Structure of ``REMOVE_ADDR`` [:rfc:`8684`, section 3.4.2, figure 13]:
.. code-block:: text
1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+---------------+---------------+-------+-------+---------------+
| Kind | Length = 3+n |Subtype|(resvd)| Address ID | ...
+---------------+---------------+-------+-------+---------------+
(followed by n-1 Address IDs, if required)
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
Raises:
ProtocolError: If the length is smaller than **3**.
Note:
``Length = 3 + n``, per the figure above: the 3 is ``Kind`` (1) +
``Length`` (1) + subtype-and-reserved (1), and each of the *n* Address
IDs is one further octet.
:attr:`~pcapkit.protocols.transport.tcp.schema.MPTCPRemoveAddress.addr_id`
sizes its list as ``pkt['length'] - 3`` from exactly this, so the
declared length governs the parse as well as the pack.
The guard permits ``3``, i.e. ``n = 0``, which the figure does not
describe -- it shows one Address ID plus "n-1 Address IDs, if
required". It is left permissive: a zero-ID REMOVE_ADDR round-trips
honestly, and rejecting an empty list would be a behaviour change.
"""
if schema.length < 3:
raise ProtocolError(f'{self.alias}: [OptNo {schema.kind}] invalid format')
data = Data_MPTCPRemoveAddress(
kind=Enum_Option.Multipath_TCP, # type: ignore[arg-type]
length=schema.length,
subtype=schema.subtype,
reserved=schema.test['reserved'],
addr_id=tuple(schema.addr_id),
)
return data
[docs]
def _read_mptcp_prio(self, schema: 'Schema_MPTCPPriority', *, options: 'Option') -> 'Data_MPTCPPriority': # pylint: disable=unused-argument
"""Read Change Subflow Priority option.
Structure of ``MP_PRIO`` [:rfc:`6824`, section 3.3.8, figure 11]:
.. code-block:: text
1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+---------------+---------------+-------+-----+-+--------------+
| Kind | Length |Subtype| |B| AddrID (opt) |
+---------------+---------------+-------+-----+-+--------------+
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
Raises:
ProtocolError: If the length is neither **3** nor **4**.
Note:
The figure above is :rfc:`6824`'s, the one with an Address ID, as this
method accepts both forms. :rfc:`8684` section 3.3.8 figure 11 draws
MP_PRIO as **3** octets -- ``Kind`` (1) + ``Length`` (1) +
subtype/reserved/``B`` (1) -- since its section 5 "specifies the
removal of the AddrID field [RFC6824] in the MP_PRIO option", closing
an attack that could force a subflow into backup mode. The 4-octet
form is legacy, and the guard stays permissive so that traffic
carrying it still parses.
On the construction side,
:class:`~pcapkit.protocols.transport.tcp.schema.MPTCPPriority`'s
``addr_id`` is conditional on ``length`` being 4, so ``length`` has
to follow whether the caller supplied an Address ID.
"""
if schema.length not in (3, 4):
raise ProtocolError(f'{self.alias}: [OptNo {schema.kind}] invalid format')
data = Data_MPTCPPriority(
kind=Enum_Option.Multipath_TCP, # type: ignore[arg-type]
length=schema.length,
subtype=schema.subtype,
reserved=schema.test['reserved'],
backup=bool(schema.test['backup']),
addr_id=schema.addr_id,
)
return data
[docs]
def _read_mptcp_fail(self, schema: 'Schema_MPTCPFallback', *, options: 'Option') -> 'Data_MPTCPFallback': # pylint: disable=unused-argument
"""Read Fallback option.
Structure of ``MP_FAIL`` [:rfc:`8684`, section 3.7, figure 16]:
.. code-block:: text
1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+---------------+---------------+-------+----------------------+
| Kind | Length=12 |Subtype| (reserved) |
+---------------+---------------+-------+----------------------+
| |
| Data Sequence Number (8 octets) |
| |
+--------------------------------------------------------------+
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
Raises:
ProtocolError: If the length is **NOT** 12.
"""
if schema.length != 12:
raise ProtocolError(f'{self.alias}: [OptNo {schema.kind}] invalid format')
data = Data_MPTCPFallback(
kind=Enum_Option.Multipath_TCP, # type: ignore[arg-type]
length=schema.length,
subtype=schema.subtype,
reserved=schema.test['reserved'],
dsn=schema.dsn,
)
return data
[docs]
def _read_mptcp_fastclose(self, schema: 'Schema_MPTCPFastclose', options: 'Option') -> 'Data_MPTCPFastclose': # pylint: disable=unused-argument
"""Read Fast Close option.
Structure of ``MP_FASTCLOSE`` [:rfc:`8684`, section 3.5, figure 14]:
.. code-block:: text
1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+---------------+---------------+-------+-----------------------+
| Kind | Length |Subtype| (reserved) |
+---------------+---------------+-------+-----------------------+
| Option Receiver's Key |
| (64 bits) |
| |
+---------------------------------------------------------------+
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
Raises:
ProtocolError: If the length is **NOT** 12.
Note:
The figure above is :rfc:`8684` section 3.5 figure 14, and the option
it draws is **12** octets: ``Kind`` (1) + ``Length`` (1) +
subtype-and-reserved (2, being 4 subtype bits and 12 reserved) + the
option receiver's key (64 bits, 8). The guard,
``_make_mptcp_fastclose`` and the schema must all agree on 12, the
schema reading the 12 reserved bits along with the subtype.
"""
if schema.length != 12:
raise ProtocolError(f'{self.alias}: [OptNo {schema.kind}] invalid format')
data = Data_MPTCPFastclose(
kind=Enum_Option.Multipath_TCP, # type: ignore[arg-type]
length=schema.length,
subtype=schema.subtype,
reserved=schema.test['reserved'],
rkey=schema.key,
)
return data
[docs]
def _read_mode_fastopen(self, schema: 'Schema_FastOpenCookie', *, options: 'Option') -> 'Data_FastOpenCookie': # pylint: disable=unused-argument
"""Read Fast Open option.
Structure of TCP ``FASTOPEN`` [:rfc:`7413`]:
.. code-block:: text
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Kind | Length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| |
~ Cookie ~
| |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
Raises:
ProtocolError: If length is **NOT** valid.
"""
if not (schema.length == 2 or (6 <= schema.length <= 18 and schema.length % 2 == 0)):
raise ProtocolError(f'{self.alias}: [OptNo {schema.kind}] invalid format')
# a length of 2 is a cookie request (RFC 7413 Section 4.1.1),
# which carries no cookie, so the schema field is absent (#2167)
data = Data_FastOpenCookie(
kind=schema.kind,
length=schema.length,
cookie=schema.cookie if schema.cookie is not None else b'',
)
return data
[docs]
def _make_mode_donone(self, code: 'Enum_Option', opt: 'Optional[Data_UnassignedOption]' = None, *,
data: 'bytes' = b'',
**kwargs: 'Any') -> 'Schema_UnassignedOption':
"""Make TCP unassigned option.
Args:
code: option code
opt: option data
data: option payload in :obj:`bytes`
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
"""
if opt is not None:
data = opt.data
return Schema_UnassignedOption(
kind=code,
length=len(data) + 2,
data=data,
)
[docs]
def _make_mode_eool(self, code: 'Enum_Option', opt: 'Optional[Data_EndOfOptionList]' = None, **kwargs: 'Any') -> 'Schema_EndOfOptionList':
"""Make TCP End of Option List option.
Args:
code: option code
opt: option data
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
"""
return Schema_EndOfOptionList(
kind=code,
length=1,
)
[docs]
def _make_mode_nop(self, code: 'Enum_Option', opt: 'Optional[Data_NoOperation]' = None, **kwargs: 'Any') -> 'Schema_NoOperation':
"""Make TCP NoOperation option.
Args:
code: option code
opt: option data
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
"""
return Schema_NoOperation(
kind=code,
length=1,
)
[docs]
def _make_mode_mss(self, code: 'Enum_Option', opt: 'Optional[Data_MaximumSegmentSize]' = None, *,
mss: 'int' = 65535, # reasonable default value
**kwargs: 'Any') -> 'Schema_MaximumSegmentSize':
"""Make TCP maximum segment size option.
Args:
code: option code
opt: option data
mss: maximum segment size
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
Raises:
ProtocolError: If ``mss`` does not fit 16 bits.
"""
if opt is not None:
mss = opt.mss
return Schema_MaximumSegmentSize(
kind=code,
length=4,
mss=self._make_uint(mss, bits=16, name=f'[OptNo {code}] mss'),
)
[docs]
def _make_mode_ws(self, code: 'Enum_Option', opt: 'Optional[Data_WindowScale]' = None, *,
shift: 'int' = 0, # reasonable default value
**kwargs: 'Any') -> 'Schema_WindowScale':
"""Make TCP window scale option.
Args:
code: option code
opt: option data
shift: window scale shift count
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
Raises:
ProtocolError: If ``shift`` does not fit 8 bits.
"""
if opt is not None:
shift = opt.shift
return Schema_WindowScale(
kind=code,
length=3,
shift=self._make_uint(shift, bits=8, name=f'[OptNo {code}] shift'),
)
[docs]
def _make_mode_sackpmt(self, code: 'Enum_Option', opt: 'Optional[Data_SACKPermitted]' = None,
**kwargs: 'Any') -> 'Schema_SACKPermitted':
"""Make TCP SACK permitted option.
Args:
code: option code
opt: option data
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
"""
return Schema_SACKPermitted(
kind=code,
length=2,
)
[docs]
def _make_mode_sack(self, code: 'Enum_Option', opt: 'Optional[Data_SACK]' = None, *,
sack: 'Optional[list[tuple[int, int]]]' = None,
**kwargs: 'Any') -> 'Schema_SACK':
"""Make TCP SACK option.
Args:
code: option code
opt: option data
sack: SACK blocks
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
Raises:
ProtocolError: If an edge of ``sack`` does not fit 32 bits.
"""
if opt is not None:
sack = [(block.left, block.right) for block in opt.sack]
sack_val = [Schema_SACKBlock(
left=self._make_uint(left, bits=32, name=f'[OptNo {code}] left edge'),
right=self._make_uint(right, bits=32, name=f'[OptNo {code}] right edge'),
) for left, right in sack or []]
return Schema_SACK(
kind=code,
length=2 + (len(sack_val) * 8),
sack=sack_val,
)
[docs]
def _make_mode_echo(self, code: 'Enum_Option', opt: 'Optional[Data_Echo]' = None, *,
data: 'bytes' = b'\x00\x00\x00\x00',
**kwargs: 'Any') -> 'Schema_Echo':
"""Make TCP echo option.
Args:
code: option code
opt: option data
data: 4 bytes of info to be echoed
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
Raises:
ProtocolError: If ``data`` is not four octets.
"""
if opt is not None:
data = opt.data
return Schema_Echo(
kind=code,
length=6,
data=self._make_octets(data, size=4, name=f'[OptNo {code}] data'),
)
[docs]
def _make_mode_echore(self, code: 'Enum_Option', opt: 'Optional[Data_EchoReply]' = None, *,
data: 'bytes' = b'\x00\x00\x00\x00',
**kwargs: 'Any') -> 'Schema_EchoReply':
"""Make TCP echo reply option.
Args:
code: option code
opt: option data
data: 4 bytes of echoed info
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
Raises:
ProtocolError: If ``data`` is not four octets.
"""
if opt is not None:
data = opt.data
return Schema_EchoReply(
kind=code,
length=6,
data=self._make_octets(data, size=4, name=f'[OptNo {code}] data'),
)
[docs]
def _make_mode_ts(self, code: 'Enum_Option', opt: 'Optional[Data_Timestamps]' = None, *,
tsval: 'int' = 0,
tsecr: 'int' = 0,
**kwargs: 'Any') -> 'Schema_Timestamps':
"""Make TCP timestamps option.
Args:
code: option code
opt: option data
tsval: timestamp value
tsecr: timestamp echo reply
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
Raises:
ProtocolError: If ``tsval`` or ``tsecr`` does not fit 32 bits.
"""
if opt is not None:
tsval = opt.timestamp
tsecr = opt.echo
return Schema_Timestamps(
kind=code,
length=10,
value=self._make_uint(tsval, bits=32, name=f'[OptNo {code}] tsval'),
reply=self._make_uint(tsecr, bits=32, name=f'[OptNo {code}] tsecr'),
)
[docs]
def _make_mode_poc(self, code: 'Enum_Option', opt: 'Optional[Data_PartialOrderConnectionPermitted]' = None,
**kwargs: 'Any') -> 'Schema_PartialOrderConnectionPermitted':
"""Make TCP partial order connection option.
Args:
code: option code
opt: option data
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
"""
return Schema_PartialOrderConnectionPermitted(
kind=code,
length=2,
)
[docs]
def _make_mode_pocsp(self, code: 'Enum_Option', opt: 'Optional[Data_PartialOrderServiceProfile]' = None, *,
start: 'bool' = False,
end: 'bool' = False,
reserved: 'int' = 0,
**kwargs: 'Any') -> 'Schema_PartialOrderServiceProfile':
"""Make TCP partial order connection service profile option.
Args:
code: option code
opt: option data
start: start partial order connection
end: end partial order connection
reserved: reserved bits (``Filler``, 6 bits)
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
Raises:
ProtocolError: If ``reserved`` does not fit 6 bits.
"""
if opt is not None:
start = opt.start
end = opt.end
reserved = opt.reserved
return Schema_PartialOrderServiceProfile(
kind=code,
length=3,
profile={
'start': start,
'end': end,
'reserved': self._make_uint(reserved, bits=6, name=f'[OptNo {code}] reserved'),
},
)
[docs]
def _make_mode_cc(self, code: 'Enum_Option', opt: 'Optional[Data_CC]' = None, *,
count: 'int' = 0,
**kwargs: 'Any') -> 'Schema_CC':
"""Make TCP connection count option.
Args:
code: option code
opt: option data
count: connection count
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
Raises:
ProtocolError: If ``count`` does not fit 32 bits.
"""
if opt is not None:
count = opt.cc
return Schema_CC(
kind=code,
length=6,
count=self._make_uint(count, bits=32, name=f'[OptNo {code}] count'),
)
[docs]
def _make_mode_ccnew(self, code: 'Enum_Option', opt: 'Optional[Data_CCNew]' = None, *,
count: 'int' = 0,
**kwargs: 'Any') -> 'Schema_CCNew':
"""Make TCP connection count new option.
Args:
code: option code
opt: option data
count: connection count
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
Raises:
ProtocolError: If ``count`` does not fit 32 bits.
"""
if opt is not None:
count = opt.cc
return Schema_CCNew(
kind=code,
length=6,
count=self._make_uint(count, bits=32, name=f'[OptNo {code}] count'),
)
[docs]
def _make_mode_ccecho(self, code: 'Enum_Option', opt: 'Optional[Data_CCEcho]' = None, *,
count: 'int' = 0,
**kwargs: 'Any') -> 'Schema_CCEcho':
"""Make TCP connection count echo option.
Args:
code: option code
opt: option data
count: connection count
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
Raises:
ProtocolError: If ``count`` does not fit 32 bits.
"""
if opt is not None:
count = opt.cc
return Schema_CCEcho(
kind=code,
length=6,
count=self._make_uint(count, bits=32, name=f'[OptNo {code}] count'),
)
[docs]
def _make_mode_chkreq(self, code: 'Enum_Option', opt: 'Optional[Data_AlternateChecksumRequest]' = None, *,
algorithm: 'Enum_Checksum | int' = Enum_Checksum.TCP_checksum,
**kwargs: 'Any') -> 'Schema_AlternateChecksumRequest':
"""Make TCP alternate checksum request option.
Args:
code: option code
opt: option data
algorithm: checksum algorithm
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
Raises:
ProtocolError: If ``algorithm`` is a name, a :obj:`bool`, or a
member or code that does not fit 8 bits.
"""
if opt is not None:
algorithm = opt.chksum
return Schema_AlternateChecksumRequest(
kind=code,
length=3,
algorithm=self._make_index(algorithm, Enum_Checksum, bits=8, name=f'[OptNo {code}] algorithm'),
)
[docs]
def _make_mode_chksum(self, code: 'Enum_Option', opt: 'Optional[Data_AlternateChecksumData]' = None, *,
data: 'bytes' = b'',
**kwargs: 'Any') -> 'Schema_AlternateChecksumData':
"""Make TCP alternate checksum data option.
Args:
code: option code
opt: option data
data: checksum data
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
"""
if opt is not None:
data = opt.data
return Schema_AlternateChecksumData(
kind=code,
length=2 + len(data),
data=data,
)
[docs]
def _make_mode_sig(self, code: 'Enum_Option', opt: 'Optional[Data_MD5Signature]' = None, *,
digest: 'bytes' = bytes(16),
**kwargs: 'Any') -> 'Schema_MD5Signature':
"""Make TCP MD5 signature option.
Args:
code: option code
opt: option data
digest: digest
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
Raises:
ProtocolError: If ``digest`` is not 16 octets.
"""
if opt is not None:
digest = opt.digest
return Schema_MD5Signature(
kind=code,
length=18,
digest=self._make_octets(digest, size=16, name=f'[OptNo {code}] digest'),
)
[docs]
def _make_mode_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartResponse]' = None, *,
rate: 'int' = 0,
diff: 'timedelta | int' = 0,
nonce: 'int' = 0,
reserved: 'int' = 0,
nonce_reserved: 'int' = 0,
**kwargs: 'Any') -> 'Schema_QuickStartResponse':
"""Make TCP quick start response option.
Args:
code: option code
opt: option data
rate: rate (in kbps), rounded down to the nearest rate the
four-bit rate code carries (:rfc:`4782#section-4.2`)
diff: TTL Diff, the IP TTL minus the QS TTL, modulo 256 (:rfc:`4782`
Section 4.2); a :class:`~datetime.timedelta` is taken as its whole
seconds
nonce: nonce value
reserved: reserved bits before the rate request (``Resv.``, 4 bits)
nonce_reserved: reserved bits after the nonce (``R``, 2 bits)
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
Raises:
ProtocolError: If the rate code cannot carry ``rate``, or ``nonce``,
``reserved`` or ``nonce_reserved`` does not fit its field.
"""
if opt is not None:
rate = opt.req_rate
diff = opt.ttl_diff
nonce = opt.nonce
reserved = opt.reserved
nonce_reserved = opt.nonce_reserved
rate_val = self._make_qs_rate(rate)
diff_val = diff if isinstance(diff, int) else math.floor(diff.total_seconds())
nonce = self._make_uint(nonce, bits=30, name='Quick-Start nonce')
reserved = self._make_uint(reserved, bits=4, name='Quick-Start reserved')
nonce_reserved = self._make_uint(nonce_reserved, bits=2, name='Quick-Start nonce reserved')
return Schema_QuickStartResponse(
kind=code,
length=8,
flags={
'reserved': reserved,
'rate': rate_val,
},
diff=diff_val,
nonce={
'nonce': nonce,
'reserved': nonce_reserved,
},
)
[docs]
def _make_mode_timeout(self, code: 'Enum_Option', opt: 'Optional[Data_UserTimeout]' = None, *,
timeout: 'timedelta | int' = 0,
granularity: 'Optional[bool]' = None,
**kwargs: 'Any') -> 'Schema_UserTimeout':
"""Make TCP user timeout option.
Args:
code: option code
opt: option data
timeout: timeout value (in seconds, if an :obj:`int`)
granularity: preferred unit, minutes (``G`` set) or seconds; it is
kept only while the value still fits in it -- seconds within 15
bits, minutes as a whole number of them. Otherwise, and if not
given, seconds are used when they fit in 15 bits, and minutes if
not.
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
Raises:
ProtocolError: If the timeout is a :obj:`bool`, is neither an
:class:`int` nor a :class:`~datetime.timedelta`, is negative,
or does not fit in 15 bits of minutes.
"""
if opt is not None:
timeout = opt.timeout
granularity = opt.granularity
if isinstance(timeout, datetime.timedelta):
seconds = math.floor(timeout.total_seconds()) # type: Optional[int]
elif isinstance(timeout, int) and not isinstance(timeout, bool):
seconds = int(timeout)
else:
seconds = None
if seconds is None or seconds < 0:
raise ProtocolError(f'TCP: invalid [OptNo {code}] timeout: {timeout!r}')
# NOTE: A parsed or requested unit the value does not fit (seconds past 15 bits, or
# minutes not whole) falls back to the size-based choice, as with no unit given.
fits_seconds = seconds.bit_length() <= 15
if granularity is None or (not granularity and not fits_seconds) or (granularity and seconds % 60):
granularity = not fits_seconds
timeout_val = math.floor(seconds / 60) if granularity else seconds
if timeout_val.bit_length() > 15:
raise ProtocolError(f'TCP: [OptNo {code}] timeout value too large: {seconds} seconds')
return Schema_UserTimeout(
kind=code,
length=4,
info={
'granularity': granularity,
'timeout': timeout_val,
},
)
[docs]
def _make_mode_ao(self, code: 'Enum_Option', opt: 'Optional[Data_Authentication]' = None, *,
key_id: 'int' = 0,
next_key_id: 'int' = 0,
mac: 'bytes' = b'',
**kwargs: 'Any') -> 'Schema_Authentication':
"""Make TCP authentication option.
Args:
code: option code
opt: option data
key_id: key ID
next_key_id: next key ID
mac: MAC value
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
Raises:
ProtocolError: If ``key_id`` or ``next_key_id`` does not fit 8
bits.
"""
if opt is not None:
key_id = opt.key_id
next_key_id = opt.next_key_id
mac = opt.mac
return Schema_Authentication(
kind=code,
length=4 + len(mac),
key_id=self._make_uint(key_id, bits=8, name=f'[OptNo {code}] key_id'),
next_key_id=self._make_uint(next_key_id, bits=8, name=f'[OptNo {code}] next_key_id'),
mac=mac,
)
[docs]
def _make_mode_mp(self, code: 'Enum_Option', opt: 'Optional[Data_MPTCP]' = None, *,
subtype: 'Enum_MPTCPOption | int' = Enum_MPTCPOption.MP_CAPABLE,
**kwargs: 'Any') -> 'Schema_MPTCP':
"""Make multipath TCP option.
Args:
code: option code
opt: option data
subtype: MPTCP subtype
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
Raises:
ProtocolError: If ``subtype`` is a name, a :obj:`bool`, or a member
or code that does not fit 4 bits.
"""
if opt is not None:
subtype = opt.subtype
subtype_val = self._make_index(subtype, Enum_MPTCPOption, bits=4, name=f'[OptNo {code}] subtype')
name = self._lookup_registry(self.__mp_option__, subtype_val)
if isinstance(name, str):
meth_name = f'_make_mptcp_{name}'
meth = cast('MPOptionConstructor',
getattr(self, meth_name, self._make_mptcp_unknown))
else:
meth = name[1]
schema = meth(subtype_val, opt, **kwargs)
# NOTE: ``Schema_MPTCP.subtype`` is no field (see the NOTE on ``MPTCP``), so it is
# set here, as ``_MPTCP.post_process`` sets it on unpacking.
schema.subtype = subtype_val
return schema
[docs]
def _make_mptcp_unknown(self, subtype: 'Enum_MPTCPOption', opt: 'Optional[Data_MPTCPUnknown]' = None, *,
data: 'bytes' = b'\x00',
**kwargs: 'Any') -> 'Schema_MPTCPUnknown':
"""Make unknown multipath TCP option.
Args:
subtype: MPTCP subtype
opt: option data
data: option payload data, its first octet the four bits after
the subtype
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
Raises:
ProtocolError: If the first octet of ``data`` does not fit four
bits.
"""
if opt is not None:
data = opt.data
# NOTE: Sized from what the schema writes (#2272): three octets, the third taking
# ``data[0]``, then ``data[1:]``; ``2 + len(data)`` would not read back for ``b''``.
return Schema_MPTCPUnknown(
kind=cast('Enum_Option', Enum_Option.Multipath_TCP),
length=3 + len(data[1:]),
test={
'subtype': int(subtype),
'data': self._make_uint(data[0], bits=4, name=f'[OptNo {Enum_Option.Multipath_TCP}] '
f'subtype {subtype} first data octet') if data else 0,
},
data=data[1:],
)
[docs]
def _make_mptcp_capable(self, subtype: 'Enum_MPTCPOption', opt: 'Optional[Data_MPTCPCapable]' = None, *,
version: 'int' = 0,
flag_req: 'bool' = False,
flag_ext: 'bool' = False,
flag_deny_join: 'bool' = False,
flag_reserved: 'int' = 0,
flag_hsa: 'bool' = False,
skey: 'Optional[int]' = 0,
rkey: 'Optional[int]' = 0,
dl_len: 'Optional[int]' = None,
checksum: 'Optional[bytes]' = None,
**kwargs: 'Any') -> 'Schema_MPTCPCapable':
"""Make multipath TCP capable option.
Each field is present only if every field before it is, which selects
one of the five :rfc:`8684` section 3.1 lengths: 4 (``skey=None``), 12
(``rkey=None``), 20 (``dl_len=None``), 22 (``checksum=None``) or 24.
Args:
subtype: MPTCP subtype
opt: option data
version: MPTCP version
flag_req: checksum required flag
flag_ext: extensibility flag
flag_deny_join: flag for not establishing new subflows to the
source address
flag_reserved: reserved flags (``D`` to ``G``)
flag_hsa: use of HMAC-SHA1 (version 0) or HMAC-SHA256
(version 1) flag
skey: option sender's key
rkey: option receiver's key
dl_len: data-level length
checksum: checksum
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
Raises:
ProtocolError: If a field is given without the field before it, or
does not fit its width.
"""
if opt is not None:
version = opt.version
flag_req = opt.flags.req
flag_ext = opt.flags.ext
flag_deny_join = opt.flags.deny_join
flag_reserved = opt.flags.reserved
flag_hsa = opt.flags.hsa
skey = opt.skey
rkey = opt.rkey
dl_len = opt.dl_len
checksum = opt.checksum
names = ('skey', 'rkey', 'dl_len', 'checksum')
fields = (skey, rkey, dl_len, checksum)
present = fields.index(None) if None in fields else len(fields)
for index in range(present, len(fields)):
if fields[index] is not None:
raise ProtocolError(
f'{self.alias}: [OptNo {Enum_Option.Multipath_TCP}] invalid {names[index]}: '
f'must be None when {names[present]} is None, not {fields[index]!r}')
where = f'[OptNo {Enum_Option.Multipath_TCP}] {subtype.name}'
return Schema_MPTCPCapable(
kind=cast('Enum_Option', Enum_Option.Multipath_TCP),
# NOTE: :rfc:`8684` section 3.1 gives MP_CAPABLE as 4, 12, 20, 22
# or 24 octets, by how many of the fields above are present.
length=(4, 12, 20, 22, 24)[present],
test={
'subtype': subtype.value,
'version': self._make_uint(version, bits=4, name=f'{where} version'),
},
flags={
'req': flag_req,
'ext': flag_ext,
'deny_join': flag_deny_join,
'reserved': self._make_uint(flag_reserved, bits=4, name=f'{where} flag_reserved'),
'hsa': flag_hsa,
},
skey=None if skey is None else self._make_uint(skey, bits=64, name=f'{where} skey'),
rkey=None if rkey is None else self._make_uint(rkey, bits=64, name=f'{where} rkey'),
dl_len=None if dl_len is None else self._make_uint(dl_len, bits=16, name=f'{where} dl_len'),
checksum=None if checksum is None else self._make_octets(checksum, size=2, name=f'{where} checksum'),
)
[docs]
def _make_mptcp_join(self, subtype: 'Enum_MPTCPOption', opt: 'Optional[Data_MPTCPJoin]' = None, **kwargs: 'Any') -> 'Schema_MPTCPJoin':
"""Make multipath TCP join option.
Args:
subtype: MPTCP subtype
opt: option data
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
Raises:
ProtocolError: If the segment's flags are none of SYN, SYN/ACK and
ACK.
"""
if Enum_Flags.SYN in self._flags and Enum_Flags.ACK not in self._flags: # MP_JOIN-SYN
return self._make_join_syn(subtype, opt, **kwargs) # type: ignore[arg-type]
if Enum_Flags.SYN in self._flags and Enum_Flags.ACK in self._flags: # MP_JOIN-SYN/ACK
return self._make_join_synack(subtype, opt, **kwargs) # type: ignore[arg-type]
if Enum_Flags.SYN not in self._flags and Enum_Flags.ACK in self._flags: # MP_JOIN-ACK
return self._make_join_ack(subtype, opt, **kwargs) # type: ignore[arg-type]
raise ProtocolError(f'{self.alias}: [OptNo {Enum_Option.Multipath_TCP}] {subtype}: invalid flags combination')
[docs]
def _make_mptcp_dss(self, subtype: 'Enum_MPTCPOption', opt: 'Optional[Data_MPTCPDSS]' = None, *,
reserved: 'int' = 0,
data_fin: 'bool' = False,
ack_wide: 'Optional[bool]' = None,
ack: 'Optional[int]' = None,
dsn_wide: 'Optional[bool]' = None,
dsn: 'Optional[int]' = None,
ssn: 'Optional[int]' = None,
dl_len: 'Optional[int]' = None,
checksum: 'Optional[bytes]' = None,
**kwargs: 'Any') -> 'Schema_MPTCPDSS':
"""Make multipath TCP DSS option.
Args:
subtype: MPTCP subtype
opt: option data
reserved: reserved bits (7 bits)
data_fin: ``DATA_FIN`` flag
ack_wide: whether Data ACK is 8 octets; if not given, it is 8 octets
only when ``ack`` does not fit in 4
ack: Data ACK
dsn_wide: whether data sequence number is 8 octets; if not given, it
is 8 octets only when ``dsn`` does not fit in 4
dsn: data sequence number
ssn: subflow sequence number
dl_len: data-level length
checksum: checksum, or ``None`` if checksums were not negotiated
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
Raises:
ProtocolError: If the mapping fields are given only in part, a
4-octet field is asked to hold a value wider than 32 bits, or a
field does not fit its width.
"""
if opt is not None:
reserved = opt.reserved
data_fin = opt.data_fin
ack_wide = opt.ack_wide
ack = opt.ack
dsn_wide = opt.dsn_wide
dsn = opt.dsn
ssn = opt.ssn
dl_len = opt.dl_len
checksum = opt.checksum
where = f'[OptNo {Enum_Option.Multipath_TCP}] {subtype.name}'
reserved = self._make_uint(reserved, bits=7, name=f'{where} reserved')
if ack is not None:
ack = self._make_uint(ack, bits=64, name=f'{where} ack')
if dsn is not None:
dsn = self._make_uint(dsn, bits=64, name=f'{where} dsn')
flag_A = ack is not None
flag_a = ack_wide if ack_wide is not None else (flag_A and cast('int', ack).bit_length() > 32)
flag_M = dsn is not None
flag_m = dsn_wide if dsn_wide is not None else (flag_M and cast('int', dsn).bit_length() > 32)
if (flag_A and not flag_a and cast('int', ack).bit_length() > 32) or \
(flag_M and not flag_m and cast('int', dsn).bit_length() > 32):
raise ProtocolError(f'{self.alias}: [OptNo {Enum_Option.Multipath_TCP}] {subtype}: value too wide for 4 octets')
if flag_M and (ssn is None or dl_len is None):
raise ProtocolError(f'{self.alias}: [OptNo {Enum_Option.Multipath_TCP}] {subtype}: missing required fields')
if not flag_M and (ssn is not None or dl_len is not None or checksum is not None):
raise ProtocolError(f'{self.alias}: [OptNo {Enum_Option.Multipath_TCP}] {subtype}: missing required fields')
return Schema_MPTCPDSS(
kind=cast('Enum_Option', Enum_Option.Multipath_TCP),
# NOTE: RFC 8684, section 3.3, figure 9. ``a`` and ``m`` widen only a present
# field (the Data ACK if ``A``, the DSN if ``M``), so a stray flag packs back.
length=4 + ((8 if flag_a else 4) if flag_A else 0)
+ ((8 if flag_m else 4) + 6 + (2 if checksum is not None else 0) if flag_M else 0),
test={
'subtype': subtype.value,
'reserved': reserved >> 3,
},
flags={
'reserved': reserved & 0b111,
'F': data_fin,
'A': flag_A,
'a': flag_a,
'M': flag_M,
'm': flag_m,
},
ack=ack,
dsn=dsn,
ssn=None if ssn is None else self._make_uint(ssn, bits=32, name=f'{where} ssn'),
dl_len=None if dl_len is None else self._make_uint(dl_len, bits=16, name=f'{where} dl_len'),
checksum=None if checksum is None else self._make_octets(checksum, size=2, name=f'{where} checksum'),
)
[docs]
def _make_mptcp_addaddr(self, subtype: 'Enum_MPTCPOption', opt: 'Optional[Data_MPTCPAddAddress]' = None, *,
reserved: 'Optional[int]' = None,
echo: 'Optional[bool]' = None,
addr_id: 'int' = 0,
addr: 'IPv4Address | IPv6Address | int | bytes | str' = '0.0.0.0', # nosec: B104
port: 'Optional[int]' = None,
hmac: 'Optional[bytes]' = None,
**kwargs: 'Any') -> 'Schema_MPTCPAddAddress':
"""Make multipath TCP add address option.
Args:
subtype: MPTCP subtype
opt: option data
reserved: reserved bits
echo: echo flag
addr_id: address ID
addr: address
port: port number
hmac: truncated HMAC
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
Raises:
ProtocolError: If ``hmac`` is not 8 octets, or is given together
with ``echo`` set, or a field does not fit its width.
FieldValueError: If ``addr`` is a :obj:`bool` or not a valid IP
address.
Note:
:rfc:`8684` section 3.4.1 allows an ADD_ADDR without the truncated
HMAC only as an echo (``E`` set). So when none of ``reserved``,
``echo`` and ``hmac`` is given, the option is built in the
:rfc:`6824` layout instead, its ``IPVer`` nibble taking the place of
the reserved bits and ``E``; otherwise ``reserved`` defaults to 0 and
``echo`` to whether ``hmac`` is absent. A parsed ``opt`` is rebuilt
as it was read, without those checks.
"""
if opt is not None:
reserved = opt.reserved
echo = opt.echo
addr_id = opt.addr_id
addr_val = opt.addr
port = opt.port
hmac = opt.hmac
else:
# NOTE: ``parse_ip_address`` refuses a ``bool``; ``ipaddress.ip_address`` would
# make it ``0.0.0.1`` before the schema's own guard sees it.
addr_val = parse_ip_address(
addr, f'{self.alias}: [OptNo {Enum_Option.Multipath_TCP}] invalid address')
version = addr_val.version
if reserved is None and echo is None and hmac is None:
reserved, echo = version >> 1, bool(version & 1)
if reserved is None:
reserved = 0
if echo is None:
echo = hmac is None
if opt is None and hmac is not None and (echo or len(hmac) != 8):
raise ProtocolError(f'{self.alias}: [OptNo {Enum_Option.Multipath_TCP}] {subtype}: invalid HMAC')
where = f'[OptNo {Enum_Option.Multipath_TCP}] {subtype.name}'
return Schema_MPTCPAddAddress(
kind=cast('Enum_Option', Enum_Option.Multipath_TCP),
length=4 + (4 if version == 4 else 16) + (2 if port is not None else 0) + (8 if hmac is not None else 0),
test={
'subtype': subtype.value,
'reserved': self._make_uint(reserved, bits=3, name=f'{where} reserved'),
'echo': echo,
},
addr_id=self._make_uint(addr_id, bits=8, name=f'{where} addr_id'),
address=addr_val,
port=None if port is None else self._make_uint(port, bits=16, name=f'{where} port'),
hmac=hmac,
)
[docs]
def _make_mptcp_remove(self, subtype: 'Enum_MPTCPOption', opt: 'Optional[Data_MPTCPRemoveAddress]' = None, *,
reserved: 'int' = 0,
addr_id: 'Optional[list[int]]' = None,
**kwargs: 'Any') -> 'Schema_MPTCPRemoveAddress':
"""Make multipath TCP remove address option.
Args:
subtype: MPTCP subtype
opt: option data
reserved: reserved bits
addr_id: address ID list
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
Raises:
ProtocolError: If ``reserved`` does not fit 4 bits, or an address
ID 8 bits.
"""
if opt is not None:
reserved = opt.reserved
addr_id_list = cast('list[int]', opt.addr_id)
else:
addr_id_list = addr_id if addr_id is not None else []
where = f'[OptNo {Enum_Option.Multipath_TCP}] {subtype.name}'
for item in addr_id_list:
self._make_uint(item, bits=8, name=f'{where} addr_id')
return Schema_MPTCPRemoveAddress(
kind=cast('Enum_Option', Enum_Option.Multipath_TCP),
# NOTE: RFC 8684, section 3.4.2, figure 13: ``Length = 3 + n``, one octet per
# Address ID, which ``MPTCPRemoveAddress.addr_id`` also sizes itself from.
length=3 + len(addr_id_list),
test={
'subtype': subtype.value,
'reserved': self._make_uint(reserved, bits=4, name=f'{where} reserved'),
},
addr_id=addr_id_list,
)
[docs]
def _make_mptcp_prio(self, subtype: 'Enum_MPTCPOption', opt: 'Optional[Data_MPTCPPriority]' = None, *,
reserved: 'int' = 0,
backup: 'bool' = False,
addr_id: 'Optional[int]' = None,
**kwargs: 'Any') -> 'Schema_MPTCPPriority':
"""Make multipath TCP priority option.
Args:
subtype: MPTCP subtype
opt: option data
reserved: reserved bits
backup: backup flag
addr_id: address ID
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
Raises:
ProtocolError: If ``reserved`` does not fit 3 bits, or ``addr_id``
8 bits.
"""
if opt is not None:
# NOTE: ``backup`` is carried over, or rebuilding a parsed MP_PRIO would write
# ``B=0``; the ``B`` flag is this option's entire payload.
reserved = opt.reserved
backup = opt.backup
addr_id = opt.addr_id
where = f'[OptNo {Enum_Option.Multipath_TCP}] {subtype.name}'
return Schema_MPTCPPriority(
kind=cast('Enum_Option', Enum_Option.Multipath_TCP),
# NOTE: MP_PRIO is 3 octets (RFC 8684, section 3.3.8, figure 11); RFC 6824's
# 4 with an Address ID is still accepted, so ``addr_id`` decides the length.
length=3 if addr_id is None else 4,
test={
'subtype': subtype.value,
'reserved': self._make_uint(reserved, bits=3, name=f'{where} reserved'),
'backup': backup,
},
addr_id=None if addr_id is None else self._make_uint(addr_id, bits=8, name=f'{where} addr_id'),
)
[docs]
def _make_mptcp_fail(self, subtype: 'Enum_MPTCPOption', opt: 'Optional[Data_MPTCPFallback]' = None, *,
reserved: 'int' = 0,
dsn: 'int' = 0,
**kwargs: 'Any') -> 'Schema_MPTCPFallback':
"""Make multipath TCP fail option.
Args:
subtype: MPTCP subtype
opt: option data
reserved: reserved bits
dsn: data sequence number
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
Raises:
ProtocolError: If ``reserved`` does not fit 12 bits, or ``dsn`` 64
bits.
"""
if opt is not None:
reserved = opt.reserved
dsn = opt.dsn
where = f'[OptNo {Enum_Option.Multipath_TCP}] {subtype.name}'
return Schema_MPTCPFallback(
kind=cast('Enum_Option', Enum_Option.Multipath_TCP),
length=12,
test={
'subtype': subtype.value,
'reserved': self._make_uint(reserved, bits=12, name=f'{where} reserved'),
},
dsn=self._make_uint(dsn, bits=64, name=f'{where} dsn'),
)
[docs]
def _make_mptcp_fastclose(self, subtype: 'Enum_MPTCPOption', opt: 'Optional[Data_MPTCPFastclose]' = None, *,
reserved: 'int' = 0,
key: 'int' = 0,
**kwargs: 'Any') -> 'Schema_MPTCPFastclose':
"""Make multipath TCP fastclose option.
Args:
subtype: MPTCP subtype
opt: option data
reserved: reserved bits
key: option receiver's key
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
Raises:
ProtocolError: If ``reserved`` does not fit 12 bits, or ``key`` 64
bits.
"""
if opt is not None:
reserved = opt.reserved
key = opt.rkey
where = f'[OptNo {Enum_Option.Multipath_TCP}] {subtype.name}'
return Schema_MPTCPFastclose(
kind=cast('Enum_Option', Enum_Option.Multipath_TCP),
length=12,
test={
'subtype': subtype.value,
'reserved': self._make_uint(reserved, bits=12, name=f'{where} reserved'),
},
key=self._make_uint(key, bits=64, name=f'{where} key'),
)
[docs]
def _make_mode_fastopen(self, code: 'Enum_Option', opt: 'Optional[Data_FastOpenCookie]' = None, *,
cookie: 'Optional[bytes]' = None,
**kwargs: 'Any') -> 'Schema_FastOpenCookie':
"""Make TCP Fast Open option.
Args:
code: option code
opt: option data
cookie: fast open cookie; :obj:`None` or ``b''`` for a cookie request
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
"""
if opt is not None:
cookie = opt.cookie
return Schema_FastOpenCookie(
kind=code,
length=2 + (len(cookie) if cookie is not None else 0),
cookie=cookie,
)
[docs]
def _verify_pseudo_checksum(self, pseudo: 'Optional[IP_Pseudo]') -> 'None':
"""Verify the checksum over the pseudo-header of the enclosing IP layer.
That layer calls this once parsed, see
:func:`~pcapkit.protocols.misc.checksum.ip_pseudo.protocol.verify_upper`. A zero
checksum, or the pseudo-header's sum alone, is what checksum offload
leaves on the sending host, and is
:attr:`~pcapkit.corekit.checksum.internet.ChecksumStatus.NOT_COMPUTED`.
Args:
pseudo: Pseudo-header, or :data:`None` if the IP layer gives none.
"""
from pcapkit.corekit.checksum.internet import ( # isort: skip # pylint: disable=import-outside-toplevel
ChecksumStatus, partial_checksum, verify_checksum)
if pseudo is None:
status = ChecksumStatus.NO_PSEUDO_HEADER
else:
status = verify_checksum(self.alias, self._data, self._info.checksum, length=pseudo.info.len,
prefix=pseudo.data, offload=(b'\x00\x00', partial_checksum(pseudo.data)))
self._info.__update__(checksum_status=status)
[docs]
def _read_join_syn(self, schema: 'Schema_MPTCPJoinSYN', *, options: 'Option') -> 'Data_MPTCPJoinSYN': # pylint: disable=unused-argument
"""Read Join Connection option for Initial SYN.
Structure of ``MP_JOIN-SYN`` [:rfc:`8684`, section 3.2, figure 5]:
.. code-block:: text
1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+---------------+---------------+-------+-----+-+---------------+
| Kind | Length = 12 |Subtype| |B| Address ID |
+---------------+---------------+-------+-----+-+---------------+
| Receiver's Token (32 bits) |
+---------------------------------------------------------------+
| Sender's Random Number (32 bits) |
+---------------------------------------------------------------+
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
Raises:
ProtocolError: If length is **NOT** ``12``.
"""
if schema.length != 12:
raise ProtocolError(f'{self.alias}: [OptNo {schema.kind}] invalid format')
data = Data_MPTCPJoinSYN(
kind=Enum_Option.Multipath_TCP, # type: ignore[arg-type]
length=schema.length,
subtype=schema.subtype,
connection=Enum_Flags.SYN, # type: ignore[arg-type]
reserved=schema.test['reserved'],
backup=bool(schema.test['backup']),
addr_id=schema.addr_id,
token=schema.token,
nonce=schema.nonce,
)
return data
[docs]
def _read_join_synack(self, schema: 'Schema_MPTCPJoinSYNACK', options: 'Option') -> 'Data_MPTCPJoinSYNACK': # pylint: disable=unused-argument
"""Read Join Connection option for Responding SYN/ACK.
Structure of ``MP_JOIN-SYN/ACK`` [:rfc:`8684`, section 3.2, figure 6]:
.. code-block:: text
1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+---------------+---------------+-------+-----+-+---------------+
| Kind | Length = 16 |Subtype| |B| Address ID |
+---------------+---------------+-------+-----+-+---------------+
| |
| Sender's Truncated HMAC (64 bits) |
| |
+---------------------------------------------------------------+
| Sender's Random Number (32 bits) |
+---------------------------------------------------------------+
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
Raises:
ProtocolError: If length is **NOT** ``16``.
Note:
The accepted length is ``16``, as in the figure above (:rfc:`8684`
section 3.2 figure 6) and in what
:class:`~pcapkit.protocols.transport.tcp.schema.MPTCPJoinSYNACK`
packs and unpacks: ``Kind`` (1) + ``Length`` (1) + subtype/flags (1)
+ ``Address ID`` (1) + the truncated HMAC (8) + the random number (4).
"""
if schema.length != 16:
raise ProtocolError(f'{self.alias}: [OptNo {schema.kind}] invalid format')
data = Data_MPTCPJoinSYNACK(
kind=Enum_Option.Multipath_TCP, # type: ignore[arg-type]
length=schema.length,
subtype=schema.subtype,
connection=Enum_Flags.SYN | Enum_Flags.ACK, # type: ignore[arg-type]
reserved=schema.test['reserved'],
backup=bool(schema.test['backup']),
addr_id=schema.addr_id,
hmac=schema.hmac,
nonce=schema.nonce,
)
return data
[docs]
def _read_join_ack(self, schema: 'Schema_MPTCPJoinACK', *, options: 'Option') -> 'Data_MPTCPJoinACK': # pylint: disable=unused-argument
"""Read Join Connection option for Third ACK.
Structure of ``MP_JOIN-ACK`` [:rfc:`8684`, section 3.2, figure 7]:
.. code-block:: text
1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+---------------+---------------+-------+-----------------------+
| Kind | Length = 24 |Subtype| (reserved) |
+---------------+---------------+-------+-----------------------+
| |
| |
| Sender's HMAC (160 bits) |
| |
| |
+---------------------------------------------------------------+
Arguments:
schema: parsed option schema
options: extracted TCP options
Returns:
Parsed option data.
Raises:
ProtocolError: If length is **NOT** ``24``.
"""
if schema.length != 24:
raise ProtocolError(f'{self.alias}: [OptNo {schema.kind}] invalid format')
data = Data_MPTCPJoinACK(
kind=Enum_Option.Multipath_TCP, # type: ignore[arg-type]
length=schema.length,
subtype=schema.subtype,
connection=Enum_Flags.ACK, # type: ignore[arg-type]
reserved=schema.test['reserved'],
hmac=schema.hmac,
)
return data
[docs]
def _make_join_syn(self, subtype: 'Enum_MPTCPOption', opt: 'Optional[Data_MPTCPJoinSYN]' = None, *,
reserved: 'int' = 0,
backup: 'bool' = False,
addr_id: 'int' = 0,
token: 'int' = 0,
nonce: 'int' = 0,
**kwargs: 'Any') -> 'Schema_MPTCPJoinSYN':
"""Make multipath TCP join SYN option.
Args:
subtype: MPTCP subtype
opt: option data
reserved: reserved bits
backup: backup flag
addr_id: address ID
token: receiver's token
nonce: sender's random number
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
Raises:
ProtocolError: If a field does not fit its width.
"""
if opt is not None:
reserved = opt.reserved
backup = opt.backup
addr_id = opt.addr_id
token = opt.token
nonce = opt.nonce
where = f'[OptNo {Enum_Option.Multipath_TCP}] {subtype.name}'
return Schema_MPTCPJoinSYN(
kind=cast('Enum_Option', Enum_Option.Multipath_TCP),
length=12,
test={
'subtype': subtype.value,
'reserved': self._make_uint(reserved, bits=3, name=f'{where} reserved'),
'backup': backup,
},
addr_id=self._make_uint(addr_id, bits=8, name=f'{where} addr_id'),
token=self._make_uint(token, bits=32, name=f'{where} token'),
nonce=self._make_uint(nonce, bits=32, name=f'{where} nonce'),
)
[docs]
def _make_join_synack(self, subtype: 'Enum_MPTCPOption', opt: 'Optional[Data_MPTCPJoinSYNACK]' = None, *,
reserved: 'int' = 0,
backup: 'bool' = False,
addr_id: 'int' = 0,
hmac: 'bytes' = bytes(8),
nonce: 'int' = 0,
**kwargs: 'Any') -> 'Schema_MPTCPJoinSYNACK':
"""Make multipath TCP join SYN/ACK option.
Args:
subtype: MPTCP subtype
opt: option data
reserved: reserved bits
backup: backup flag
addr_id: address ID
hmac: sender's truncated HMAC
nonce: sender's random number
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
Raises:
ProtocolError: If a field does not fit its width, or ``hmac`` is
not eight octets.
"""
if opt is not None:
reserved = opt.reserved
backup = opt.backup
addr_id = opt.addr_id
# NOTE: ``hmac`` is carried over, or rebuilding a parsed MP_JOIN-SYN/ACK would
# substitute the ``bytes(8)`` default for the truncated HMAC on the wire.
hmac = opt.hmac
nonce = opt.nonce
where = f'[OptNo {Enum_Option.Multipath_TCP}] {subtype.name}'
return Schema_MPTCPJoinSYNACK(
kind=cast('Enum_Option', Enum_Option.Multipath_TCP),
# NOTE: RFC 8684, section 3.2, figure 6: 4 octets, the truncated
# HMAC (8) and the random number (4).
length=16,
test={
'subtype': subtype.value,
'reserved': self._make_uint(reserved, bits=3, name=f'{where} reserved'),
'backup': backup,
},
addr_id=self._make_uint(addr_id, bits=8, name=f'{where} addr_id'),
hmac=self._make_octets(hmac, size=8, name=f'{where} hmac'),
nonce=self._make_uint(nonce, bits=32, name=f'{where} nonce'),
)
[docs]
def _make_join_ack(self, subtype: 'Enum_MPTCPOption', opt: 'Optional[Data_MPTCPJoinACK]' = None, *,
reserved: 'int' = 0,
hmac: 'bytes' = bytes(20),
**kwargs: 'Any') -> 'Schema_MPTCPJoinACK':
"""Make multipath TCP join ACK option.
Args:
subtype: MPTCP subtype
opt: option data
reserved: reserved bits
hmac: sender's HMAC
**kwargs: arbitrary keyword arguments
Returns:
Constructed option schema.
Raises:
ProtocolError: If ``reserved`` does not fit 12 bits, or ``hmac`` is
not 20 octets.
"""
if opt is not None:
reserved = opt.reserved
hmac = opt.hmac
where = f'[OptNo {Enum_Option.Multipath_TCP}] {subtype.name}'
return Schema_MPTCPJoinACK(
kind=cast('Enum_Option', Enum_Option.Multipath_TCP),
# NOTE: :rfc:`8684` section 3.2 figure 7: 4 octets, then the full 20-octet HMAC;
# ``_read_join_ack``'s guard requires the same 24.
length=24,
test={
'subtype': subtype.value,
'reserved': self._make_uint(reserved, bits=12, name=f'{where} reserved'),
},
hmac=self._make_octets(hmac, size=20, name=f'{where} hmac'),
)