# -*- coding: utf-8 -*-
"""data models for TCP reassembly"""
from typing import TYPE_CHECKING, Generic, TypeVar
from pcapkit.corekit.infoclass import Info, info_final
from pcapkit.corekit.packet import DeferredPacket
from pcapkit.foundation.reassembly.data.data import Completion, Deferred
from pcapkit.utilities.compat import Tuple
__all__ = [
'Packet', 'DatagramID', 'Datagram', 'HoleDescriptor',
'Fragment', 'Buffer', 'BufferID', 'HOLE_OPEN_BOUND',
]
if TYPE_CHECKING:
from ipaddress import IPv4Address, IPv6Address
from typing import Optional
from typing_extensions import TypeAlias
from pcapkit.const.reg.apptype import TCP as AppType_TCP
from pcapkit.protocols.protocol import ProtocolBase
_AT = TypeVar('_AT', 'IPv4Address', 'IPv6Address')
#: Buffer ID.
BufferID: 'TypeAlias' = Tuple[_AT, int, _AT, int]
#: Open bound of the :class:`HoleDescriptor` list. A hole whose ``last`` is
#: this value is still open above, and one whose ``first`` is its negation
#: stands for whatever preceded a capture that missed the handshake.
#:
#: It is ``2 ** 63 - 1`` on every platform, not :data:`sys.maxsize`, which is
#: ``2 ** 31 - 1`` on GraalPy and 32-bit builds and so falls inside the
#: ``2 ** 32`` sequence space. The unwrapped sequence line reaches it only
#: after ``2 ** 31`` wraps. It stays an :class:`int`, like the descriptor's
#: own fields, and in both signs fits a standard JSON number and a
#: :mod:`plistlib` integer, which :data:`math.inf` does not.
#:
#: .. versionadded:: 1.5.0
HOLE_OPEN_BOUND = (1 << 63) - 1
[docs]
@info_final
class Packet(Info):
"""Data model for :term:`TCP <reasm.tcp.packet>` packet representation."""
#: Buffer ID.
bufid: 'BufferID'
#: Data sequence number.
dsn: 'int'
#: Acknowledgment number.
ack: 'int'
#: Original packet range number.
num: 'int'
#: Synchronise flag.
syn: 'bool'
#: Finish flag.
fin: 'bool'
#: Reset connection flag.
rst: 'bool'
#: Payload length, header excluded.
len: 'int'
#: Sequence number of the first octet of :attr:`payload`, i.e. the segment's
#: own sequence number. Absolute, not an offset into any payload buffer.
first: 'int'
#: Sequence number of the last octet of :attr:`payload`, i.e. ``first +
#: len - 1``. **Inclusive**, so a segment carrying no payload at all has
#: :attr:`last` one below :attr:`first`.
last: 'int'
#: Raw :obj:`bytes` type header.
header: 'bytes'
#: Raw :obj:`bytearray` type payload.
payload: 'bytearray'
#: Capture timestamp of the segment, in seconds since the Unix epoch, i.e.
#: the *capture's* clock rather than the host's. It drives the reassembly
#: timeout, which for TCP is off by default -- see
#: :attr:`TCP.__timeout__ <pcapkit.foundation.reassembly.tcp.TCP.__timeout__>`.
timestamp: 'float'
if TYPE_CHECKING:
def __init__(self, bufid: 'BufferID', dsn: 'int', ack: 'int', num: 'int', syn: 'bool', fin: 'bool', rst: 'bool', len: 'int', first: 'int', last: 'int', header: 'bytes', payload: 'bytearray', timestamp: 'float') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin
[docs]
@info_final
class DatagramID(Info, Generic[_AT]):
"""Data model for :term:`TCP <reasm.tcp.datagram>` original packet identifier.
Note:
The ports here stay bare :obj:`int` port numbers, where the pair
:meth:`TCP.submit <pcapkit.foundation.reassembly.tcp.TCP.submit>` hands
:class:`~pcapkit.foundation.reassembly.data.data.Deferred` is one of
:class:`~pcapkit.const.reg.apptype.tcp.TCP` members
(:issue:`2417`). The two are not the same thing to name: this is the
flow's identity, split out of :data:`BufferID`, which every
:mod:`pcapkit.toolkit` adapter builds from a plain port number -- the
native ones by reading :attr:`AppType.port
<pcapkit.const.reg.apptype.apptype.AppType.port>` off the parsed header,
the third-party ones from whatever their own engine exposes -- and which
is then used as a :class:`dict` key and dumped as part of the report.
The deferred pair is a *dispatch* key for
:meth:`Transport.analyze
<pcapkit.protocols.transport.transport.Transport.analyze>`, which is
what gains from carrying the service name.
"""
#: Source address.
src: 'tuple[_AT, int]'
#: Destination address.
dst: 'tuple[_AT, int]'
#: Original packet ACK number.
ack: 'int'
if TYPE_CHECKING:
def __init__(self, src: 'tuple[_AT, int]', dst: 'tuple[_AT, int]', ack: 'int') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin
[docs]
@info_final
class Datagram(DeferredPacket, Info, Generic[_AT]):
"""Data model for :term:`TCP <reasm.tcp.datagram>` reassembled datagram."""
#: How completely the datagram was reassembled, and why reassembly stopped;
#: see :class:`~pcapkit.foundation.reassembly.data.data.Completion`. Only
#: :attr:`Completion.COMPLETE
#: <pcapkit.foundation.reassembly.data.data.Completion.COMPLETE>` is truthy, so
#: ``if datagram.completed:`` reads as it would for a :obj:`bool`.
completed: 'Completion'
#: Listing ``packet`` here is what makes it lazy -- see
#: :class:`~pcapkit.corekit.packet.DeferredPacket`.
__additional__ = ['packet']
#: Placeholder class :attr:`packet` may hold, for
#: :class:`~pcapkit.corekit.packet.DeferredPacket` to resolve. Unsubscripted
#: for the reason given on
#: :class:`~pcapkit.foundation.reassembly.data.ip.Datagram`'s own
#: ``__deferred__``: :func:`isinstance` rejects a parameterized generic.
__deferred__ = Deferred
#: Original packet identifier.
id: 'DatagramID[_AT]'
#: Packet numbers.
index: 'tuple[int, ...]'
#: Initial TCP header.
header: 'bytes'
#: Reassembled payload (application layer data).
payload: 'bytes | tuple[bytes, ...]'
#: Parsed TCP payload, as :func:`~pcapkit.foundation.analysis.tcp.tcp_datagram`
#: returns it unless the reassembly's ``__protocol_type__`` names another
#: protocol, which then parses it. Analysed on first read rather than at
#: construction; a :class:`~pcapkit.foundation.reassembly.data.data.Deferred`
#: may be passed in its place, and reading this attribute then runs it and
#: keeps the result.
packet: 'Optional[ProtocolBase]'
#: Sequence ranges on which two segments disagreed, i.e. where an arriving
#: segment overlapped bytes already buffered but did not repeat them.
#: Each entry is ``(first, last)``, absolute TCP sequence numbers and both
#: **inclusive** -- the same convention as :attr:`Packet.first` and
#: :attr:`Packet.last`. Empty when the stream never saw a contested byte.
#: A range that crosses ``2 ** 32`` therefore has ``last < first``.
#:
#: Resolution keeps the already-buffered bytes and discards the
#: conflicting portion of whichever segment arrived later, per
#: :rfc:`9293#section-3.10` ("we reconstruct the segment to contain just
#: the new data"); this field is what lets a caller tell a clean stream
#: from a contested one, since a contested range does not, on its own, leave a
#: hole for :attr:`completed` to report.
conflict: 'tuple[tuple[int, int], ...]'
if TYPE_CHECKING:
# NOTE: one signature, no ``@overload`` pair keyed on ``completed``, as on
# :class:`~pcapkit.foundation.reassembly.data.ip.Datagram`.
def __init__(self, completed: 'Completion', id: 'DatagramID[_AT]', index: 'tuple[int, ...]', header: 'bytes', payload: 'bytes | tuple[bytes, ...]', packet: 'Optional[ProtocolBase | Deferred[tuple[AppType_TCP, AppType_TCP]]]', conflict: 'tuple[tuple[int, int], ...]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin
[docs]
@info_final
class HoleDescriptor(Info):
"""Data model for :term:`TCP <reasm.tcp.buffer>` hole descriptor.
Both bounds are **unwrapped TCP sequence numbers** -- read modulo
``2 ** 32`` relative to the data already buffered, so a stream crossing
``2 ** 32`` continues past it -- and both are **inclusive**, so a hole
covers ``last - first + 1`` octets. They are not offsets into
:attr:`Fragment.raw`: the descriptor list is kept once per buffer ID,
across every acknowledgement number's payload buffer, and describes the
stream as a whole. It decides when a stream whose FIN has been seen is
whole. A hole whose ``first`` is ``-HOLE_OPEN_BOUND`` stands for whatever
preceded a capture that missed the handshake, and does not count; one
whose ``last`` is :data:`HOLE_OPEN_BOUND` is still open above.
.. versionadded:: 1.5.0
Replaces the misspelt ``HoleDiscriptor``.
.. versionremoved:: 1.5.0
``HoleDiscriptor``, and its re-export
``pcapkit.foundation.reassembly.data.TCP_HoleDiscriptor``, with no
alias; import :class:`~pcapkit.foundation.reassembly.data.tcp.HoleDescriptor` and ``TCP_HoleDescriptor`` instead.
"""
#: Sequence number of the first missing octet.
first: 'int'
#: Sequence number of the last missing octet, inclusive.
last: 'int'
if TYPE_CHECKING:
def __init__(self, first: 'int', last: 'int') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin
[docs]
@info_final
class Fragment(Info):
"""Data model for :term:`TCP <reasm.tcp.buffer>` ACK list fragment item."""
#: Numbers of the packets merged into this fragment.
ind: 'list[int]'
#: Sequence number of the octet held in ``raw[0]``, i.e. the origin this
#: buffer is indexed from: ``raw[n]`` holds the octet whose sequence number
#: is ``isn + n``. Revised downwards whenever a segment turns up below the
#: data already buffered, so it is not necessarily the connection's own
#: initial sequence number. Like every sequence number held in this
#: fragment -- :attr:`gap` and :attr:`conflict` included -- it is
#: unwrapped, and so may exceed ``2 ** 32`` on a stream that crossed it.
isn: 'int'
#: Length of payload buffer.
len: 'int'
#: Reassembled payload, holes set to ``b'\x00'``.
raw: 'bytearray'
#: Sequence ranges, absolute and inclusive, still zero-fill placeholder in
#: :attr:`raw` rather than an actually-received byte *of this fragment*.
#: Only the gap-creating sites in :meth:`TCP.reassembly
#: <pcapkit.foundation.reassembly.tcp.TCP.reassembly>` -- the forward
#: append and the reach-back prepend, the only places that ever splice a
#: ``bytearray(GAP)`` filler into :attr:`raw` -- add an entry; an overlap
#: merge only ever shrinks or removes one, filling it from the arriving
#: segment. The one exception lies wholly *outside* :attr:`raw`: when a
#: stream whose SYN was captured is split rather than padded across a gap
#: wider than :data:`~pcapkit.utilities.config.TCP_GAP_LIMIT`, both halves
#: record the range left between them. Any entry at all makes the fragment
#: incomplete.
#:
#: This is deliberately **not** derived from
#: :attr:`Buffer.hdl <pcapkit.foundation.reassembly.data.tcp.Buffer.hdl>`.
#: ``hdl`` is one list shared by every acknowledgement number under the
#: same buffer ID, so a segment landing in a *different* fragment can
#: close a hole in ``hdl`` that this fragment's own :attr:`raw` never
#: filled -- and consulting ``hdl`` to decide whether an overlapping
#: position here was "already received" then answers a question about
#: the wrong fragment. Tracking gaps on the fragment itself is what keeps
#: the merge in :meth:`TCP.reassembly
#: <pcapkit.foundation.reassembly.tcp.TCP.reassembly>` from discarding
#: this fragment's own real bytes because some *other* fragment happened
#: to have received something at the same absolute sequence numbers.
#:
#: Absolute sequence numbers rather than offsets into :attr:`raw` --
#: :attr:`conflict` below uses the same convention -- because it is what
#: :meth:`TCP.reassembly <pcapkit.foundation.reassembly.tcp.TCP.reassembly>`
#: already computes (``GAP = PSN - (ISN + LEN)`` and its mirror), so this
#: reuses an existing concept rather than adding a second one; and because a
#: gap entry then never needs shifting when :attr:`isn` is revised downward by the
#: reach-back path, unlike an offset-based or a per-octet representation.
#: A typical fragment carries zero or a handful of entries, against a
#: per-octet marker the length of the whole payload -- the difference
#: that matters on a clean stream, where the per-octet form pays a
#: buffer-sized cost to record that nothing is missing at all.
gap: 'list[tuple[int, int]]'
#: Sequence ranges, absolute and inclusive, on which an arriving segment
#: disagreed with bytes already held in :attr:`raw`. Accumulated across
#: every merge into this fragment, in the order the conflicts were found;
#: carried onto :attr:`Datagram.conflict
#: <pcapkit.foundation.reassembly.data.tcp.Datagram.conflict>` verbatim
#: when the buffer is submitted.
conflict: 'list[tuple[int, int]]'
if TYPE_CHECKING:
def __init__(self, ind: 'list[int]', isn: 'int', len: 'int', raw: 'bytearray', gap: 'list[tuple[int, int]]', conflict: 'list[tuple[int, int]]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin
[docs]
@info_final
class Buffer(Info):
"""Data model for :term:`TCP <reasm.tcp.buffer>` buffer entry."""
#: Hole descriptor list.
hdl: 'list[HoleDescriptor]'
#: Initial TCP header.
hdr: 'bytes'
#: ACK list.
ack: 'dict[int, Fragment]'
#: Capture timestamp of the **first** segment buffered under this buffer ID,
#: in seconds since the Unix epoch. Origin of the reassembly timer, and never
#: revised: a later segment does not extend the deadline.
timestamp: 'float'
if TYPE_CHECKING:
def __init__(self, hdl: 'list[HoleDescriptor]', hdr: 'bytes', ack: 'dict[int, Fragment]', timestamp: 'float') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin