# -*- coding: utf-8 -*-
"""Base Class
================
.. module:: pcapkit.foundation.reassembly.reassembly
:mod:`pcapkit.foundation.reassembly.reassembly` contains
:class:`~pcapkit.foundation.reassembly.reassembly.Reassembly` only,
which is an abstract base class for all reassembly classes,
based on algorithms described in :rfc:`791` and :rfc:`815`,
that implement datagram reassembly of IP and TCP packets.
"""
import abc
import math
from typing import TYPE_CHECKING, Generic, Type, TypeVar, cast
from pcapkit.protocols.misc.raw.protocol import Raw
from pcapkit.protocols.protocol import ProtocolBase
from pcapkit.utilities.exceptions import FieldValueError, UnsupportedCall
from pcapkit.utilities.logging import get_logger
from pcapkit.utilities.pickle import ReduceAsProtocol2
# NB: declared above the ``TYPE_CHECKING`` block, not below it, so that
# ``CallbackFn`` can name ``_DT`` outright. As a quoted forward reference it would
# resolve only from this module's namespace, and every module that spells
# ``CallbackFn`` in an annotation -- ``pcapkit.foundation.registry.foundation``
# does -- has to evaluate the alias in its own.
# packet
_PT = TypeVar('_PT', bound='Info')
# datagram
_DT = TypeVar('_DT', bound='Info')
# buffer ID
_IT = TypeVar('_IT', bound='tuple')
# buffer
_BT = TypeVar('_BT', bound='Info')
if TYPE_CHECKING:
from typing import Any, Callable, List, Optional, Type
from typing_extensions import Self
from pcapkit.corekit.infoclass import Info
from pcapkit.corekit.module import ModuleDescriptor
CallbackFn = Callable[[List[_DT]], None]
__all__ = ['Reassembly']
#: logging.Logger: Module-level logger, a child of the package-wide
#: :data:`pcapkit.utilities.logging.logger`.
logger = get_logger(__name__)
[docs]
class ReassemblyBase(ReduceAsProtocol2, Generic[_PT, _DT, _IT, _BT], metaclass=ReassemblyMeta):
"""Base class for reassembly procedure.
Args:
strict: if :data:`True`, report a datagram that is not completely
reassembled as the tuple of its received runs; otherwise as one
contiguous payload, its holes zero-filled -- or, for IP while the
total length is unknown, only the prefix up to the first hole
store: if store reassembled datagram in memory, i.e.,
:attr:`self._dtgram
<pcapkit.foundation.reassembly.reassembly.Reassembly._dtgram>`
(if not, datagram will be discarded after callback)
timeout: reassembly timeout in seconds, measured on the
*capture's* clock; :data:`None` selects the protocol's own
:attr:`__timeout__
<pcapkit.foundation.reassembly.reassembly.Reassembly.__timeout__>`
default
Note:
This class is for internal use only. For customisation, please use
:class:`Reassembly` instead.
"""
if TYPE_CHECKING:
#: Protocol name of current reassembly object.
__protocol_name__: 'str'
#: Protocol of current reassembly object.
__protocol_type__: 'Type[ProtocolBase]'
#: List of callback functions upon reassembled datagram.
__callback_fn__: 'list[CallbackFn]'
_flag_s: 'bool'
_flag_d: 'bool'
_flag_n: 'bool'
_timeout: 'float'
# Internal data storage for cached properties.
__cached__: 'dict[str, Any]'
##########################################################################
# Defaults.
##########################################################################
#: float: Default reassembly timeout, in seconds, for this protocol --
#: overridden per protocol, e.g.
#: :attr:`IPv6.__timeout__ <pcapkit.foundation.reassembly.ipv6.IPv6.__timeout__>`.
#: :data:`math.inf` means "never expire", which is the base default because
#: nothing here knows what a protocol's specification asks for.
__timeout__: 'float' = math.inf
##########################################################################
# Properties.
##########################################################################
@property
def timeout(self) -> 'float':
"""Reassembly timeout, in seconds, of the current reassembly object.
A buffer whose first-arriving fragment is older than this many seconds
**on the capture's own clock** is abandoned rather than held for the life
of the object -- see :meth:`expire`. :data:`math.inf` disables expiry.
"""
return self._timeout
@property
def name(self) -> 'str':
"""Protocol name of current reassembly object.
Note:
The same value is also available as a class attribute,
from the property of the same name on :class:`ReassemblyMeta`.
"""
if hasattr(self, '__protocol_name__'):
return self.__protocol_name__
return type(self).name # type: ignore[return-value]
@property
def protocol(self) -> 'Type[ProtocolBase]':
"""Protocol of current reassembly object.
Note:
The same value is also available as a class attribute,
from the property of the same name on :class:`ReassemblyMeta`.
"""
if hasattr(self, '__protocol_type__'):
return self.__protocol_type__
return type(self).protocol # type: ignore[return-value]
# total number of reassembled packets
@property
def count(self) -> 'int':
"""Total number of reassembled packets."""
if self._flag_n:
self.__cached__.clear()
self._flag_n = False
if (cached := self.__cached__.get('count')) is not None:
return cached
ret = len(self.datagram)
self.__cached__['count'] = ret
return ret
# reassembled datagram
@property
def datagram(self) -> 'tuple[_DT, ...]':
"""Reassembled datagram.
Raises:
UnsupportedCall: If :attr:`self._flag_d <_flag_d>` is
set to :data:`False`.
"""
if not self._flag_d:
raise UnsupportedCall(f'{self.__class__.__name__}(store=False) has no attribute "datagram"')
if self._buffer:
return self.fetch()
return tuple(self._dtgram)
##########################################################################
# Methods.
##########################################################################
# reassembly procedure
@abc.abstractmethod
def reassembly(self, info: '_PT') -> 'None':
"""Reassembly procedure.
Arguments:
info: :term:`info` record of packets to be reassembled
"""
# clear cache
self._flag_n = False
self.__cached__['count'] = None
self.__cached__['fetch'] = None
# submit reassembled payload
@abc.abstractmethod
def submit(self, buf: '_BT', **kwargs: 'Any') -> 'list[_DT]':
"""Submit reassembled payload.
Arguments:
buf: buffer dict of reassembled packets
**kwargs: arbitrary keyword arguments; implementations accept
``timeout``, set when the buffer is being submitted because
:meth:`expire` abandoned it rather than because it completed or
the capture ended
"""
# abandon timed-out buffers
def expire(self, timestamp: 'float') -> 'list[_DT]':
"""Abandon every buffer whose reassembly timeout has elapsed.
Arguments:
timestamp: Current time on the capture's clock, in seconds since the
Unix epoch -- i.e. the capture timestamp of the packet just
handed to :meth:`reassembly`.
Returns:
Datagrams of the buffers abandoned, reported with
:attr:`Completion.TIMEOUT <pcapkit.foundation.reassembly.data.data.Completion.TIMEOUT>`.
Empty when nothing expired, which is the overwhelmingly common case.
A buffer expires when more than :attr:`timeout` seconds separate
``timestamp`` from the capture timestamp of its **first-arriving**
fragment, which is the deadline :rfc:`8200#section-4.5` states ("within
60 seconds of the reception of the first-arriving fragment") and which
:rfc:`815` suggests implementing by reading "the clock when each first
fragment arrives". A later fragment therefore does not extend the
deadline.
Note:
The clock only advances when *this* reassembly object is fed, since
a packet handed to it is the only evidence an offline parser has that
capture time has moved on. For TCP that is every segment, for IPv4
every frame whose **DF** flag is clear, and for IPv6 only the
fragments, so an IPv6 buffer that stalls and is followed by no
further IPv6 fragment is reported as
:attr:`Completion.PARTIAL <pcapkit.foundation.reassembly.data.data.Completion.PARTIAL>`
at the end of the capture. That is the honest answer: the capture
never shows that the deadline passed. A caller with an outside
source of time may call this method itself to advance the clock.
"""
if math.isinf(self._timeout) or not self._buffer:
return []
# NOTE: compared against the buffer's own origin, not a per-packet countdown, so a
# replay depends only on the capture's timestamps and stays deterministic.
deadline = timestamp - self._timeout
expired = [
bufid for (bufid, buffer) in self._buffer.items()
if cast('Any', buffer).timestamp < deadline
]
ret = [] # type: list[_DT]
for bufid in expired:
buffer = self._buffer.pop(bufid)
# NOTE: :rfc:`8200#section-4.5` and :rfc:`1122#section-3.3.2` owe an ICMP Time Exceeded
# once the offset-zero fragment arrived; offline, it is reported instead.
owed_icmp = bool(getattr(buffer, 'header', None) or getattr(buffer, 'hdr', None))
logger.debug('%s: abandoning buffer %s after %.6fs > %.6fs timeout '
'(ICMP Time Exceeded owed: %s)', self.name, bufid,
timestamp - cast('Any', buffer).timestamp, self._timeout, owed_icmp)
ret.extend(self.submit(buffer, bufid=bufid, timeout=True))
return ret
# fetch datagram
def fetch(self) -> 'tuple[_DT, ...]':
"""Fetch datagram.
Returns:
Tuple of reassembled datagrams.
Fetch reassembled datagrams from
:attr:`self._dtgram <pcapkit.foundation.reassembly.reassembly.Reassembly._dtgram>`
and returns a *tuple* of such datagrams.
If no cache found, the method will call
:meth:`self.submit <pcapkit.foundation.reassembly.reassembly.Reassembly.submit>`
to *forcedly* obtain newly reassembled payload. Otherwise, the
already calculated
:attr:`self._dtgram <pcapkit.foundation.reassembly.reassembly.Reassembly._dtgram>`
will be returned.
"""
if self._flag_n:
self.__cached__.clear()
self._flag_n = False
if (cached := self.__cached__.get('fetch')) is not None:
return cached
logger.debug('%s: flushing %d outstanding buffer(s)', self.name, len(self._buffer))
temp_dtgram = [] # type: list[_DT]
for (bufid, buffer) in self._buffer.items():
temp_dtgram.extend(
self.submit(buffer, bufid=bufid)
)
temp_dtgram.extend(self._dtgram)
ret = tuple(temp_dtgram)
logger.debug('%s: fetched %d datagram(s)', self.name, len(ret))
self.__cached__['fetch'] = ret
return ret
# return datagram index
def index(self, pkt_num: 'int') -> 'Optional[int]':
"""Return datagram index.
Arguments:
pkt_num: index of packet
Returns:
Reassembled datagram index which was from No. ``pkt_num`` packet;
if not found, returns :obj:`None`.
"""
for counter, datagram in enumerate(self.datagram):
if pkt_num in datagram.index: # type: ignore[attr-defined]
return counter
return None
# run automatically
def run(self, packets: 'list[_PT]') -> 'None':
"""Run automatically.
Arguments:
packets: list of packet dicts to be reassembled
"""
logger.debug('%s: reassembling %d packet(s)', self.name, len(packets))
for packet in packets:
self.reassembly(packet)
# register callback function
@classmethod
def register(cls, callback: 'CallbackFn', *, index: 'Optional[int]' = None) -> 'None':
"""Register callback function.
Arguments:
callback: callback function, which will be called
when reassembled datagram is obtained, with the
list of reassembled datagrams as its only argument
index: index to be inserted in the callback list; by
default, the callback will be appended to the end
of the list
"""
if index is not None:
cls.__callback_fn__.insert(index, callback)
else:
cls.__callback_fn__.append(callback)
##########################################################################
# Data models.
##########################################################################
def __new__(cls, *args: 'Any', **kwargs: 'Any') -> 'Self': # pylint: disable=unused-argument
self = super().__new__(cls)
# NOTE: Assign this attribute after ``__new__`` to avoid shared memory
# reference between instances.
self.__cached__ = {}
return self
def __init__(self, *, strict: 'bool' = True, store: 'bool' = True,
timeout: 'Optional[float]' = None) -> 'None':
"""Initialise packet reassembly.
Args:
strict: if :data:`True`, report a datagram that is not completely
reassembled as the tuple of its received runs; otherwise as one
contiguous payload, its holes zero-filled -- or, for IP while the
total length is unknown, only the prefix up to the first hole
store: if store reassembled datagram in memory, i.e.,
:attr:`self._dtgram <_dtgram>` (if not, datagram
will be discarded after callback)
timeout: reassembly timeout in seconds, measured on the capture's
own clock rather than the host's; :data:`None` selects this
protocol's :attr:`__timeout__` default, and
:data:`math.inf` disables expiry entirely
Raises:
FieldValueError: If ``timeout`` is negative.
"""
#: bool: Strict mode flag. If set to :data:`True`, a datagram
#: not completely reassembled is reported as the tuple of its
#: received runs; otherwise, as one contiguous payload, its
#: holes zero-filled -- or, for IP while the total length is
#: unknown, only the prefix up to the first hole.
self._flag_s = strict
#: bool: Store mode flag. If set to :data:`True`, all
#: reassembled datagram will be stored in memory, i.e.,
#: :attr:`self._dtgram <_dtgram>`; otherwise, datagram
#: will be discarded after callback.
self._flag_d = store
#: bool: New datagram flag. If set to :data:`True`, the
#: :attr:`self._dtgram <_dtgram>` will be repopulated.
self._flag_n = False
if timeout is None:
timeout = self.__timeout__
elif timeout < 0:
raise FieldValueError(f'{type(self).__name__}: reassembly timeout must not be '
f'negative, got {timeout!r}')
#: float: Reassembly timeout in seconds, on the capture's clock.
#: :data:`math.inf` disables expiry.
self._timeout = float(timeout)
#: dict[_IT, _BT]: Dict buffer field. This field is used to
#: store reassembled packets in the form of ``{bufid: buffer}``.
self._buffer = {} # type: dict[_IT, _BT]
#: list[_DT]: List reassembled datagram. This list is used
#: to store reassembled datagrams.
self._dtgram = [] # type: list[_DT]
logger.debug('%s reassembly initialised (strict=%s, store=%s, timeout=%s)',
self.name, strict, store, self._timeout)
def __call__(self, packet: '_PT') -> 'None':
"""Call packet reassembly.
Arguments:
packet: packet dict to be reassembled
(detailed format described in corresponding protocol)
"""
self._flag_n = True
self.reassembly(packet)
def __init_subclass__(cls) -> 'None':
"""Initialise subclass.
This method is to be used for generating necessary attributes
for the :class:`Reassembly` class. It can be useful to reduce
unnecessary registry calls and simplify the customisation process.
"""
cls.__callback_fn__ = []
[docs]
class Reassembly(ReassemblyBase[_PT, _DT, _IT, _BT], Generic[_PT, _DT, _IT, _BT]):
"""Base reassembly class.
Example:
Registration is opt-in. Pass keyword argument ``protocol`` at class
definition to register the reassembly under that protocol name:
.. code-block:: python
class MyProtocol(Reassembly, protocol='my_protocol'):
...
Omit it and the subclass is *not* registered, which is how a class
that is not meant to be selectable by name declines:
.. code-block:: python
class MyMixin(Reassembly): # not registered
...
Such a class can still be registered later, on demand:
.. code-block:: python
Extractor.register_reassembly('my_mixin', MyMixin)
Arguments:
strict: if :data:`True`, report a datagram that is not completely
reassembled as the tuple of its received runs; otherwise as one
contiguous payload, its holes zero-filled -- or, for IP while the
total length is unknown, only the prefix up to the first hole
store: if store reassembled datagram in memory, i.e.,
:attr:`self._dtgram <_dtgram>` (if not, datagram
will be discarded after callback)
timeout: reassembly timeout in seconds, on the capture's own clock;
:data:`None` selects the protocol's :attr:`__timeout__` default
.. versionchanged:: 1.3.0
:attr:`name` and :attr:`protocol` are read from ``__protocol_name__``
and ``__protocol_type__``, where they were abstract properties each
reassembly implemented, and the built-in names are short: ``'IPv4'``
rather than ``'Internet Protocol version 4'``. A subclass registers
itself with :class:`~pcapkit.foundation.extraction.Extractor` at class
definition.
.. versionchanged:: 1.5.0
Registration is opt-in, under the ``protocol`` class keyword; 1.4.x
registered every subclass, under its :attr:`name` when the keyword was
omitted (:issue:`514`). The ``timeout`` argument is added: a buffer
older than it on the capture's clock is abandoned.
"""
[docs]
def __init_subclass__(cls, /, protocol: 'Optional[str]' = None, *args: 'Any', **kwargs: 'Any') -> 'None':
"""Initialise subclass.
This method is to be used for registering the reassembly class to
:class:`~pcapkit.foundation.extraction.Extractor` class.
Args:
protocol: Protocol name to register the subclass under, lowercased.
:data:`None` (the default) skips registration entirely.
*args: Arbitrary positional arguments.
**kwargs: Arbitrary keyword arguments.
Raises:
UnsupportedCall: If any unrecognised class keyword is given.
Registration is **opt-in**: the subclass is registered if and only if
``protocol`` is given. This is what lets a subclass decline registration
rather than having to inherit :class:`ReassemblyBase` to avoid it, and it
matches :meth:`EnumSchema.__init_subclass__
<pcapkit.protocols.schema.EnumSchema.__init_subclass__>`, which
guards on its own ``code`` keyword the same way.
Note:
:attr:`__protocol_name__` is *not* an opt-in. It supplies the
:attr:`name <pcapkit.foundation.reassembly.reassembly.Reassembly.name>`
the reassembly reports, which it does whether or not the class is
registered; only the keyword decides registration.
See Also:
For more details, please refer to
:meth:`pcapkit.foundation.extraction.Extractor.register_reassembly`.
"""
# NOTE: ``engine=`` or ``name=`` here is the expected mistake, and left in ``**kwargs``
# it would silently skip registration; see ``Engine.__init_subclass__``.
if args or kwargs:
unexpected = ', '.join([*map(repr, args), *sorted(kwargs)])
raise UnsupportedCall(f'{cls.__name__}: unexpected class keyword(s): {unexpected}')
if protocol is not None:
from pcapkit.foundation.extraction import \
Extractor # pylint: disable=import-outside-toplevel
Extractor.register_reassembly(protocol.lower(), cls)
return super().__init_subclass__()