# -*- coding: utf-8 -*-
# mypy: disable-error-code=dict-item
"""Frame Header
==================
.. module:: pcapkit.protocols.misc.pcap.frame
:mod:`pcapkit.protocols.misc.pcap.frame` contains
:class:`~pcapkit.protocols.misc.pcap.frame.Frame` only,
which implements an extractor for frame headers [*]_ of PCAP,
whose structure is described as below:
.. code-block:: c
typedef struct pcaprec_hdr_s {
guint32 ts_sec; /* timestamp seconds */
guint32 ts_usec; /* timestamp microseconds */
guint32 incl_len; /* number of octets of packet saved in file */
guint32 orig_len; /* actual length of packet */
} pcaprec_hdr_t;
.. [*] https://wiki.wireshark.org/Development/LibpcapFileFormat#Record_.28Packet.29_Header
.. versionchanged:: 1.5.0
``bytes(frame)`` and :attr:`~pcapkit.protocols.protocol.Protocol.packet`
hold this frame's own octets, where they held the next frame's
(:issue:`357`), and the parsed ``time`` is timezone-aware UTC. The parsed
``len`` is the on-wire length and ``cap_len`` the captured one, where the
two were swapped (:issue:`618`).
"""
import collections
import datetime
import decimal
import io
import sys
import time
from typing import TYPE_CHECKING, cast, overload
from pcapkit.const.reg.linktype import LinkType as Enum_LinkType
from pcapkit.corekit.io import advance
from pcapkit.corekit.module import ModuleDescriptor
from pcapkit.protocols.data.misc.pcap.frame import Frame as Data_Frame
from pcapkit.protocols.data.misc.pcap.frame import FrameInfo as Data_FrameInfo
from pcapkit.protocols.protocol import ProtocolBase
from pcapkit.protocols.schema.misc.pcap.frame import RECORD_HEADER_LEN
from pcapkit.protocols.schema.misc.pcap.frame import Frame as Schema_Frame
from pcapkit.protocols.schema.misc.pcap.header import RESOLUTION
from pcapkit.utilities.compat import localcontext
from pcapkit.utilities.exceptions import StreamEOFError, UnsupportedCall, stacklevel
from pcapkit.utilities.warnings import ProtocolWarning, RegistryWarning, warn
if TYPE_CHECKING:
from datetime import datetime as dt_type
from decimal import Decimal
from typing import IO, Any, DefaultDict, Optional, Type
from typing_extensions import Literal
from pcapkit.protocols.data.misc.pcap.header import Header as Data_Header
from pcapkit.protocols.schema.schema import Schema
__all__ = ['Frame']
# check Python version
py37 = ((version_info := sys.version_info).major >= 3 and version_info.minor >= 7)
[docs]
class Frame(ProtocolBase[Data_Frame, Schema_Frame],
schema=Schema_Frame, data=Data_Frame):
"""Per packet frame header extractor.
This class supports parsing of the following protocols, which are
registered in the :attr:`self.__proto__ <pcapkit.protocols.misc.pcap.frame.Frame.__proto__>`
attribute:
.. list-table::
:header-rows: 1
* - Index
- Protocol
* - :attr:`pcapkit.const.reg.linktype.LinkType.NULL`
- :class:`pcapkit.protocols.link.loopback.Loopback`
* - :attr:`pcapkit.const.reg.linktype.LinkType.ETHERNET`
- :class:`pcapkit.protocols.link.ethernet.Ethernet`
* - :attr:`pcapkit.const.reg.linktype.LinkType.LOOP`
- :class:`pcapkit.protocols.link.loopback.Loopback`
* - :attr:`pcapkit.const.reg.linktype.LinkType.IPV4`
- :class:`pcapkit.protocols.internet.ipv4.IPv4`
* - :attr:`pcapkit.const.reg.linktype.LinkType.IPV6`
- :class:`pcapkit.protocols.internet.ipv6.IPv6`
:attr:`pcapkit.const.reg.linktype.LinkType.RAW` has no entry of its own:
its data is dispatched as ``IPV4`` or ``IPV6`` by the version nibble of
its first octet, see :meth:`_dispatch_linktype`.
"""
##########################################################################
# Defaults.
##########################################################################
#: Protocol index mapping for decoding the next layer; see
#: :meth:`self._decode_next_layer <pcapkit.protocols.protocol.Protocol._decode_next_layer>`
#: and :meth:`self._import_next_layer <pcapkit.protocols.protocol.Protocol._import_next_layer>`.
#: Each value is a :class:`~pcapkit.corekit.module.ModuleDescriptor` naming the
#: module and class, or a :class:`~pcapkit.protocols.protocol.Protocol` subclass.
__proto__ = collections.defaultdict(
lambda: ModuleDescriptor('pcapkit.protocols.misc.raw', 'Raw'),
{
Enum_LinkType.NULL: ModuleDescriptor('pcapkit.protocols.link', 'Loopback'),
Enum_LinkType.ETHERNET: ModuleDescriptor('pcapkit.protocols.link', 'Ethernet'),
Enum_LinkType.LOOP: ModuleDescriptor('pcapkit.protocols.link', 'Loopback'),
Enum_LinkType.IPV4: ModuleDescriptor('pcapkit.protocols.internet', 'IPv4'),
Enum_LinkType.IPV6: ModuleDescriptor('pcapkit.protocols.internet', 'IPv6'),
},
) # type: DefaultDict[Enum_LinkType | int, ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]]
##########################################################################
# Properties.
##########################################################################
@property
def name(self) -> 'str':
"""Name of corresponding protocol."""
return f'Frame {self._fnum}'
@property
def length(self) -> 'Literal[16]':
"""Header length of corresponding protocol."""
return RECORD_HEADER_LEN # type: ignore[return-value]
@property
def header(self) -> 'Data_Header':
"""Global header of the PCAP file."""
return self._ghdr
##########################################################################
# Methods.
##########################################################################
[docs]
@classmethod
def register(cls, code: 'Enum_LinkType', protocol: 'ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]') -> 'None': # type: ignore[override]
r"""Register a new protocol class.
Notes:
The fully qualified class name of the new protocol class
should be ``{protocol.module}.{protocol.name}``.
Arguments:
code: protocol code as in :class:`~pcapkit.const.reg.linktype.LinkType`
protocol: module descriptor or a
:class:`~pcapkit.protocols.protocol.Protocol` subclass
Raises:
pcapkit.utilities.exceptions.RegistryError: If ``protocol`` is not a
class, or not a :class:`~pcapkit.protocols.protocol.Protocol` subclass.
Warns:
pcapkit.utilities.warnings.RegistryWarning: If this link type is
already registered against PCAP frames, naming the displaced
entry and its replacement so a caller can tell *what* was lost.
Fires only when the incumbent differs from the replacement --
see :meth:`ProtocolBase.register
<pcapkit.protocols.protocol.ProtocolBase.register>` for the
guard this shares with ``register_protocol``, and for the
shipped descriptor a restore puts back (:issue:`1504`).
"""
incumbent = cls.__proto__.get(code)
entry = cls._next_layer_entry(code, protocol)
if entry is None:
return
if incumbent is not None and incumbent is not entry:
warn(f'protocol {code} already registered, overwriting '
f'{incumbent!r} with {entry!r}', RegistryWarning)
cls.__proto__[code] = entry
[docs]
def index(self, name: 'str | ProtocolBase | Type[ProtocolBase]') -> 'int':
"""Call :meth:`ProtoChain.index <pcapkit.corekit.protochain.ProtoChain.index>`.
Args:
name: ``name`` to be searched
Returns:
First index of ``name``.
Raises:
IndexNotFound: if ``name`` is not present
"""
return self._protos.index(name)
[docs]
def pack(self, **kwargs: 'Any') -> 'bytes':
"""Pack (construct) packet data.
Args:
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed packet data.
Notes:
We used a special keyword argument ``__packet__`` to pass the
global packet data to underlying methods. This is useful when
the packet data is not available in the current instance.
As in :meth:`Protocol.pack <pcapkit.protocols.protocol.Protocol.pack>`,
the header is made on a copy outside construction, so the instance
keeps its own (:issue:`1769`).
"""
# pylint: disable=protected-access
target = self._packing_target()
target.__header__ = target.make(**kwargs)
packet = kwargs.get('__packet__', {}) # packet data
packet['byteorder'] = target._ghdr.magic_number.byteorder
return target.__header__.pack(packet)
[docs]
def unpack(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_Frame':
r"""Unpack (parse) packet data.
Args:
length: Length of packet data.
\_seek_set (int): File offset before reading, forwarded to
:meth:`self.read <read>` through ``**kwargs``.
**kwargs: Arbitrary keyword arguments.
Returns:
Parsed packet data.
Notes:
We used a special keyword argument ``__packet__`` to pass the
global packet data to underlying methods. This is useful when
the packet data is not available in the current instance.
"""
if cast('Optional[Schema_Frame]', self.__header__) is None:
packet = kwargs.get('__packet__', {}) # packet data
# NOTE: Reached on the parsing path only; the construction path
# sets ``__header__`` before unpacking its own buffer.
self._check_record_floor(length)
packet['byteorder'] = self._ghdr.magic_number.byteorder
self.__header__ = cast('Schema_Frame', self.__schema__.unpack(self._file, length, packet)) # type: ignore[call-arg,misc]
return self.read(length, **kwargs)
[docs]
def read(self, length: 'Optional[int]' = None, *, _read: 'bool' = True,
_seek_set: 'int' = 0, **kwargs: 'Any') -> 'Data_Frame':
r"""Read each block after global header.
Args:
length: Length of data to be read.
\_read: If the class is called in a parsing scenario.
\_seek_set: File offset of the record's own first octet, i.e. where
the stream stood before :meth:`self.unpack <unpack>` consumed
the record. It cannot be recovered from the stream here, since
that unpack has consumed the whole record, packet data included.
**kwargs: Arbitrary keyword arguments.
Returns:
Data_Frame: Parsed packet data.
Warns:
pcapkit.utilities.warnings.ProtocolWarning: If the timestamp is out
of range, or if the record's ``incl_len`` runs past the end of
the data being parsed, in which case the record is kept as
captured.
"""
schema = self.__header__
_tsss = schema.ts_sec
_tsus = schema.ts_usec
_ilen = schema.incl_len
_olen = schema.orig_len
with localcontext(prec=64):
_epch = _tsss + decimal.Decimal(_tsus) / RESOLUTION[bool(self._nsec)]
_irat = _epch.as_integer_ratio()
try:
# NOTE: Aware and anchored to UTC, not naive. ``ts_sec`` is an offset
# from the UNIX epoch, so the instant it names must not depend on
# where the file is read; a bare ``fromtimestamp`` renders it in the
# reading host's zone, drops the offset, and cannot be compared
# against an aware datetime. The ``except`` branch below returns an
# aware UTC datetime too, and so does the PCAP-NG reader, which fills
# the same
# :attr:`Data_Frame.time <pcapkit.protocols.data.misc.pcap.frame.Frame.time>`
# field through :func:`pcapkit.toolkit.pcapng.block2frame`.
_time = datetime.datetime.fromtimestamp(_irat[0] / _irat[1], datetime.timezone.utc)
except ValueError:
warn(f'PCAP: invalid timestamp: {_epch}', ProtocolWarning, stacklevel=stacklevel())
_time = datetime.datetime.fromtimestamp(0, datetime.timezone.utc)
frame = Data_Frame(
frame_info=Data_FrameInfo(
ts_sec=_tsss,
ts_usec=_tsus,
incl_len=_ilen,
orig_len=_olen,
),
time=_time,
number=self._fnum,
time_epoch=_epch,
# NOTE: ``len`` is the on-wire length (``orig_len``) and ``cap_len``
# the captured one (``incl_len``). The two differ only for a frame
# the snapshot length cut short.
#
# The names are Wireshark's: its ``packet-frame.c`` registers
# ``frame.len`` as "Frame length on the wire" and ``frame.cap_len``
# as "Frame length stored into the capture file", and raises
# ``frame.len_lt_caplen`` (``PI_MALFORMED``) on ``frame_len <
# cap_len``, which could not be malformed if ``len`` were the smaller,
# captured one. The PCAP-NG reader uses the same convention.
len=_olen,
cap_len=_ilen,
)
if not _read:
# move backward to the beginning of the packet
self._file.seek(0, io.SEEK_SET)
else:
# NOTE: We create a copy of the frame data here for parsing
# scenarios to keep the original frame data intact.
#
# NOTE: The record spans ``self.length + incl_len`` octets, and
# :meth:`self.unpack <unpack>` has consumed *all* of them by now --
# the schema's payload field carries the ``incl_len`` octets of
# packet data, not just the 16-octet record header. Seeking back
# over ``self.length`` alone would land ``incl_len`` octets too late,
# on the tail of this record followed by the head of the next one.
# So the start is recorded before the unpack, in
# :meth:`self.__post_init__ <__post_init__>`, and seeked to
# absolutely, as the PCAP-NG reader does; see
# :meth:`pcapkit.protocols.misc.pcapng.PCAPNG.read`.
seek_cur = _seek_set + self.length + _ilen
# move backward to the beginning of the frame
self._file.seek(_seek_set, io.SEEK_SET)
#: bytes: Raw frame data.
#
# NOTE: Asked for no further than one read can go, :data:`sys.maxsize`.
# Past it the read raises :exc:`OverflowError` instead of coming back
# short, so a record claiming more than that would never reach the
# overrun handling below (#1975).
bound = min(self.length + _ilen, sys.maxsize)
self._data = self._read_fileng(bound)
# NOTE: A record whose ``incl_len`` runs past the end of the data is
# kept as captured, as the PCAP-NG reader keeps a captured length
# running past its block (#1405): ``incl_len`` and ``cap_len``
# stay as declared, so the record rebuilds byte-exactly, and the
# packet is the octets present. A writer must not copy the declared
# length; see :meth:`PCAPIO._append_value
# <pcapkit.dumpkit.pcap.PCAPIO._append_value>` (#1541).
held = len(self._data) - self.length
if held < _ilen:
warn(f'PCAP: [Frame {self._fnum}] captured length {_ilen} runs past the data, '
f'which holds {held} octet(s); kept as captured',
ProtocolWarning, stacklevel=stacklevel())
# NOTE: A read that came back whole at the bound stopped there, not
# at the end of the data: the record runs on past it, and the next
# one starts where this one declares it ends, so the stream is
# moved over the rest (#1975). A shorter read met the end of the
# data, where a seek past :data:`sys.maxsize` could not go either,
# so the next frame then starts where the octets read end.
if len(self._data) == bound < self.length + _ilen:
advance(self._file, self.length + _ilen - bound)
else:
if seek_cur > sys.maxsize:
seek_cur = _seek_set + len(self._data)
# move forward to the beginning of the next frame
self._file.seek(seek_cur, io.SEEK_SET)
#: io.BytesIO: Source data stream.
self._file = io.BytesIO(self._data)
# NOTE: The dissector is handed the octets actually *present*, i.e.
# ``cap_len`` (``incl_len``), never the on-wire ``len``. For a frame the
# snapshot length cut short the latter exceeds what the file holds, and
# handing it over is the declared-length-exceeds-available-octets fault.
#
# No test pins this line: reverting just this argument to ``frame.len``
# leaves the whole suite green. ``big_endian.pcap``'s truncated frame
# dissects to ``Ethernet:IPv4:UDP:Raw``, and :meth:`Raw.read` ignores the
# ``length`` it is handed, so the over-long value never reaches anything
# that checks it. Catching a regression needs a fixture whose truncation
# lands in a length-checked field instead of bottoming out in ``Raw``.
return self._decode_next_layer(frame, self._ghdr.network, frame.cap_len)
[docs]
def make(self,
timestamp: 'Optional[float | Decimal | int | dt_type]' = None,
ts_sec: 'Optional[int]' = None,
ts_usec: 'Optional[int]' = None,
incl_len: 'Optional[int]' = None,
orig_len: 'Optional[int]' = None,
packet: 'bytes | ProtocolBase | Schema' = b'',
nanosecond: 'bool' = False,
**kwargs: 'Any') -> 'Schema_Frame':
"""Make frame packet data.
Args:
timestamp: UNIX-Epoch timestamp
ts_sec: timestamp seconds
ts_usec: timestamp microseconds
incl_len: number of octets of packet saved in file
orig_len: actual length of packet
packet: raw packet data
nanosecond: nanosecond-resolution file flag
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed packet data.
"""
ts_sec, ts_usec = self._make_timestamp(timestamp, ts_sec, ts_usec, nanosecond)
if incl_len is None:
# NOTE: A snaplen of 0 is no limit, not a limit of zero octets. The
# PCAP draft says the field "MUST NOT be zero" (draft-ietf-opsawg-pcap,
# section 4), and libpcap reads such a bogus value as the maximum for
# the link type (``pcapint_adjust_snapshot`` in ``savefile.c``); the
# PCAP-NG reader treats a zero ``SnapLen`` the same way.
snaplen = self._ghdr.snaplen
incl_len = min(len(packet), snaplen) if snaplen else len(packet)
if orig_len is None:
orig_len = len(packet)
# NOTE: The record holds exactly ``incl_len`` octets of packet data, so a
# packet longer than that -- e.g. than the snaplen -- is cut to it rather
# than written out whole behind a clipped length.
if len(packet) > incl_len:
packet = bytes(packet)[:incl_len]
return Schema_Frame(
ts_sec=ts_sec,
ts_usec=ts_usec,
incl_len=incl_len,
orig_len=orig_len,
packet=packet,
)
##########################################################################
# Data models.
##########################################################################
@overload # pragma: no cover
def __post_init__(self, file: 'IO[bytes] | bytes', length: 'Optional[int]' = ..., *, # pylint: disable=arguments-differ
num: 'int', header: 'Data_Header', **kwargs: 'Any') -> 'None': ...
@overload # pragma: no cover
def __post_init__(self, *, num: 'int', header: 'Data_Header', # pylint: disable=arguments-differ
**kwargs: 'Any') -> 'None': ...
[docs]
def __post_init__(self, file: 'Optional[IO[bytes] | bytes]' = None, length: 'Optional[int]' = None, *, # pylint: disable=arguments-differ
num: 'int', header: 'Data_Header', **kwargs: 'Any') -> 'None':
"""Initialisation.
Args:
file: Source packet stream.
length: Length of packet data.
num: Frame index number.
header: Global header of the PCAP file.
**kwargs: Arbitrary keyword arguments.
See Also:
For construction arguments, please refer to :meth:`make`.
"""
#: int: frame index number
self._fnum = num
#: pcapkit.protocols.misc.pcap.header.Header: Global header of the PCAP file.
self._ghdr = header
#: pcapkit.const.reg.linktype.LinkType: next layer protocol index
self._prot = header.network
#: bool: nanosecond-timestamp PCAP flag
self._nsec = header.magic_number.nanosecond
if file is None:
_read = False
#: bytes: Raw packet data.
self._data = self.pack(**kwargs)
#: io.BytesIO: Source packet stream.
self._file = io.BytesIO(self._data)
else:
_read = True
#: io.BytesIO: Source packet stream.
self._file = io.BytesIO(file) if isinstance(file, bytes) else file
# NOTE: Taken *before* the unpack, which advances the stream past the
# whole record; :meth:`self.read <read>` needs the record's own start and
# cannot work it out afterwards. See the note there.
_seek_set = self._file.tell()
#: pcapkit.corekit.infoclass.Info: Parsed packet data.
self._info = self.unpack(length, _read=_read, _seek_set=_seek_set, **kwargs)
def __length_hint__(self) -> 'Literal[16]':
"""Return an estimated length for the object."""
return RECORD_HEADER_LEN # type: ignore[return-value]
def _check_record_floor(self, length: 'Optional[int]') -> 'None':
"""Refuse a tail too short to hold a record header.
Args:
length: Length of packet data, as the caller declared it, or
:obj:`None` to measure what is left in
:attr:`self._file <pcapkit.protocols.protocol.Protocol._file>`.
Raises:
StreamEOFError: If fewer than sixteen octets are left to read.
A PCAP record header is sixteen octets, which is what
:meth:`__length_hint__` reports. A shorter tail is not a record, so
parsing one would only invent fields from zero padding. Reporting the
end of the stream instead lets :meth:`Extractor.record_frames
<pcapkit.foundation.extraction.Extractor.record_frames>` stop with the
frames read so far, as the PCAP-NG reader does for a truncated block
(:issue:`678`). See :issue:`1265`.
"""
if length is None:
current = self._file.tell()
length = self._file.seek(0, io.SEEK_END) - current
self._file.seek(current, io.SEEK_SET)
if length < RECORD_HEADER_LEN:
raise StreamEOFError(f'PCAP: record truncated: {length} octet(s) left, '
f'fewer than the {RECORD_HEADER_LEN} a record header needs',
quiet=True)
# NOTE: This is a hack to make the ``__index__`` method work both as a
# class method and an instance method.
[docs]
def __index__(self: 'Optional[Frame]' = None) -> 'int': # type: ignore[override]
"""Index of the frame.
Args:
self: :class:`Frame` object or :obj:`None`.
Returns:
If the object is initialised, i.e. ``self._fnum`` exists, returns the
frame index number of itself; else raises
:exc:`~pcapkit.utilities.exceptions.UnsupportedCall`.
Raises:
UnsupportedCall: If called on the class rather than on a frame.
"""
if self is None:
raise UnsupportedCall("'Frame' object cannot be interpreted as an integer")
return self._fnum
##########################################################################
# Utilities.
##########################################################################
[docs]
@classmethod
def _make_data(cls, data: 'Data_Frame') -> '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.
Notes:
The frame index and the global header are context of the PCAP file
rather than fields of the frame, so they are not returned here: pass
them to :meth:`~pcapkit.protocols.protocol.ProtocolBase.from_data` as
``num`` and ``header``.
"""
return {
'ts_sec': data.frame_info.ts_sec,
'ts_usec': data.frame_info.ts_usec,
'incl_len': data.frame_info.incl_len,
'orig_len': data.frame_info.orig_len,
'packet': cls._make_payload(data),
}
def _make_timestamp(self, timestamp: 'Optional[float | Decimal | dt_type | int]' = None, ts_sec: 'Optional[int]' = None,
ts_usec: 'Optional[int]' = None, nanosecond: 'bool' = False) -> 'tuple[int, int]':
"""Make timestamp.
Args:
timestamp: UNIX-Epoch timestamp
ts_sec: timestamp seconds
ts_usec: timestamp microseconds
nanosecond: nanosecond-resolution file flag
Returns:
Second and microsecond/nanosecond value of timestamp.
"""
with localcontext(prec=64):
if timestamp is None:
if py37 and nanosecond:
timestamp = decimal.Decimal(time.time_ns()) / 1_000_000_000
else:
timestamp = decimal.Decimal(time.time())
else:
if isinstance(timestamp, datetime.datetime):
timestamp = timestamp.timestamp()
timestamp = decimal.Decimal(timestamp)
if ts_sec is None:
ts_sec = int(timestamp)
if ts_usec is None:
ts_usec = int((timestamp - ts_sec) * RESOLUTION[bool(nanosecond)])
return ts_sec, ts_usec
def _decode_next_layer(self, dict_: 'Data_Frame', proto: 'Optional[int]' = None,
length: 'Optional[int]' = None, *, packet: 'Optional[dict[str, Any]]' = None) -> 'Data_Frame': # pylint: disable=arguments-differ
r"""Decode next layer protocol.
Arguments:
dict\_: info buffer
proto: next layer protocol index
length: valid (*non-padding*) length
packet: packet info (passed from :meth:`self.unpack <pcapkit.protocols.protocol.Protocol.unpack>`)
Returns:
Current protocol with packet extracted.
Notes:
We added a new key ``__next_type__`` to ``dict_`` to store the
next layer protocol type, and a new key ``__next_name__`` to
store the next layer protocol name. :meth:`Info.to_dict
<pcapkit.corekit.infoclass.Info.to_dict>` carries both, and the
dumpers leave them out.
We also added a new key ``protocols`` to ``dict_`` to store the
protocol chain of the current packet (frame).
``proto`` is dispatched as :meth:`_dispatch_linktype` says, so
``LINKTYPE_RAW`` data is read as IPv4 or IPv6 (:issue:`1502`).
"""
proto = self._dispatch_linktype(proto, length)
next_ = cast('ProtocolBase', self._import_next_layer(proto, length, packet=packet)) # type: ignore[misc,call-arg,redundant-cast]
info, chain = next_.info, next_.protochain
# make next layer protocol name
layer = next_.info_name
# proto = next_.__class__.__name__
# write info and protocol chain into dict
dict_.__update__({
layer: info,
'protocols': chain.chain if chain else '',
'__next_type__': type(next_),
'__next_name__': layer,
})
self._next = next_ # pylint: disable=attribute-defined-outside-init
self._protos = chain # pylint: disable=attribute-defined-outside-init
return dict_
[docs]
def _dispatch_linktype(self, proto: 'Optional[int]', length: 'Optional[int]' = None) -> 'Optional[int]':
"""Link type a frame's data is dispatched under.
Arguments:
proto: link type of the capture
length: valid (*non-padding*) length
Returns:
For :attr:`LinkType.RAW <pcapkit.const.reg.linktype.LinkType.RAW>`
data whose first octet carries IP version 4 or 6,
:attr:`LinkType.IPV4 <pcapkit.const.reg.linktype.LinkType.IPV4>` or
:attr:`LinkType.IPV6 <pcapkit.const.reg.linktype.LinkType.IPV6>`;
otherwise ``proto``.
Note:
``LINKTYPE_RAW`` (101) carries a bare IPv4 or IPv6 datagram, and
only the version nibble of its first octet tells which, so it has no
entry of its own in :attr:`__proto__` (:issue:`1502`). Dispatched as
``IPV4`` or ``IPV6``, it reaches whatever those link types are
registered against. An entry registered under ``RAW`` itself is used
as it stands, and data of any other version is kept as
:class:`~pcapkit.protocols.misc.raw.Raw`.
:meth:`PCAPNG._dispatch_linktype
<pcapkit.protocols.misc.pcapng.PCAPNG._dispatch_linktype>` does the
same for PCAP-NG.
"""
if proto != Enum_LinkType.RAW or Enum_LinkType.RAW in self.__proto__:
return proto
data = self._get_payload()[:length]
version = data[0] >> 4 if data else None
if version == 4:
return Enum_LinkType.IPV4
if version == 6:
return Enum_LinkType.IPV6
return proto