# -*- 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`.
.. _DPKT: https://dpkt.readthedocs.io
"""
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.protocols.schema.misc.pcap.header import RESOLUTION
from pcapkit.protocols.schema.misc.pcapng 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 types import ModuleType
from typing import IO, Iterator, Literal, Optional, Type, Union
from dpkt.dpkt import Packet as DPKTPacket
from dpkt.pcap import Reader as PCAPReader
from pcapkit.foundation.engines.pcapng import Context
from pcapkit.foundation.extraction import Extractor
from pcapkit.protocols.data.misc.pcapng import IF_TSResolOption as Data_IF_TSResolOption
from pcapkit.protocols.data.misc.pcapng import \
InterfaceDescriptionBlock as Data_InterfaceDescriptionBlock
from pcapkit.protocols.data.misc.pcapng import SectionHeaderBlock as Data_SectionHeaderBlock
from pcapkit.protocols.data.misc.pcapng import SimplePacketBlock as Data_SimplePacketBlock
Reader = Union[PCAPReader, 'PCAPNGReader']
#: 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.schema.misc.pcapng.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.
:class:`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.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. :class:`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.PCAPNG` as well
(:issue:`1520`).
.. _DPKT: https://dpkt.readthedocs.io
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 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`: the
:obj:`float` this reader always yielded, which 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 = interface.options.get(Enum_OptionType.if_tsresol)
tsoffset = 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.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.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 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.
.. _DPKT: https://dpkt.readthedocs.io
"""
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): # type: ignore[name-defined,misc]
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 :meth:`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: the loop is :meth:`dpkt.pcap.Reader.__iter__`'s, over the stream
# and the record header type that method reads them through. A generator,
# so ``dpkt.pcap.Reader.__init__`` calling it runs none of it before
# :attr:`_upstream` is set.
def __iter__(self) -> 'Iterator[tuple[float, 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 (:issue:`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.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.
.. _DPKT: https://dpkt.readthedocs.io
"""
import dpkt # isort:skip
from pcapkit.protocols.link.loopback import family_byteorder # isort:skip
layers = {2: ip, 24: ip6, 28: ip6, 30: ip6}
class Loopback(base):
"""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
:class:`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.
.. _DPKT: https://dpkt.readthedocs.io
Args:
extractor: :class:`~pcapkit.foundation.extraction.Extractor` instance.
"""
if TYPE_CHECKING:
#: Engine extraction package, i.e. :mod:`dpkt`.
_expkg: 'ModuleType'
#: 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 :mod:`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.
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())
# setup verbose handler
if ext._flag_v:
from pcapkit.toolkit.dpkt import packet2chain # isort:skip
ext._vfunc = lambda e, f: print(
f'Frame {e._frnum:>3d}: {packet2chain(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)
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:`ProtocolBase.analyze
# <pcapkit.protocols.protocol.ProtocolBase.analyze>` does: dpkt raises
# more than its own ``UnpackError`` while parsing -- dpkt 1.9.8 reads
# ``frag_off`` from whatever header follows an IPv6 Fragment header,
# so a Fragment header followed by ESP or Destination Options raises
# :exc:`AttributeError` (:issue:`1351`). The frame is kept as a raw
# packet, so one undecodable frame neither ends the extraction nor
# shifts the numbering of the frames after it.
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
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