# -*- coding: utf-8 -*-
"""DPKT Support
==================
.. module:: pcapkit.foundation.engines.dpkt
This module contains the implementation for `DPKT`_ engine
support, as is used by :class:`pcapkit.foundation.extraction.Extractor`.
"""
import decimal
import functools
from typing import TYPE_CHECKING, cast
from pcapkit.const.pcapng.block_type import BlockType as Enum_BlockType
from pcapkit.const.pcapng.option_type import OptionType as Enum_OptionType
from pcapkit.const.reg.linktype import LinkType as Enum_LinkType
from pcapkit.foundation.engines.engine import EngineBase
from pcapkit.foundation.reassembly import ReassemblyManager
from pcapkit.protocols.misc.pcap.header.schema import RESOLUTION
from pcapkit.protocols.misc.pcapng.schema import (BYTE_ORDER_MAGIC, DEFAULT_RESOLUTION,
GENERIC_BLOCK_MINIMUM)
from pcapkit.utilities.compat import localcontext
from pcapkit.utilities.exceptions import FormatError, stacklevel
from pcapkit.utilities.logging import get_logger
from pcapkit.utilities.warnings import AttributeWarning, DPKTWarning, warn
__all__ = ['DPKT', 'PCAPNGReader']
if TYPE_CHECKING:
from typing import IO, Iterator, Literal, Optional, Protocol, Type, Union
from dpkt.dpkt import Packet as DPKTPacket
from dpkt.ethernet import Ethernet as DPKTEthernet
from dpkt.ip import IP as DPKTIP
from dpkt.ip6 import IP6 as DPKTIP6
from dpkt.loopback import Loopback as DPKTLoopback
from dpkt.pcap import Reader as PCAPReader
from pcapkit.foundation.engines.pcapng import Context
from pcapkit.foundation.extraction import Extractor
from pcapkit.protocols.misc.pcapng.data import IF_TSOffsetOption as Data_IF_TSOffsetOption
from pcapkit.protocols.misc.pcapng.data import IF_TSResolOption as Data_IF_TSResolOption
from pcapkit.protocols.misc.pcapng.data import \
InterfaceDescriptionBlock as Data_InterfaceDescriptionBlock
from pcapkit.protocols.misc.pcapng.data import SectionHeaderBlock as Data_SectionHeaderBlock
from pcapkit.protocols.misc.pcapng.data import SimplePacketBlock as Data_SimplePacketBlock
Reader = Union[PCAPReader, 'PCAPNGReader']
class _DPKTCoreModule(Protocol):
""":mod:`dpkt.dpkt`, as :class:`DPKT` reads it."""
@property
def Packet(self) -> 'Type[DPKTPacket]': ...
class _DPKTEthernetModule(Protocol):
""":mod:`dpkt.ethernet`, as :class:`DPKT` reads it."""
@property
def Ethernet(self) -> 'Type[DPKTEthernet]': ...
class _DPKTIPModule(Protocol):
""":mod:`dpkt.ip`, as :class:`DPKT` reads it."""
@property
def IP(self) -> 'Type[DPKTIP]': ...
class _DPKTIP6Module(Protocol):
""":mod:`dpkt.ip6`, as :class:`DPKT` reads it."""
@property
def IP6(self) -> 'Type[DPKTIP6]': ...
class _DPKTLoopbackModule(Protocol):
""":mod:`dpkt.loopback`, as :class:`DPKT` reads it."""
@property
def Loopback(self) -> 'Type[DPKTLoopback]': ...
class _DPKTPackage(Protocol):
"""|dpkt|_, as :class:`DPKT` reads it through :attr:`DPKT._expkg` (:issue:`1878`).
Each attribute is a submodule, described by what the engine reads of it.
mypy matches a module with a protocol at the top level only, so the
submodules are checked through the stub types they name, under
:file:`typings/dpkt`.
"""
@property
def dpkt(self) -> '_DPKTCoreModule': ...
@property
def ethernet(self) -> '_DPKTEthernetModule': ...
@property
def ip(self) -> '_DPKTIPModule': ...
@property
def ip6(self) -> '_DPKTIP6Module': ...
@property
def loopback(self) -> '_DPKTLoopbackModule': ...
#: logging.Logger: Module-level logger, a child of the package-wide
#: :data:`pcapkit.utilities.logging.logger`.
logger = get_logger(__name__)
#: The Block Type of a Section Header Block as the octets read, the same in
#: either byte order, spelled once rather than per block read.
_SHB_TYPE_OCTETS = Enum_BlockType.Section_Header_Block.to_bytes(4, 'big')
#: :data:`~pcapkit.protocols.misc.pcapng.schema.BYTE_ORDER_MAGIC` as a
#: little-endian section writes it, spelled once rather than per block read.
_LITTLE_BYTE_ORDER_OCTETS = BYTE_ORDER_MAGIC.to_bytes(4, 'little')
[docs]
class PCAPNGReader:
"""PCAP-NG reader for the DPKT engine, aware of every interface.
``dpkt.pcapng.Reader`` reads only the first Interface Description
Block (IDB) of a file, and applies its link type and timestamp resolution
to every packet (:issue:`1379`). This reader walks the blocks itself
instead: it parses each Section Header Block and IDB with :mod:`pcapkit`'s
own :class:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG`, keeps the interfaces
of the current section, and resolves every packet's link type,
``if_tsresol`` and ``if_tsoffset`` from the interface it names. Packet
blocks are still decoded by `DPKT`_.
It yields Enhanced Packet Blocks, (obsolete) Packet Blocks and Simple Packet
Blocks, and skips every other block. ``dpkt.pcapng.Reader`` skips the
Simple Packet Block too, and `DPKT`_ has no type for one, so that block is
parsed with :class:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG` as well
(:issue:`1520`).
Args:
file: Source PCAP-NG stream, positioned at its first Section Header Block.
"""
def __init__(self, file: 'IO[bytes]') -> 'None':
#: Source PCAP-NG stream.
self._file = file
#: Context of the current section.
self._ctx = cast('Context', None)
#: Section index number.
self._sect = 0
#: Link type of the packet last yielded.
self._linktype = cast('Enum_LinkType', Enum_LinkType.NULL)
#: Underlying block iterator.
self._iter = self._read()
[docs]
def datalink(self) -> 'Enum_LinkType':
"""Link type of the interface the packet last yielded was captured on."""
return self._linktype
def __iter__(self) -> 'PCAPNGReader':
return self
def __next__(self) -> 'tuple[float, bytes]':
return next(self._iter)
def _read(self) -> 'Iterator[tuple[float, bytes]]':
"""Read blocks until the next packet block, and yield its timestamp and data."""
from pcapkit.foundation.engines.pcapng import Context
from pcapkit.protocols.misc.pcapng.protocol import PCAPNG as P_PCAPNG
while True:
head = self._file.read(8)
if len(head) < 8:
return
# the SHB block type is a palindrome, so it reads the same in
# either byte order; any other block uses the section's order
byteorder: 'Literal["big", "little"]'
if head[:4] == _SHB_TYPE_OCTETS:
head += self._file.read(4)
byteorder = 'little' if head[8:] == _LITTLE_BYTE_ORDER_OCTETS else 'big'
else:
byteorder = self._ctx.section.byteorder
block_type = int.from_bytes(head[:4], byteorder)
length = int.from_bytes(head[4:8], byteorder)
if length < GENERIC_BLOCK_MINIMUM:
raise FormatError(f'PCAP-NG: [Block {block_type}] invalid block length: {length}')
buf = head + self._file.read(length - len(head))
if len(buf) < length:
return
if block_type == Enum_BlockType.Section_Header_Block:
self._sect += 1
shb = P_PCAPNG(buf, num=0, sct=self._sect, ctx=None)
self._ctx = Context(cast('Data_SectionHeaderBlock', shb.info))
elif block_type == Enum_BlockType.Interface_Description_Block:
idb = P_PCAPNG(buf, num=0, sct=self._sect, ctx=self._ctx)
self._ctx.interfaces.append(cast('Data_InterfaceDescriptionBlock', idb.info))
elif block_type in (Enum_BlockType.Enhanced_Packet_Block, Enum_BlockType.Packet_Block):
yield self._read_packet(buf, block_type, byteorder == 'little')
elif block_type == Enum_BlockType.Simple_Packet_Block:
yield self._read_simple(buf)
def _read_packet(self, buf: 'bytes', block_type: 'int',
little: 'bool') -> 'tuple[float, bytes]':
"""Decode a packet block with `DPKT`_, against the interface it names.
Args:
buf: Whole block, as read from the file.
block_type: Block type, an EPB or an (obsolete) Packet Block.
little: Whether the enclosing section is little-endian.
Returns:
Timestamp in seconds since the UNIX epoch, and the packet data. The
timestamp is a :class:`~pcapkit.toolkit.dpkt.Timestamp`: a
:obj:`float` that also carries the exact value, the interface's
resolution and the packet's original length,
for the PCAP frame record of the packet (:issue:`1507`).
Raises:
FormatError: If the section describes no interface of the ID the
block names.
"""
import dpkt # isort:skip
from pcapkit.toolkit.dpkt import Timestamp # isort:skip
if block_type == Enum_BlockType.Enhanced_Packet_Block:
tag = 'EPB'
pkt = (dpkt.pcapng.EnhancedPacketBlockLE(buf) if little
else dpkt.pcapng.EnhancedPacketBlock(buf))
else:
tag = 'Packet'
pkt = dpkt.pcapng.PacketBlockLE(buf) if little else dpkt.pcapng.PacketBlock(buf)
iface_id = pkt.iface_id # pylint: disable=no-member
if iface_id >= len(self._ctx.interfaces):
raise FormatError(f'PCAP-NG: [{tag}] invalid interface ID: {iface_id}')
interface = self._ctx.interfaces[iface_id]
tsresol = cast('Optional[Data_IF_TSResolOption]',
interface.options.get(Enum_OptionType.if_tsresol))
tsoffset = cast('Optional[Data_IF_TSOffsetOption]',
interface.options.get(Enum_OptionType.if_tsoffset))
resolution = DEFAULT_RESOLUTION if tsresol is None else tsresol.resolution
offset = 0 if tsoffset is None else tsoffset.offset
# same arithmetic as :meth:`PCAPNG._read_timestamp
# <pcapkit.protocols.misc.pcapng.protocol.PCAPNG._read_timestamp>`
ticks = (pkt.ts_high << 32) | pkt.ts_low
with localcontext(prec=64):
timestamp = decimal.Decimal(ticks) / resolution + offset
self._linktype = interface.linktype
return (Timestamp(float(timestamp), timestamp, resolution, pkt.pkt_len), # pylint: disable=no-member
pkt.pkt_data)
def _read_simple(self, buf: 'bytes') -> 'tuple[float, bytes]':
"""Decode a Simple Packet Block, against the section's first interface.
`DPKT`_ has no type for this block, so it is parsed by
:class:`~pcapkit.protocols.misc.pcapng.protocol.PCAPNG`: the packet data is
``min(original_len, snaplen)`` octets, with the snaplen of interface 0.
Args:
buf: Whole block, as read from the file.
Returns:
Timestamp and packet data, as :meth:`_read_packet` returns them. A
Simple Packet Block carries no timestamp, so the packet is dated at
the UNIX epoch; the original length is the block's own.
Raises:
FormatError: If the section describes no interface. A Simple Packet
Block names none, so it refers to the first (Section 4.4 of the
PCAP-NG specification).
"""
from pcapkit.protocols.misc.pcapng.protocol import PCAPNG as P_PCAPNG
from pcapkit.toolkit.dpkt import Timestamp # isort:skip
if not self._ctx.interfaces:
raise FormatError('PCAP-NG: [SPB] section has no interface description block')
interface = self._ctx.interfaces[0]
block = P_PCAPNG(buf, num=0, sct=self._sect, ctx=self._ctx)
spb = cast('Data_SimplePacketBlock', block.info)
tsresol = cast('Optional[Data_IF_TSResolOption]',
interface.options.get(Enum_OptionType.if_tsresol))
resolution = DEFAULT_RESOLUTION if tsresol is None else tsresol.resolution
self._linktype = interface.linktype
return (Timestamp(0.0, decimal.Decimal(0), resolution, spb.original_len),
bytes(spb.packet))
def _pcap_reader() -> 'Type[PCAPReader]':
"""A :class:`dpkt.pcap.Reader` that keeps each record's original length.
:class:`dpkt.pcap.Reader` parses every record header and then yields only
``(timestamp, octets)``, so the record's original length and the file's
resolution are lost by the time a PCAP frame record of the packet is built
(:issue:`1507`). This subclass yields the same timestamp, as a
:class:`~pcapkit.toolkit.dpkt.Timestamp` or, for a nanosecond file, a
:class:`~pcapkit.toolkit.dpkt.DecimalTimestamp`, which carries them.
Built when called, so that importing this module does not import `DPKT`_,
and so that it always subclasses the :class:`dpkt.pcap.Reader` of the moment.
Warns:
DPKTWarning: If this `DPKT`_'s reader lacks the attributes the loop reads
-- they are private, and |dpkt|_ is not pinned. The reader then runs
unchanged, and a traced PCAP record's original length is its captured
length.
Returns:
The reader type.
"""
import dpkt # isort:skip
from pcapkit.toolkit.dpkt import DecimalTimestamp, Timestamp # isort:skip
#: Attributes of :class:`dpkt.pcap.Reader` the loop below reads, as of dpkt 1.9.8.
private = ('_Reader__f', '_Reader__ph', '_divisor')
class _Reader(dpkt.pcap.Reader):
def __init__(self, fileobj: 'IO[bytes]') -> 'None':
super().__init__(fileobj)
missing = [name for name in private if not hasattr(self, name)]
#: Whether to read with ``dpkt.pcap.Reader.__iter__`` unchanged.
self._upstream = bool(missing)
if missing:
warn(f"dpkt {getattr(dpkt, '__version__', '(unknown version)')}: "
f"'dpkt.pcap.Reader' has no {', '.join(repr(name) for name in missing)}; "
"reading PCAP records without their original length, so a traced "
'PCAP record gives its captured length instead', DPKTWarning,
stacklevel=stacklevel())
# NOTE: ``dpkt.pcap.Reader.__iter__``'s loop, as a generator so ``Reader.__init__`` runs
# none of it before :attr:`_upstream` is set; nanosecond files yield ``Decimal`` timestamps.
def __iter__(self) -> 'Iterator[tuple[float | decimal.Decimal, bytes]]':
if self._upstream:
yield from super().__iter__()
return
file, header = self._Reader__f, self._Reader__ph # pylint: disable=no-member
nanosecond = isinstance(self._divisor, decimal.Decimal)
resolution = RESOLUTION[nanosecond]
while True:
buf = file.read(header.__hdr_len__)
if not buf:
break
hdr = header(buf)
buf = file.read(hdr.caplen)
value = hdr.tv_sec + (hdr.tv_usec / self._divisor)
if nanosecond:
yield DecimalTimestamp(value, resolution, hdr.len), buf
else:
# at a precision of 64 digits, not the caller's, which may
# be too low to hold it (#1668)
with localcontext(prec=64):
exact = decimal.Decimal(hdr.tv_sec) + decimal.Decimal(hdr.tv_usec) / resolution
yield Timestamp(value, exact, resolution, hdr.len), buf
return _Reader
@functools.lru_cache(maxsize=None)
def _loopback_protocol(base: 'Type[DPKTPacket]', ip: 'Type[DPKTPacket]', ip6: 'Type[DPKTPacket]',
linktype: 'int') -> 'Type[DPKTPacket]':
"""A :class:`dpkt.loopback.Loopback` that reads the address family by link type.
It reads the address family in the byte order
:func:`~pcapkit.protocols.link.loopback.protocol.family_byteorder` gives for
``linktype`` -- network byte order for ``LINKTYPE_LOOP``, inferred for
``LINKTYPE_NULL`` -- and dispatches on families 2 (IPv4) and 24, 28 and 30
(IPv6). Unlike `DPKT`_'s own, it names an IPv4 or IPv6 payload ``ip`` or
``ip6``, as :class:`dpkt.ethernet.Ethernet` does and as
:mod:`pcapkit.toolkit.dpkt` looks for it, and keeps a payload of any other
family as raw octets rather than parsing it as Ethernet (:issue:`1574`).
Args:
base: :class:`dpkt.loopback.Loopback` itself.
ip: :class:`dpkt.ip.IP`, the payload of family 2.
ip6: :class:`dpkt.ip6.IP6`, the payload of families 24, 28 and 30.
linktype: ``LINKTYPE_NULL`` or ``LINKTYPE_LOOP``.
Returns:
The packet type, built once per argument tuple rather than once per
packet. Every class it is built from is part of that key, so a reloaded
`DPKT`_ module gets a class of its own instead of one holding stale
classes.
"""
import dpkt # isort:skip
from pcapkit.protocols.link.loopback.protocol import family_byteorder # isort:skip
layers = {2: ip, 24: ip6, 28: ip6, 30: ip6}
# ``base`` is an argument so that the class derives from the given module's
# own ``Loopback``; mypy takes a base class only from a name it can resolve
# statically
class Loopback(base): # type: ignore[valid-type,misc]
"""BSD loopback encapsulation."""
def unpack(ext, buf: 'bytes') -> 'None':
dpkt.dpkt.Packet.unpack(ext, buf)
family = buf[:ext.__hdr_len__]
ext.family = int.from_bytes(family, family_byteorder(family, linktype))
layer = layers.get(ext.family)
if layer is not None:
payload = layer(buf[ext.__hdr_len__:])
ext.data = payload
setattr(ext, layer.__name__.lower(), payload)
return Loopback
[docs]
class DPKT(EngineBase['DPKTPacket']):
"""DPKT engine support.
A PCAP-NG file is read by :class:`PCAPNGReader` rather than by
``dpkt.pcapng.Reader``, so each packet takes the link type and the
timestamp resolution and offset of the interface it was captured on.
It reads the Enhanced, the Simple and the (obsolete) Packet Block, and
skips every other block without a warning. `DPKT`_'s own reader skips the
Simple Packet Block too.
Args:
extractor: :class:`~pcapkit.foundation.extraction.Extractor` instance.
"""
if TYPE_CHECKING:
#: Engine extraction package, i.e. |dpkt|_.
_expkg: '_DPKTPackage'
#: Engine extraction temporary storage.
_extmp: 'Reader'
##########################################################################
# Defaults.
##########################################################################
#: Engine name.
__engine_name__ = 'DPKT'
#: Engine module name.
__engine_module__ = 'dpkt'
##########################################################################
# Data models.
##########################################################################
def __init__(self, extractor: 'Extractor') -> 'None':
import dpkt # isort:skip
self._expkg = dpkt
self._extmp = cast('Reader', None)
super().__init__(extractor)
##########################################################################
# Methods.
##########################################################################
[docs]
def run(self) -> 'None':
"""Call :class:`dpkt.pcap.Reader` to extract PCAP files.
This method assigns :attr:`self._expkg <DPKT._expkg>`
as |dpkt|_ and :attr:`self._extmp <DPKT._extmp>`
as an iterator from :class:`dpkt.pcap.Reader`, or from
:class:`PCAPNGReader` for a PCAP-NG file.
The global header is parsed and dumped first, by
:meth:`self.extractor.record_header <pcapkit.foundation.extraction.Extractor.record_header>`.
Warns:
AttributeWarning: If :attr:`self.extractor._exlyr <pcapkit.foundation.extraction.Extractor._exlyr>`
and/or :attr:`self.extractor._exptl <pcapkit.foundation.extraction.Extractor._exptl>`
is provided as the DPKT engine does not support such operations;
or if :attr:`self.extractor._exctx <pcapkit.foundation.extraction.Extractor._exctx>`
is provided, as the DPKT engine does not parse with :mod:`pcapkit`'s own
protocol implementations; or if DTLS reassembly is enabled, for the
same reason.
Raises:
FormatError: If the file format is not supported, i.e., not a PCAP
and/or PCAP-NG file.
"""
from pcapkit.foundation.engines.pcap import PCAP
from pcapkit.foundation.engines.pcapng import PCAPNG
ext = self._extractor
if ext._exlyr != 'none' or ext._exptl != 'null':
warn("'Extractor(engine=dpkt)' does not support protocol and layer threshold; "
f"'layer={ext._exlyr}' and 'protocol={ext._exptl}' ignored",
AttributeWarning, stacklevel=stacklevel())
if ext._exctx:
warn("'Extractor(engine=dpkt)' does not parse with pcapkit's own protocol "
"implementations, so the parsing context supplied through "
"'context=' is ignored",
AttributeWarning, stacklevel=stacklevel())
if ext._flag_r and ext._dtls:
ext._dtls = False
ext._reasm = ReassemblyManager(ipv4=ext._reasm.ipv4, ipv6=ext._reasm.ipv6,
tcp=ext._reasm.tcp, dtls=None)
warn("'Extractor(engine=dpkt)' object does not support DTLS reassembly, "
"which reads pcapkit's own DTLS layer; so 'dtls=True' will be ignored",
AttributeWarning, stacklevel=stacklevel())
# setup verbose handler, unless the caller gave one of their own (#1874)
if ext._vfunc_default:
from pcapkit.toolkit.dpkt import packet2chain # isort:skip
ext._vfunc = lambda e, f: print(
f'Frame {e._frnum:>3d}: {packet2chain(cast("DPKTPacket", f))}' # pylint: disable=protected-access
)
# extract global header
ext.record_header()
if ext.magic_number in PCAP.MAGIC_NUMBER:
logger.debug('dpkt: reading %s as PCAP', ext._ifnm)
reader = _pcap_reader()(ext._ifile) # type: Reader
elif ext.magic_number in PCAPNG.MAGIC_NUMBER:
logger.debug('dpkt: reading %s as PCAP-NG', ext._ifnm)
reader = PCAPNGReader(ext._ifile)
else:
raise FormatError(f'unsupported file format: {ext.magic_number!r}')
# extract & analyse file
self._extmp = reader
[docs]
def read_frame(self) -> 'DPKTPacket':
"""Read frames with DPKT engine.
Returns:
Parsed frame instance.
See Also:
Please refer to :meth:`PCAP.read_frame <pcapkit.foundation.engines.pcap.PCAP.read_frame>`
for more operational information.
"""
from pcapkit.toolkit.dpkt import (attach_buffer, attach_timestamp, ipv4_reassembly,
ipv6_reassembly, packet2dict, tcp_reassembly,
tcp_traceflow)
ext = self._extractor
reader = self._extmp
# fetch DPKT packet; a PCAP-NG reader reports the link type of the
# interface this packet was captured on, so ask only after reading it
timestamp, pkt = cast('tuple[float, bytes]', next(reader))
linktype = Enum_LinkType.get(reader.datalink())
protocol = self._get_protocol(linktype, pkt)
try:
packet = protocol(pkt) # type: DPKTPacket
except Exception as exc: # pylint: disable=broad-except
# NOTE: caught broadly, as :meth:`Protocol.analyze` does, since dpkt raises more than
# ``UnpackError`` (#1351); the frame is kept raw so later frames keep their numbering.
warn(f'Frame {ext._frnum + 1}: dpkt cannot parse the frame '
f'({type(exc).__name__}: {exc}); kept as raw data',
DPKTWarning, stacklevel=stacklevel())
packet = self._get_raw_protocol()(pkt)
# DPKT hands the record's timestamp back beside its octets and only the
# octets become a packet, so the frame would otherwise not know when it was
# captured -- and anything reading a *stored* frame after this loop has
# moved on could not find out. Keep the two together from the start.
attach_timestamp(packet, timestamp)
# Nor does the packet keep the octets it was parsed from, and serialising
# it again recomputes zeroed checksums and lengths into it, so keep those
# too for the adapters to slice the wire octets out of.
attach_buffer(packet, pkt)
# verbose output
ext._frnum += 1
ext._vfunc(ext, packet)
# write output
frnum = f'Frame {ext._frnum}'
if not ext._flag_q:
info = packet2dict(packet, timestamp, data_link=linktype)
if ext._flag_f:
ofile = ext._ofile(f'{ext._ofnm}/{frnum}.{ext._fext}')
ofile(info, name=frnum)
else:
ext._ofile(info, name=frnum)
# record fragments
if ext._flag_r:
if ext._ipv4:
data_ipv4 = ipv4_reassembly(packet, timestamp, count=ext._frnum)
if data_ipv4 is not None:
ext._reasm.ipv4(data_ipv4)
if ext._ipv6:
data_ipv6 = ipv6_reassembly(packet, timestamp, count=ext._frnum)
if data_ipv6 is not None:
ext._reasm.ipv6(data_ipv6)
if ext._tcp:
data_tcp = tcp_reassembly(packet, timestamp, count=ext._frnum)
if data_tcp is not None:
ext._reasm.tcp(data_tcp)
# trace flows
if ext._flag_t:
if ext._tcp:
data_tf_tcp = tcp_traceflow(packet, timestamp, data_link=linktype, count=ext._frnum)
if data_tf_tcp is not None:
ext._trace.tcp(data_tf_tcp)
# record frames
if ext._flag_d:
# setattr(packet, 'packet2dict', packet2dict)
# setattr(packet, 'packet2chain', packet2chain)
ext._frame.append(packet)
# return frame record
return packet
##########################################################################
# Utilities.
##########################################################################
def _get_protocol(self, linktype: 'Optional[Enum_LinkType]' = None,
data: 'bytes' = b'') -> 'Type[DPKTPacket]':
"""Returns the protocol for parsing the current packet.
Args:
linktype: Link type code.
data: Octets of the current packet. ``LINKTYPE_RAW`` (101) carries
a bare IPv4 or IPv6 datagram, and only the version nibble of
its first octet tells which (:issue:`1502`).
Notes:
``LINKTYPE_NULL`` (0) and ``LINKTYPE_LOOP`` (108) are parsed with
:func:`_loopback_protocol` (:issue:`1574`).
"""
dpkt = self._expkg
reader = self._extmp
if linktype is None:
linktype = Enum_LinkType.get(reader.datalink())
version = data[0] >> 4 if data else None
pkg: 'Type[DPKTPacket]'
if linktype == Enum_LinkType.ETHERNET:
pkg = dpkt.ethernet.Ethernet
elif linktype.value == Enum_LinkType.IPV4 or (linktype.value == Enum_LinkType.RAW and version == 4):
pkg = dpkt.ip.IP
elif linktype.value == Enum_LinkType.IPV6 or (linktype.value == Enum_LinkType.RAW and version == 6):
pkg = dpkt.ip6.IP6
elif linktype.value in (Enum_LinkType.NULL, Enum_LinkType.LOOP):
pkg = _loopback_protocol(dpkt.loopback.Loopback, dpkt.ip.IP, dpkt.ip6.IP6, linktype.value)
else:
warn('unrecognised link layer protocol; all analysis functions ignored',
DPKTWarning, stacklevel=stacklevel())
pkg = self._get_raw_protocol()
return pkg
def _get_raw_protocol(self) -> 'Type[DPKTPacket]':
"""Returns a protocol that keeps the packet as raw data."""
dpkt = self._expkg
class RawPacket(dpkt.dpkt.Packet): # type: ignore[name-defined]
"""Raw packet."""
def __len__(ext) -> 'int':
return len(ext.data)
def __bytes__(ext) -> 'bytes':
return ext.data
def unpack(ext, buf: 'bytes') -> 'None':
ext.data = buf
return RawPacket