From 8e8777030fe1aa7879a6e37dcfe022b1cd349347 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Fri, 9 Oct 2026 20:09:49 -0400 Subject: [PATCH] fix(engines): write real PCAP traces for dpkt and scapy (#1507) - toolkit/dpkt, toolkit/scapy: tcp_traceflow reports a FrameRecord, the engine's packet2dict mapping as before, carrying the captured octets, the PCAP record header and the exact time_epoch that only the PCAP trace dumper reads. - engines/dpkt: the PCAP and PCAP-NG readers yield a Timestamp (float) or, for a nanosecond PCAP, a DecimalTimestamp: the value dpkt yields, carrying the exact timestamp, the resolution and the original length. Under a dpkt without the reader internals this relies on, it reads with dpkt's own loop and warns. - engines/scapy: captures are read through PcapReader subclasses that stamp each packet with its PCAP or PCAP-NG interface resolution, and close once. - foundation/traceflow/data: new FrameRecord, a dict with packet, frame_info and time_epoch attributes, picklable with every protocol. - foundation/extraction, interface/misc: dpkt and scapy leave the trace_format='pcap' -> JSON fallback; #1508's Gap(1507) keeps pypcapfile. Traced PCAP files are byte-identical to the default engine's at both trace resolutions, and JSON, plist and tree output byte-identical to main's. --- pcapkit/foundation/engines/dpkt.py | 86 +++- pcapkit/foundation/engines/scapy.py | 78 +++- pcapkit/foundation/extraction.py | 13 +- pcapkit/foundation/traceflow/data/data.py | 66 ++- pcapkit/interface/misc.py | 26 +- pcapkit/toolkit/dpkt.py | 151 ++++++- pcapkit/toolkit/scapy.py | 77 +++- .../engines/test_engine_agreement_runtime.py | 14 +- .../engines/test_runtime_engines.py | 35 +- tests/foundation/test_extraction.py | 54 ++- tests/interface/test_misc.py | 81 +--- .../toolkit/test_engine_pcap_trace_runtime.py | 384 ++++++++++++++++++ 12 files changed, 912 insertions(+), 153 deletions(-) create mode 100644 tests/toolkit/test_engine_pcap_trace_runtime.py diff --git a/pcapkit/foundation/engines/dpkt.py b/pcapkit/foundation/engines/dpkt.py index d9d79e8005..6139fe90d7 100644 --- a/pcapkit/foundation/engines/dpkt.py +++ b/pcapkit/foundation/engines/dpkt.py @@ -134,7 +134,11 @@ def _read_packet(self, buf: 'bytes', block_type: 'int', little: Whether the enclosing section is little-endian. Returns: - Timestamp in seconds since the UNIX epoch, and the packet data. + 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 @@ -143,6 +147,8 @@ def _read_packet(self, buf: 'bytes', block_type: 'int', """ 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 @@ -168,7 +174,81 @@ def _read_packet(self, buf: 'bytes', block_type: 'int', timestamp = decimal.Decimal(ticks) / resolution + offset self._linktype = interface.linktype - return float(timestamp), pkt.pkt_data + return (Timestamp(float(timestamp), timestamp, resolution, pkt.pkt_len), # pylint: disable=no-member + pkt.pkt_data) + + +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 the two. + + 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 = 1_000_000_000 if nanosecond else 1_000_000 + 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: + exact = decimal.Decimal(hdr.tv_sec) + decimal.Decimal(hdr.tv_usec) / resolution + yield Timestamp(value, exact, resolution, hdr.len), buf + + return _Reader class DPKT(EngineBase['DPKTPacket']): @@ -276,7 +356,7 @@ def run(self) -> 'None': if ext.magic_number in PCAP.MAGIC_NUMBER: logger.debug('dpkt: reading %s as PCAP', ext._ifnm) - reader = dpkt.pcap.Reader(ext._ifile) + 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) diff --git a/pcapkit/foundation/engines/scapy.py b/pcapkit/foundation/engines/scapy.py index 9a2c47743f..4c19d4a513 100644 --- a/pcapkit/foundation/engines/scapy.py +++ b/pcapkit/foundation/engines/scapy.py @@ -47,9 +47,10 @@ __all__ = ['Scapy'] if TYPE_CHECKING: - from typing import Iterator + from typing import Any, Iterator, Type from scapy.packet import Packet as ScapyPacket + from scapy.utils import PcapReader from pcapkit.foundation.extraction import Extractor @@ -57,6 +58,68 @@ #: :data:`pcapkit.utilities.logging.logger`. logger = get_logger(__name__) +def _reader_type() -> 'Type[PcapReader]': + """A :class:`scapy.utils.PcapReader` that stamps each packet with its resolution. + + `Scapy`_'s readers convert a record's timestamp to a :class:`~decimal.Decimal` + and drop the resolution it was counted in, and an exact count of microseconds + comes out with the same digits whether the capture counted microseconds or + nanoseconds. The PCAP frame record of a packet needs that resolution, so these + subclasses attach it with :func:`pcapkit.toolkit.scapy.attach_resolution`: + from the global header of a PCAP, and from the interface of each PCAP-NG + packet block, which :meth:`~scapy.utils.RawPcapNgReader._read_block` is the + last place to see. + + Built when called, so that importing this module does not import `Scapy`_, + and so that it always subclasses the :class:`scapy.utils.PcapReader` of the + moment. + + Returns: + A reader type, which opens a PCAP-NG file as its PCAP-NG alternative. + + .. _Scapy: https://scapy.net + + """ + from scapy.utils import PcapNgReader, PcapReader # isort:skip + + from pcapkit.toolkit.scapy import attach_resolution # isort:skip + + class _Closing: + # ``sniff`` closes the reader at the end of the file, and the engine closes + # it again in case ``sniff`` raised first; only the first close is real, + # since closing a gzip reader twice raises + _closed = False + + def close(self) -> 'None': + if not self._closed: + self._closed = True + super().close() # type: ignore[misc] + + class _PcapReader(_Closing, PcapReader): + def read_packet(self, *args: 'Any', **kwargs: 'Any') -> 'ScapyPacket': + packet = super().read_packet(*args, **kwargs) + attach_resolution(packet, 1_000_000_000 if self.nano else 1_000_000) + return packet + + class _PcapNgReader(_Closing, PcapNgReader): + # the metaclass makes this ``_PcapReader``'s alternative in turn + alternative = _PcapReader + + _resolution = 1_000_000 + + def _read_block(self, *args: 'Any', **kwargs: 'Any') -> 'Any': + block = super()._read_block(*args, **kwargs) + if block is not None: + self._resolution = block[1].tsresol + return block + + def read_packet(self, *args: 'Any', **kwargs: 'Any') -> 'ScapyPacket': + packet = super().read_packet(*args, **kwargs) + attach_resolution(packet, self._resolution) + return packet + + return _PcapReader + class Scapy(EngineBase['ScapyPacket']): """Scapy engine support. @@ -179,7 +242,18 @@ def run(self) -> 'None': # extract & analyse file logger.debug('scapy: sniffing %s', ext._ifnm) - self._extmp = iter(self._expkg.sniff(offline=ext._ifnm)) + # NOTE: the reader is opened here rather than by ``sniff(offline=...)`` so + # that every packet carries its capture's timestamp resolution, which + # Scapy's own readers discard (#1507). Labelled with the file name, as + # ``offline=`` labels it, so ``Packet.sniffed_on`` is unchanged. ``sniff`` + # closes it at the end of the file, as it closes the reader ``offline=`` + # opens; it is closed here as well in case ``sniff`` raised first, which + # the reader takes as a no-op once closed. + reader = _reader_type()(ext._ifnm) + try: + self._extmp = iter(self._expkg.sniff(opened_socket={reader: ext._ifnm})) + finally: + reader.close() def read_frame(self) -> 'ScapyPacket': """Read frames with Scapy engine. diff --git a/pcapkit/foundation/extraction.py b/pcapkit/foundation/extraction.py index a9d6ab8683..f6b953cd21 100644 --- a/pcapkit/foundation/extraction.py +++ b/pcapkit/foundation/extraction.py @@ -1431,13 +1431,12 @@ def __init__(self, # ` # itself substitutes ``'pcap'`` for ``None``. # - # DPKT and Scapy belong on this list too: their adapters build the frame - # with ``packet2dict`` exactly as the other two do, so both crash the - # same way. The guard is written from what the adapters produce, not - # from which engines are seen to crash, because an engine that - # dissects nothing never feeds the tracer and so hides the defect - # rather than avoiding it (#406). - if (self._exnam in ('dpkt', 'scapy', 'pyshark', 'pypcapfile') + # The guard is written from what the adapters produce, not from which + # engines are seen to crash, because an engine that dissects nothing + # never feeds the tracer and so hides the defect rather than avoiding + # it (#406). DPKT and Scapy are off the list because their adapters + # build a PCAP frame record from the captured octets (#1507). + if (self._exnam in ('pyshark', 'pypcapfile') and trace_format in ('pcap', 'cap', None)): warn(f"'Extractor(engine={self._exnam})' does not support 'trace_format={trace_format}'; " "using 'trace_format=\"json\"' instead", FormatWarning, stacklevel=stacklevel()) diff --git a/pcapkit/foundation/traceflow/data/data.py b/pcapkit/foundation/traceflow/data/data.py index 46e9fdcb0b..29e7754fd9 100644 --- a/pcapkit/foundation/traceflow/data/data.py +++ b/pcapkit/foundation/traceflow/data/data.py @@ -5,14 +5,16 @@ from pcapkit.corekit.infoclass import Info, info_final -__all__ = ['TraceFlowData', 'Deferred', 'DeferredPacket'] +__all__ = ['TraceFlowData', 'Deferred', 'DeferredPacket', 'FrameRecord'] if TYPE_CHECKING: + from decimal import Decimal from typing import Any, Optional from pcapkit.foundation.reassembly.data.tcp import Datagram as TCP_Datagram from pcapkit.foundation.reassembly.tcp import TCP as TCP_Reassembly from pcapkit.foundation.traceflow.data.tcp import Index as TCP_Index + from pcapkit.protocols.data.misc.pcap.frame import FrameInfo as Data_FrameInfo class Deferred: @@ -141,6 +143,68 @@ def to_dict(self) -> 'dict[str, Any]': return super().to_dict() # type: ignore[misc] # pylint: disable=no-member +class FrameRecord(dict): + """A frame as a third-party engine's flow tracing adapter reports it. + + The mapping is the engine's own dissection of the frame, exactly as before, so + every trace format that takes a mapping writes it unchanged. Beside it, as + attributes the mapping does not show, it carries what the PCAP trace dumper + reads of a frame -- :meth:`PCAPIO._append_value + ` takes ``packet``, + ``frame_info`` and ``time_epoch`` -- so ``trace_format='pcap'`` writes the + frame's record too (:issue:`1507`). + + Args: + mapping: The engine's dissection of the frame. + packet: The frame's captured octets. + frame_info: The frame's PCAP record header. + time_epoch: The frame's exact UNIX timestamp. + + """ + + __slots__ = ('packet', 'frame_info', 'time_epoch') + + #: The frame's captured octets. + packet: 'bytes' + #: The frame's PCAP record header, with the fraction in the capture's own + #: resolution. + frame_info: 'Data_FrameInfo' + #: The frame's exact UNIX timestamp, from which + #: :meth:`PCAPIO._make_timestamp ` + #: tells which resolution ``frame_info.ts_usec`` is in. A + #: :class:`~decimal.Decimal`, since only that settles it exactly. + time_epoch: 'Decimal' + + def __init__(self, mapping: 'dict[str, Any]', *, packet: 'bytes', + frame_info: 'Data_FrameInfo', time_epoch: 'Decimal') -> 'None': + super().__init__(mapping) + self.packet = packet + self.frame_info = frame_info + self.time_epoch = time_epoch + + def __reduce__(self) -> 'tuple[Any, ...]': + # NOTE: spelled out because a ``dict`` subclass with ``__slots__`` cannot + # be pickled with protocol 0 or 1 otherwise + return _frame_record, (dict(self), self.packet, self.frame_info, self.time_epoch) + + +def _frame_record(mapping: 'dict[str, Any]', packet: 'bytes', + frame_info: 'Data_FrameInfo', time_epoch: 'Decimal') -> 'FrameRecord': + """Rebuild a pickled :class:`FrameRecord`. + + Args: + mapping: The engine's dissection of the frame. + packet: The frame's captured octets. + frame_info: The frame's PCAP record header. + time_epoch: The frame's exact UNIX timestamp. + + Returns: + The frame record. + + """ + return FrameRecord(mapping, packet=packet, frame_info=frame_info, time_epoch=time_epoch) + + @info_final class TraceFlowData(Info): """Data storage for flow tracing.""" diff --git a/pcapkit/interface/misc.py b/pcapkit/interface/misc.py index a7401c9d9d..f8d8ffc3a0 100644 --- a/pcapkit/interface/misc.py +++ b/pcapkit/interface/misc.py @@ -21,7 +21,7 @@ from pcapkit.foundation.extraction import Extractor from pcapkit.foundation.reassembly.tcp import TCP as TCP_Reassembly from pcapkit.utilities.exceptions import stacklevel -from pcapkit.utilities.warnings import EngineWarning, FormatWarning, warn +from pcapkit.utilities.warnings import EngineWarning, warn if TYPE_CHECKING: from typing import Any, Callable, Optional @@ -115,30 +115,6 @@ def follow_tcp_stream(fin: 'Optional[str]' = None, verbose: 'bool' = False, EngineWarning, stacklevel=stacklevel()) engine = None - # NOTE: the DPKT and Scapy engines hand their frames to the flow tracer as plain - # :obj:`dict`\\ s, which the PCAP trace dumper cannot re-serialise -- it reaches - # for ``frame.packet`` and dies with ``AttributeError: 'dict' object has no - # attribute 'packet'`` (#399). The tracer defaults an unset ``format`` to - # ``'pcap'``, so following a stream through either engine would crash *during - # extraction*, before the reassembly below ever runs. - # - # :class:`Extractor ` guards both - # engines itself, so this is not what keeps the extraction alive -- it is - # what keeps it *quiet*. The two guards choose the same replacement format and so - # produce byte-identical traces; they differ only in when they complain. The - # Extractor warns for every substitution it makes, including the one nobody asked - # for, whereas here an unset ``format`` is not a request and is upgraded silently, - # and only an explicit but unusable one draws a warning -- with a message naming - # the engine's limitation rather than the ``trace_format=`` argument this function - # does not expose. Removing this therefore would not change any trace file, but it - # would make ``follow_tcp_stream(engine='dpkt')`` warn about a default the caller - # never chose. - if engine is not None and engine.lower() in ('dpkt', 'scapy') and format in ('pcap', 'cap', None): - if format is not None: - warn(f"extraction engine {engine} cannot write '{format}' trace files; " - "using 'json' instead", FormatWarning, stacklevel=stacklevel()) - format = 'json' - extraction = Extractor(fin=fin, fout=None, format=None, auto=True, extension=extension, store=True, files=False, nofile=True, verbose=verbose, engine=engine, layer=None, protocol=None, ip=False, ipv4=False, ipv6=False, tcp=True, diff --git a/pcapkit/toolkit/dpkt.py b/pcapkit/toolkit/dpkt.py index cd781ff6db..756afd3b46 100644 --- a/pcapkit/toolkit/dpkt.py +++ b/pcapkit/toolkit/dpkt.py @@ -13,16 +13,20 @@ """ import copy +import decimal import ipaddress from typing import TYPE_CHECKING, cast from pcapkit.const.reg.transtype import TransType as Enum_TransType from pcapkit.foundation.reassembly.data.ip import Packet as IP_Packet from pcapkit.foundation.reassembly.data.tcp import Packet as TCP_Packet +from pcapkit.foundation.traceflow.data.data import FrameRecord from pcapkit.foundation.traceflow.data.tcp import Packet as TF_TCP_Packet +from pcapkit.protocols.data.misc.pcap.frame import FrameInfo as Data_FrameInfo from pcapkit.utilities.exceptions import ProtocolError, UnsupportedCall if TYPE_CHECKING: + from decimal import Decimal from ipaddress import IPv4Address, IPv6Address from typing import Any, Optional @@ -34,8 +38,8 @@ from pcapkit.const.reg.linktype import LinkType as Enum_LinkType __all__ = [ - 'ipv6_hdr_len', 'attach_timestamp', 'packet2timestamp', 'attach_buffer', 'packet2bytes', - 'packet2chain', 'packet2dict', + 'RecordTimestamp', 'Timestamp', 'DecimalTimestamp', 'ipv6_hdr_len', 'attach_timestamp', 'packet2timestamp', 'attach_buffer', 'packet2bytes', + 'packet2frame', 'packet2chain', 'packet2dict', 'ipv4_reassembly', 'ipv6_reassembly', 'tcp_reassembly', 'tcp_traceflow' ] @@ -69,6 +73,87 @@ BUFFER_ATTR = '__pcapkit_buffer__' +class RecordTimestamp: + """A record's capture timestamp, with what a PCAP frame record needs of it. + + The readers :class:`~pcapkit.foundation.engines.dpkt.DPKT` uses yield + ``(timestamp, octets)`` and nothing else, so a record's exact timestamp, its + resolution and its original length would be lost between the reader and + :func:`packet2frame`. The engine's readers yield a :class:`Timestamp` or a + :class:`DecimalTimestamp` instead, which *is* the value `DPKT`_ yields -- + so reassembly and flow labels see exactly what they did before -- and also + carries those three. + + .. _DPKT: https://dpkt.readthedocs.io + + """ + + __slots__ = () + + if TYPE_CHECKING: + #: Exact timestamp, in seconds since the UNIX epoch. + exact: 'Decimal' + #: Timestamp resolution of the record, in units per second. + resolution: 'int' + #: Original length of the packet, or :data:`None` if unknown. + orig_len: 'Optional[int]' + + +class Timestamp(RecordTimestamp, float): + """A :obj:`float` capture timestamp; see :class:`RecordTimestamp`. + + Args: + value: The timestamp, as `DPKT`_ yields it. + exact: Exact timestamp; ``value`` itself if not given. + resolution: Timestamp resolution of the record, in units per second. + orig_len: Original length of the packet, or :data:`None` if unknown. + + .. _DPKT: https://dpkt.readthedocs.io + + """ + + __slots__ = ('exact', 'resolution', 'orig_len') + + def __new__(cls, value: 'Any' = 0.0, exact: 'Optional[Decimal]' = None, + resolution: 'int' = 1_000_000, orig_len: 'Optional[int]' = None) -> 'Timestamp': + self = super().__new__(cls, value) + self.exact = decimal.Decimal(float(self)) if exact is None else exact + self.resolution = resolution + self.orig_len = orig_len + return self + + def __reduce__(self) -> 'tuple[Any, ...]': + return type(self), (float(self), self.exact, self.resolution, self.orig_len) + + +class DecimalTimestamp(RecordTimestamp, decimal.Decimal): + """A :class:`~decimal.Decimal` capture timestamp; see :class:`RecordTimestamp`. + + `DPKT`_ yields a :class:`~decimal.Decimal` for a nanosecond PCAP. + + Args: + value: The timestamp, as `DPKT`_ yields it, which is exact. + resolution: Timestamp resolution of the record, in units per second. + orig_len: Original length of the packet, or :data:`None` if unknown. + + .. _DPKT: https://dpkt.readthedocs.io + + """ + + __slots__ = ('exact', 'resolution', 'orig_len') + + def __new__(cls, value: 'Any' = '0', resolution: 'int' = 1_000_000_000, + orig_len: 'Optional[int]' = None) -> 'DecimalTimestamp': + self = super().__new__(cls, value) + self.exact = decimal.Decimal(self) + self.resolution = resolution + self.orig_len = orig_len + return self + + def __reduce__(self) -> 'tuple[Any, ...]': + return type(self), (str(self), self.resolution, self.orig_len) + + def attach_timestamp(packet: 'Packet', timestamp: 'float') -> 'None': """Stash a frame's capture timestamp on the frame. @@ -145,6 +230,66 @@ def packet2bytes(packet: 'Packet') -> 'bytes': return cast('bytes', copy.deepcopy(packet).pack()) +def packet2frame(packet: 'Packet', timestamp: 'float | Decimal', *, + data_link: 'Enum_LinkType') -> 'FrameRecord': + """Report a DPKT packet to the flow tracer, with its PCAP frame record. + + Args: + packet: DPKT packet. + timestamp: Capture timestamp of the packet. The one attached by + :func:`attach_timestamp`, if any, is preferred, since it is the value + `DPKT`_'s reader yielded. + data_link: Data link type. + + Returns: + The :func:`packet2dict` mapping of the packet, which every trace format + but PCAP writes as it always has, carrying the captured octets, from + :func:`packet2bytes`, the PCAP record header the PCAP trace dumper + writes, and the exact timestamp it reads the header's resolution from. + + Note: + The fraction in ``frame_info.ts_usec`` is in the capture's own resolution, + as the default engine's is, and both it and ``orig_len`` come from the + reader: :class:`~pcapkit.foundation.engines.dpkt.DPKT` reads PCAP and + PCAP-NG alike into a :class:`RecordTimestamp`. A timestamp that is not one + -- a packet built by hand, or read by a ``dpkt`` whose reader the engine + could not extend -- is read as `DPKT`_ yields it: a + :class:`~decimal.Decimal` counts nanoseconds and a :obj:`float` + microseconds, and the octets are taken to be whole. + + .. _DPKT: https://dpkt.readthedocs.io + + """ + stamp = getattr(packet, TIMESTAMP_ATTR, timestamp) + octets = packet2bytes(packet) + if isinstance(stamp, RecordTimestamp): + epoch, resolution, orig_len = stamp.exact, stamp.resolution, stamp.orig_len + elif isinstance(stamp, decimal.Decimal): + epoch, resolution, orig_len = stamp, 1_000_000_000, None + else: + epoch, resolution, orig_len = round(decimal.Decimal(stamp), 6), 1_000_000, None + ts_sec = int(epoch) + ts_usec = int((epoch - ts_sec) * (1_000_000_000 if resolution > 1_000_000 else 1_000_000)) + + # the mapping is exactly the one reported before #1507, plain value types + # included, so that the dict-capable trace formats write the same bytes + if isinstance(timestamp, float): + timestamp = float(timestamp) + elif isinstance(timestamp, decimal.Decimal): + timestamp = decimal.Decimal(timestamp) + return FrameRecord( + packet2dict(packet, timestamp, data_link=data_link), + packet=octets, + frame_info=Data_FrameInfo( + ts_sec=ts_sec, + ts_usec=ts_usec, + incl_len=len(octets), + orig_len=len(octets) if orig_len is None else orig_len, + ), + time_epoch=decimal.Decimal(epoch), + ) + + def _ipv6_ext_hdrs(ipv6: 'IP6') -> 'list[Any]': """Fetch an IPv6 packet's extension headers, in the order they were parsed. @@ -597,7 +742,7 @@ def tcp_traceflow(packet: 'Packet', timestamp: 'float', *, data = TF_TCP_Packet( # type: ignore[type-var] protocol=data_link, # data link type from global header index=count, # frame number - frame=packet2dict(packet, timestamp, data_link=data_link), # extracted packet + frame=packet2frame(packet, timestamp, data_link=data_link), # extracted packet syn=bool(int(flags[6])), # TCP synchronise (SYN) flag fin=bool(int(flags[7])), # TCP finish (FIN) flag rst=bool(int(flags[5])), # TCP reset (RST) flag diff --git a/pcapkit/toolkit/scapy.py b/pcapkit/toolkit/scapy.py index 3b45d816db..9dcf82632e 100644 --- a/pcapkit/toolkit/scapy.py +++ b/pcapkit/toolkit/scapy.py @@ -35,6 +35,7 @@ importing :mod:`scapy.all` in its constructor. """ +import decimal import ipaddress from typing import TYPE_CHECKING, cast @@ -42,7 +43,9 @@ from pcapkit.const.reg.transtype import TransType as Enum_TransType from pcapkit.foundation.reassembly.data.ip import Packet as IP_Packet from pcapkit.foundation.reassembly.data.tcp import Packet as TCP_Packet +from pcapkit.foundation.traceflow.data.data import FrameRecord from pcapkit.foundation.traceflow.data.tcp import Packet as TF_TCP_Packet +from pcapkit.protocols.data.misc.pcap.frame import FrameInfo as Data_FrameInfo from pcapkit.utilities.compat import ModuleNotFoundError # pylint: disable=redefined-builtin from pcapkit.utilities.exceptions import MissingKeyError, ModuleNotFound, stacklevel from pcapkit.utilities.warnings import ScapyWarning, warn @@ -63,11 +66,28 @@ from scapy.packet import Packet __all__ = [ - 'packet2chain', 'packet2dict', + 'attach_resolution', 'packet2chain', 'packet2dict', 'packet2frame', 'ipv4_reassembly', 'ipv6_reassembly', 'tcp_reassembly', 'tcp_traceflow' ] +#: Attribute a packet's timestamp resolution is stashed under, in units per +#: second, by :func:`attach_resolution`. +RESOLUTION_ATTR = '__pcapkit_resolution__' + + +def attach_resolution(packet: 'Packet', resolution: 'int') -> 'None': + """Stash the timestamp resolution of a packet's capture on the packet. + + Args: + packet: Scapy packet. + resolution: Timestamp resolution, in units per second: the PCAP global + header's, or the PCAP-NG interface's the packet was captured on. + + """ + setattr(packet, RESOLUTION_ATTR, resolution) + + def packet2chain(packet: 'Packet') -> 'str': """Fetch Scapy packet protocol chain. @@ -137,6 +157,59 @@ def wrapper(packet: 'Packet') -> 'dict[str, Any]': } +def packet2frame(packet: 'Packet') -> 'FrameRecord': + """Report a Scapy packet to the flow tracer, with its PCAP frame record. + + Args: + packet: Scapy packet. + + Returns: + The :func:`packet2dict` mapping of the packet, which every trace format + but PCAP writes as it always has, carrying the octets `Scapy`_ dissected + the packet from, the PCAP record header the PCAP trace dumper writes, and + the exact timestamp it reads the header's resolution from. + + Raises: + ModuleNotFound: If `Scapy`_ is not installed. + + Note: + The octets are ``packet.original``, which `Scapy`_'s readers keep as read, + rather than ``bytes(packet)``, which rebuilds them. ``orig_len`` is + ``packet.wirelen``, the record's original length. + + The fraction in ``frame_info.ts_usec`` is in the capture's own resolution, + as the default engine's is. That resolution is the one + :class:`~pcapkit.foundation.engines.scapy.Scapy` attaches with + :func:`attach_resolution`, never one guessed from the value: a + nanosecond count that is a whole number of microseconds has the same + digits as the microsecond count. A packet that carries none, e.g. one + built by hand, is taken to count microseconds. + + """ + if scapy is None: + raise ModuleNotFound("No module named 'scapy'", name='scapy') + mapping = packet2dict(packet) + + original = getattr(packet, 'original', None) + octets = bytes(original) if isinstance(original, (bytes, bytearray)) else bytes(packet) + wirelen = getattr(packet, 'wirelen', None) + orig_len = len(octets) if wirelen is None else int(wirelen) + + nanosecond = getattr(packet, RESOLUTION_ATTR, 1_000_000) > 1_000_000 + stamp = packet.time + epoch = (decimal.Decimal(stamp) if isinstance(stamp, decimal.Decimal) + else round(decimal.Decimal(stamp), 9 if nanosecond else 6)) + ts_sec = int(epoch) + ts_usec = int((epoch - ts_sec) * (1_000_000_000 if nanosecond else 1_000_000)) + + return FrameRecord(mapping, packet=octets, frame_info=Data_FrameInfo( + ts_sec=ts_sec, + ts_usec=ts_usec, + incl_len=len(octets), + orig_len=orig_len, + ), time_epoch=decimal.Decimal(epoch)) + + def ipv4_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'IP_Packet[IPv4Address] | None': """Make data for IPv4 reassembly. @@ -360,7 +433,7 @@ def tcp_traceflow(packet: 'Packet', *, count: 'int' = -1) -> 'TF_TCP_Packet | No data = TF_TCP_Packet( # type: ignore[type-var] protocol=protocol, # data link type index=count, # frame number - frame=packet2dict(packet), # extracted packet + frame=packet2frame(packet), # extracted packet syn=bool(tcp.flags.S), # TCP synchronise (SYN) flag fin=bool(tcp.flags.F), # TCP finish (FIN) flag rst=bool(tcp.flags.R), # TCP reset (RST) flag diff --git a/tests/foundation/engines/test_engine_agreement_runtime.py b/tests/foundation/engines/test_engine_agreement_runtime.py index 2acb389412..e36049a3b3 100644 --- a/tests/foundation/engines/test_engine_agreement_runtime.py +++ b/tests/foundation/engines/test_engine_agreement_runtime.py @@ -287,14 +287,12 @@ class TestEngineAgreement(harness.RoundTripBase): STATUSES = ('ERROR', 'FRAMES', 'RECORDS', 'INPUTS', 'DATAGRAMS', 'FLOWS') KNOWN_FAILURES = ( - Gap(1507, "the dpkt, scapy and pypcapfile engines write every flow as JSON when asked " - "for trace_format='pcap': their tcp_traceflow adapters build the frame with " - 'packet2dict (pcapkit/toolkit/dpkt.py:600, pcapkit/toolkit/scapy.py:363, ' - 'pcapkit/toolkit/pypcapfile.py:620), which PCAPIO cannot write, so Extractor ' - 'swaps the format with a FormatWarning (pcapkit/foundation/extraction.py:1440-1444)', - 'FLOWS', 'file of ', tuple(f'{capture}/{engine}/flows' for engine in ('dpkt', 'scapy') - for capture in TRACED) - + tuple(f'{capture}/pypcapfile/flows' for capture in TRACED_IPV4_ONLY_PCAP)), + Gap(1507, "the pypcapfile engine writes every flow as JSON when asked for " + "trace_format='pcap': its tcp_traceflow adapter builds the frame with " + 'packet2dict (pcapkit/toolkit/pypcapfile.py:620), which PCAPIO cannot write, so ' + 'Extractor swaps the format with a FormatWarning ' + '(pcapkit/foundation/extraction.py:1439-1443)', + 'FLOWS', 'file of ', tuple(f'{capture}/pypcapfile/flows' for capture in TRACED_IPV4_ONLY_PCAP)), Gap(1512, 'the pypcapfile engine reads no frame from a big-endian PCAP, without a warning: ' 'pcapfile.savefile.load_savefile (pypcapfile 0.12.0) yields no packets for one, ' 'and the engine iterates that empty list (pcapkit/foundation/engines/' diff --git a/tests/foundation/engines/test_runtime_engines.py b/tests/foundation/engines/test_runtime_engines.py index b30c667d73..1d3efedc9e 100644 --- a/tests/foundation/engines/test_runtime_engines.py +++ b/tests/foundation/engines/test_runtime_engines.py @@ -227,9 +227,13 @@ def __next__(self): extractor, sink = make_extractor(_exlyr='link', _flag_v=True) engine = DPKT(extractor) + # A PCAP is read through the engine's :class:`dpkt.pcap.Reader` subclass, + # which keeps each record's original length (#1507). pcap_reader = FakeReader() - with mock.patch.object(engine._expkg.pcap, 'Reader', return_value=pcap_reader): + reader_type = mock.Mock(return_value=pcap_reader) + with mock.patch('pcapkit.foundation.engines.dpkt._pcap_reader', return_value=reader_type): engine.run() + reader_type.assert_called_once_with(extractor._ifile) self.assertIs(engine._extmp, pcap_reader) self.assertTrue(callable(extractor._vfunc)) extractor.record_header.assert_called_once_with() @@ -247,7 +251,8 @@ def __next__(self): clean_extractor, _ = make_extractor(_flag_v=False) clean_engine = DPKT(clean_extractor) clean_reader = FakeReader() - with mock.patch.object(clean_engine._expkg.pcap, 'Reader', return_value=clean_reader): + with mock.patch('pcapkit.foundation.engines.dpkt._pcap_reader', + return_value=mock.Mock(return_value=clean_reader)): clean_engine.run() self.assertIs(clean_engine._extmp, clean_reader) @@ -345,19 +350,29 @@ def test_scapy_run_and_read_frame(self) -> None: from pcapkit.foundation.engines.scapy import Scapy packet = object() + + # The engine opens the capture through its own reader, which stamps each + # packet with its timestamp resolution (#1507), and hands it to ``sniff`` + # under the file's name. ``sniff`` closes it at the end of the file, and the + # engine closes it once more in case ``sniff`` raised first; the reader + # makes the second close a no-op. + def run(engine: 'Scapy') -> None: + reader = mock.Mock() + reader_type = mock.Mock(return_value=reader) + with mock.patch('pcapkit.foundation.engines.scapy._reader_type', return_value=reader_type), \ + mock.patch.object(engine._expkg, 'sniff', return_value=[packet]) as sniff: + engine.run() + reader_type.assert_called_once_with('capture.pcap') + sniff.assert_called_once_with(opened_socket={reader: 'capture.pcap'}) + reader.close.assert_called_once_with() + extractor, sink = make_extractor(_exptl='tcp', _flag_v=True) - engine = Scapy(extractor) - with mock.patch.object(engine._expkg, 'sniff', return_value=[packet]) as sniff: - engine.run() - sniff.assert_called_once_with(offline='capture.pcap') + run(Scapy(extractor)) self.assertTrue(callable(extractor._vfunc)) extractor.record_header.assert_called_once_with() clean_extractor, _ = make_extractor(_flag_v=False) - clean_engine = Scapy(clean_extractor) - with mock.patch.object(clean_engine._expkg, 'sniff', return_value=[packet]) as sniff: - clean_engine.run() - sniff.assert_called_once_with(offline='capture.pcap') + run(Scapy(clean_extractor)) extractor, sink = make_extractor(_flag_f=False) engine = Scapy(extractor) diff --git a/tests/foundation/test_extraction.py b/tests/foundation/test_extraction.py index a8a2d7c238..9ab71b6a58 100644 --- a/tests/foundation/test_extraction.py +++ b/tests/foundation/test_extraction.py @@ -1086,13 +1086,13 @@ def _traced(self, temp: pathlib.Path, tag: str, **kwargs: object): self.addCleanup(extractor._ifile.close) return extractor, warn - def test_pcap_trace_format_is_replaced_for_every_dict_frame_engine(self) -> None: - # The flow-tracing adapters of these four engines report each frame as a + def test_pcap_trace_format_is_replaced_only_for_dict_frame_engines(self) -> None: + # The flow-tracing adapters of PyShark and PyPCAPFile report each frame as a # plain dict, which the PCAP trace dumper cannot re-serialise: it reaches for # ``frame.packet`` and raises AttributeError from - # pcapkit.dumpkit.pcap.PCAPIO._append_value. DPKT and Scapy were once left - # off this list even though ``packet2dict`` builds their frames exactly as it - # builds the other two's. + # pcapkit.dumpkit.pcap.PCAPIO._append_value. DPKT and Scapy build a PCAP + # frame record from the captured octets instead (#1507), so their 'pcap' is + # kept, silently. # # ``None`` is asserted alongside 'pcap' and 'cap' because TraceFlow.__init__ # substitutes 'pcap' for it, so an unset format routes to the same dumper. @@ -1110,6 +1110,10 @@ def test_pcap_trace_format_is_replaced_for_every_dict_frame_engine(self) -> None traced, warn = self._traced(temp, tag, engine=engine, trace_format=trace_format) + if engine in ('dpkt', 'scapy'): + self.assertEqual(traced._trace.tcp._fdpext, '.pcap') + warn.assert_not_called() + continue self.assertEqual(traced._trace.tcp._fdpext, '.json') warn.assert_called_once() message = warn.call_args.args[0] @@ -1152,11 +1156,12 @@ def test_an_engine_with_real_frames_keeps_the_pcap_trace_format(self) -> None: @unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') class DictFrameTraceEndToEndTests(unittest.TestCase): - """A traced extraction on a dict-frame engine must complete, not crash. + """A traced extraction on the DPKT engine must write PCAP, not crash. - The constructor-level cases above prove the format was substituted; this proves - the substitution is *sufficient* -- the tracer runs, the dumper is handed the - mapping and writes it. Before the guard covered DPKT, this raised + The constructor-level cases above prove the format is kept; this proves the + adapter's frame record is *sufficient* -- the tracer runs, the dumper is handed + the record and writes it. Before #1507 the adapter built a :obj:`dict`, the + format was swapped for JSON, and before that guard this raised ``AttributeError: 'dict' object has no attribute 'packet'`` from :meth:`pcapkit.dumpkit.pcap.PCAPIO._append_value`. @@ -1166,23 +1171,15 @@ class DictFrameTraceEndToEndTests(unittest.TestCase): orthogonal -- it suppresses the *frame* output, not the trace output, so it neither causes nor prevents the crash. - Only DPKT is exercised end to end. Scapy's adapter produces the same mapping and - is covered by the constructor cases, but reaching its dumper needs the L2 link - types registered -- measured: with only :mod:`scapy.sendrecv` imported, as the - engine does today, every frame of ``in.pcap`` dissects as ``Raw``, no TCP layer - is found and the tracer is never fed (#406). Importing :mod:`scapy.all` to force - it is a *global and irreversible* change to ``scapy.conf``: measured on - ``in.pcap``, frames 3-5 gain a TCP layer afterwards. That would silently - invalidate the Scapy case in :mod:`tests.interface.test_misc`, which asserts the - opposite, depending on which of the two ran first. So it is deliberately not done - here. + Only DPKT is exercised here; both engines' traced files are byte-compared with + the default engine's in :mod:`tests.toolkit.test_engine_pcap_trace_runtime`. """ def setUp(self) -> None: reimport_once_per_class(self) @unittest.skipUnless(HAS_DPKT, 'dpkt not installed') - def test_dpkt_traced_extraction_writes_flows_instead_of_crashing(self) -> None: + def test_dpkt_traced_extraction_writes_pcap_flows(self) -> None: import pcapkit for trace_format in ('pcap', 'cap', None): @@ -1199,26 +1196,23 @@ def test_dpkt_traced_extraction_writes_flows_instead_of_crashing(self) -> None: ) # The engine that ran is the one asked for -- a fallback to the - # default engine would not exercise the dict-frame path at all. + # default engine would not exercise the DPKT adapter at all. self.assertEqual(type(extraction.engine).__engine_name__, 'DPKT') self.assertTrue(extraction.trace.tcp, 'no TCP flow was traced, so the dumper never ran ' 'and this test proves nothing') # Each traced flow named a file, and every one of them exists and - # holds the JSON the substituted format produces. + # is a PCAP file, written from the frame records (#1507). written = sorted(path for path in flows.rglob('*') if path.is_file()) self.assertTrue(written) for path in written: - self.assertEqual(path.suffix, '.json') - self.assertGreater(path.stat().st_size, 0) - - self.assertTrue( - any('json' in str(w.message) for w in caught - if w.category.__name__ == 'FormatWarning'), - 'the format substitution was not announced', - ) + self.assertEqual(path.suffix, '.pcap') + self.assertIn(path.read_bytes()[:4], (b'\xd4\xc3\xb2\xa1', b'\xa1\xb2\xc3\xd4')) + self.assertGreater(path.stat().st_size, 24) + self.assertEqual([str(w.message) for w in caught + if w.category.__name__ == 'FormatWarning'], []) if __name__ == '__main__': unittest.main() diff --git a/tests/interface/test_misc.py b/tests/interface/test_misc.py index 073dee665b..fdc9682ed8 100644 --- a/tests/interface/test_misc.py +++ b/tests/interface/test_misc.py @@ -205,71 +205,28 @@ def test_pyshark_and_pypcap_fall_back_to_the_default_engine(self) -> None: self.assertTrue(any(issubclass(w.category, EngineWarning) for w in caught)) self.assertEqual(len(streams), IN_PCAP_TCP_STREAMS) - @unittest.skipUnless(HAS_DPKT, 'dpkt not installed') - def test_dpkt_explicit_pcap_trace_format_is_downgraded_with_a_warning(self) -> None: - # The PCAP trace dumper cannot serialise DPKT's dict frames, so an explicit - # 'pcap' trace format is replaced with a dict-capable one, with a - # FormatWarning -- and the stream is still followed rather than crashing the - # extraction the way it did before #399 was fixed. - from pcapkit.utilities.warnings import FormatWarning - - with warnings.catch_warnings(record=True) as caught: - warnings.simplefilter('always') - streams = self._follow(engine='dpkt', format='pcap') - - self.assertTrue(any(issubclass(w.category, FormatWarning) for w in caught)) - self.assertEqual(len(streams), IN_PCAP_TCP_STREAMS) - - @unittest.skipUnless(HAS_DPKT, 'dpkt not installed') - def test_dpkt_unset_trace_format_is_upgraded_quietly(self) -> None: - # An unset trace format is not a request, so upgrading it to a dict-capable - # format for DPKT does not warn -- only an explicit, unusable one does. + def test_dpkt_and_scapy_keep_the_pcap_trace_format(self) -> None: + # Both engines' flow tracing adapters now hand the tracer a PCAP frame + # record built from the captured octets (#1507), so neither an explicit + # 'pcap' nor an unset format is replaced any more: every stream is written + # as PCAP, and nothing warns about a substitution. from pcapkit.utilities.warnings import FormatWarning - with warnings.catch_warnings(record=True) as caught: - warnings.simplefilter('always') - streams = self._follow(engine='dpkt') - - self.assertFalse(any(issubclass(w.category, FormatWarning) for w in caught)) - self.assertEqual(len(streams), IN_PCAP_TCP_STREAMS) - - @unittest.skipUnless(HAS_DPKT, 'dpkt not installed') - def test_the_local_format_upgrade_is_not_redundant(self) -> None: - # Extractor now guards DPKT and Scapy itself, which makes the local upgrade - # in follow_tcp_stream look removable. It is not, and this pins the reason so - # that deleting it fails here rather than only changing what users see. - # - # The two guards pick the same replacement format, so the traces are - # identical either way; they differ in when they complain. The Extractor - # announces every substitution, including for an unset format -- which - # follow_tcp_stream does not even expose, since its argument is ``format``. - # Here an unset format is upgraded silently. Remove the local guard and - # ``follow_tcp_stream(engine='dpkt')`` starts warning about a default the - # caller never chose. - import pcapkit - from pcapkit.utilities.warnings import FormatWarning - - def format_warnings(caught): - return [str(w.message) for w in caught if issubclass(w.category, FormatWarning)] - - with warnings.catch_warnings(record=True) as local: - warnings.simplefilter('always') - streams = self._follow(engine='dpkt', format=None) - - with warnings.catch_warnings(record=True) as core: - warnings.simplefilter('always') - extraction = pcapkit.extract(fin=sample_path('in.pcap'), engine='dpkt', - store=True, nofile=True, tcp=True, trace=True, - trace_fout=self.tmp_dir, trace_format=None) - - # Same work done either way ... - self.assertEqual(len(streams), IN_PCAP_TCP_STREAMS) - self.assertEqual(len(extraction.trace.tcp), IN_PCAP_TCP_STREAMS) + engines = [name for name, present in (('dpkt', HAS_DPKT), ('scapy', HAS_SCAPY)) if present] + if not engines: + self.skipTest('neither dpkt nor scapy installed') + for engine in engines: + for format in ('pcap', None): + with self.subTest(engine=engine, format=format), \ + tempfile.TemporaryDirectory() as fout, \ + warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + streams = self._follow(engine=engine, format=format, fout=fout) - # ... and the difference is only in the reporting. - self.assertEqual(format_warnings(local), []) - self.assertEqual(len(format_warnings(core)), 1) - self.assertIn('trace_format=None', format_warnings(core)[0]) + self.assertFalse(any(issubclass(w.category, FormatWarning) for w in caught)) + self.assertEqual(len(streams), IN_PCAP_TCP_STREAMS) + for stream in streams: + self.assertTrue(stream.filename.endswith('.pcap'), stream.filename) def test_engine_without_a_reassembly_adapter_returns_no_streams(self) -> None: # An engine pcapkit ships no reassembly adapter for -- a third-party one, or diff --git a/tests/toolkit/test_engine_pcap_trace_runtime.py b/tests/toolkit/test_engine_pcap_trace_runtime.py new file mode 100644 index 0000000000..b1d9e7717d --- /dev/null +++ b/tests/toolkit/test_engine_pcap_trace_runtime.py @@ -0,0 +1,384 @@ +# -*- coding: utf-8 -*- +"""DPKT and Scapy write ``trace_format='pcap'`` flows as PCAP (#1507). + +Both engines expose each frame's captured octets, its timestamp, its original +length and its link type, so their flow tracing adapters carry a PCAP frame +record beside the frame mapping they always reported, and +:class:`~pcapkit.foundation.extraction.Extractor` no longer swaps the format for +JSON. The checks are the strongest available: every traced PCAP file is +byte-compared with the default engine's for the same input, and every JSON trace +with the one main wrote before. + +The inputs no sample has are synthesised from :file:`tcp.pcap`: nanosecond PCAP +and PCAP-NG (``if_tsresol=9``) captures, each with a timestamp that has +sub-microsecond digits and one that is a whole number of microseconds, and +captures with a TCP frame the snapshot length cut short. + +The module reads generated captures, so it belongs to the fixture-dependent tier. + +""" +from __future__ import annotations + +import hashlib +import importlib.util +import os +import struct +import tempfile +import unittest +import warnings +from typing import TYPE_CHECKING +from unittest import mock + +from tests._support import close_extractor, reimport_once_per_class, sample_path + +if TYPE_CHECKING: + from typing import Any + +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None + for name in ('tbtrim', 'aenum', 'chardet', 'dictdumper')) +HAS_DPKT = importlib.util.find_spec('dpkt') is not None +HAS_SCAPY = importlib.util.find_spec('scapy') is not None +ENGINES = tuple(name for name, present in (('dpkt', HAS_DPKT), ('scapy', HAS_SCAPY)) if present) + +#: Captures with at least one TCP flow, one per link layer and container. +CAPTURES = ('tcp.pcap', 'http6.cap', 'in.pcap', 'many_interfaces.pcapng', 'test.pcap') + +#: Little-endian microsecond and nanosecond PCAP magic, as written on this host. +MAGIC = {False: bytes.fromhex('d4c3b2a1'), True: bytes.fromhex('4d3cb2a1')} + +#: Nanoseconds added to each microsecond timestamp: sub-microsecond digits, and +#: none at all, which leaves a nanosecond count with a microsecond count's digits. +EXTRA_NS = {'unaligned': 789, 'aligned': 0} + +#: SHA-256 of each JSON flow trace of ``tcp.pcap``, as main wrote it at 6af87f608. +JSON_TRACE_SHA256 = { + 'dpkt': { + '10.20.30.130_22-10.20.30.131_53406-1500000000.000774.json': + 'c3f17eb27a735437976439ea832746340b50966b85d364514110c7a176199eed', + 'fe80..a6.87f9.2793.16ee_51774-fe80..1ccd.7c77.bac7.46b7_22-1500000000.002433.json': + 'f0e9d767454ece7604f316c229c0407aba1f84cf321cf17822ba416cbcd3652f', + }, + 'scapy': { + '10.20.30.130_22-10.20.30.131_53406-1500000000.000774.json': + '7679cddff9512107b9f03d366d4d9f39155efa6ff4d3a83bcc27bea63865c252', + 'fe80..a6.87f9.2793.16ee_51774-fe80..1ccd.7c77.bac7.46b7_22-1500000000.002433.json': + 'a1953755ce695b1008f30eb67937e554381f01909144319dbf69035afb437ce1', + }, +} + + +def _records(source: 'str | bytes') -> 'tuple[bytes, list[list]]': + """Split a little-endian PCAP, by path or by content, into its global header + and its ``[ts_sec, ts_usec, orig_len, octets]`` records.""" + if isinstance(source, str): + with open(source, 'rb') as file: + source = file.read() + header, rest, records = source[:24], source[24:], [] + while rest: + ts_sec, ts_usec, incl_len, orig_len = struct.unpack(' 'str': + with open(path, 'wb') as file: + file.write(header) + for ts_sec, ts_usec, orig_len, octets in records: + file.write(struct.pack(' 'bytes': + body += b'\x00' * (-len(body) % 4) + length = len(body) + 12 + return struct.pack(' 'str': + """Write ``[ts_sec, ts_nsec, orig_len, octets]`` records as a little-endian + PCAP-NG capture on one Ethernet interface of ``if_tsresol=tsresol``.""" + shb = _block(0x0A0D0D0A, struct.pack('> 32, ticks & 0xFFFFFFFF, + len(octets), orig_len) + octets)) + return path + + +@unittest.skipUnless(HAS_RUNTIME and ENGINES, 'runtime dependencies, dpkt or scapy not installed') +class EnginePCAPTraceTests(unittest.TestCase): + """A traced flow is the same PCAP file whichever engine read the capture.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + tmp = tempfile.TemporaryDirectory() # pylint: disable=consider-using-with + self.addCleanup(tmp.cleanup) + self.tmp = tmp.name + + def _flows(self, engine: 'str', fin: 'str', *, nanosecond: 'bool' = False, + caught: 'list | None' = None) -> 'list[tuple[str, tuple, bytes]]': + """``(label, frame numbers, file octets)`` of every flow traced as PCAP. + + The warnings the extraction raised are appended to ``caught``, if given. + + """ + from pcapkit import extract + from pcapkit.utilities.warnings import FormatWarning + + fout = tempfile.mkdtemp(dir=self.tmp) + with warnings.catch_warnings(record=True) as raised: + warnings.simplefilter('always') + extractor = extract(fin=fin, nofile=True, engine=engine, tcp=True, trace=True, + trace_fout=fout, trace_format='pcap', trace_nanosecond=nanosecond) + self.addCleanup(close_extractor, extractor) + self.assertEqual([str(w.message) for w in raised if issubclass(w.category, FormatWarning)], []) + if caught is not None: + caught.extend(raised) + + flows = [] + for flow in extractor.trace.tcp: + self.assertTrue(flow.fpout.endswith('.pcap'), flow.fpout) + with open(flow.fpout, 'rb') as file: + flows.append((flow.label, tuple(flow.index), file.read())) + return flows + + def _assert_same_files(self, fin: 'str', *, nanosecond: 'bool', labels: 'bool' = True, + caught: 'list | None' = None) -> None: + base = self._flows('default', fin, nanosecond=nanosecond) + self.assertTrue(base) + for engine in ENGINES: + with self.subTest(engine=engine): + flows = self._flows(engine, fin, nanosecond=nanosecond, caught=caught) + if labels: + self.assertEqual([flow[:2] for flow in flows], [flow[:2] for flow in base]) + self.assertEqual([flow[1] for flow in flows], [flow[1] for flow in base]) + for (label, _, mine), (_, _, theirs) in zip(flows, base): + self.assertEqual(mine[:4], MAGIC[nanosecond], label) + self.assertEqual(mine, theirs, label) + + def _nanosecond_records(self, extra: 'int') -> 'list[list]': + """``tcp.pcap``'s records, each microsecond fraction made nanoseconds plus ``extra``.""" + _, records = _records(sample_path('tcp.pcap')) + return [[ts_sec, ts_usec * 1000 + extra, orig_len, octets] + for ts_sec, ts_usec, orig_len, octets in records] + + def test_traced_pcap_matches_the_default_engine_byte_for_byte(self) -> None: + for capture in CAPTURES: + for nanosecond in (False, True): + with self.subTest(capture=capture, nanosecond=nanosecond): + self._assert_same_files(sample_path(capture), nanosecond=nanosecond) + + def test_a_nanosecond_pcap_keeps_its_resolution(self) -> None: + # The record fraction stays in the capture's own resolution, as the default + # engine's does. Labels are left out: dpkt's Decimal timestamp spells them + # with nine places (#1501). + header, _ = _records(sample_path('tcp.pcap')) + for case, extra in EXTRA_NS.items(): + fin = _write(os.path.join(self.tmp, f'{case}.pcap'), MAGIC[True] + header[4:], + self._nanosecond_records(extra)) + for nanosecond in (False, True): + with self.subTest(case=case, nanosecond=nanosecond): + self._assert_same_files(fin, nanosecond=nanosecond, labels=False) + + def test_a_nanosecond_pcapng_keeps_its_resolution(self) -> None: + # ``if_tsresol=9``: an unaligned timestamp must keep its sub-microsecond + # digits, and an aligned one must still be read as nanoseconds, not as the + # microsecond count it has the digits of. + for case, extra in EXTRA_NS.items(): + fin = _write_pcapng(os.path.join(self.tmp, f'{case}.pcapng'), + self._nanosecond_records(extra), tsresol=9) + for nanosecond in (False, True): + with self.subTest(case=case, nanosecond=nanosecond): + self._assert_same_files(fin, nanosecond=nanosecond) + + def test_a_cross_resolution_trace_rescales_the_fraction(self) -> None: + # Not only the same bytes as the default engine: the right ones. The first + # frame of tcp.pcap is at 1500000000.000774; the synthetic nanosecond + # captures put it at 1500000000.000774789. A trace in the other resolution + # than its capture's scales the fraction, truncating nanoseconds (#1529). + header, _ = _records(sample_path('tcp.pcap')) + captures = ( + ('us pcap', sample_path('tcp.pcap'), {False: 774, True: 774000}), + ('ns pcap', _write(os.path.join(self.tmp, 'cross.pcap'), MAGIC[True] + header[4:], + self._nanosecond_records(789)), {False: 774, True: 774789}), + ('ns pcapng', _write_pcapng(os.path.join(self.tmp, 'cross.pcapng'), + self._nanosecond_records(789), tsresol=9), {False: 774, True: 774789}), + ) + for name, fin, fraction in captures: + for nanosecond, expected in fraction.items(): + for engine in ENGINES: + with self.subTest(capture=name, nanosecond=nanosecond, engine=engine): + first = min((_records(flow[2])[1][0] for flow in + self._flows(engine, fin, nanosecond=nanosecond)), + key=lambda record: record[:2]) + self.assertEqual(first[:2], [1500000000, expected]) + + def test_a_snapped_frame_keeps_its_original_length(self) -> None: + # Both engines' readers report a record's original length, PCAP and + # PCAP-NG alike, so a frame the snapshot length cut short is written as + # the default engine writes it. + header, records = _records(sample_path('tcp.pcap')) + longest = max(range(len(records)), key=lambda index: len(records[index][3])) + records[longest][3] = records[longest][3][:-4] + captures = { + 'pcap': _write(os.path.join(self.tmp, 'snapped.pcap'), header, records), + 'pcapng': _write_pcapng(os.path.join(self.tmp, 'snapped.pcapng'), + [[ts_sec, ts_usec * 1000, orig_len, octets] + for ts_sec, ts_usec, orig_len, octets in records], tsresol=6), + } + for container, fin in captures.items(): + with self.subTest(container=container): + self._assert_same_files(fin, nanosecond=False) + + def test_no_engine_warns_about_original_lengths(self) -> None: + # The frame record is complete, so there is nothing to warn about -- not on + # a PCAP trace, not on a JSON one, and not from follow_tcp_stream. + from pcapkit.interface.misc import follow_tcp_stream + + for engine in ENGINES: + for trace_format in ('pcap', 'json'): + with self.subTest(engine=engine, trace_format=trace_format), \ + warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + follow_tcp_stream(fin=sample_path('tcp.pcap'), engine=engine, format=trace_format, + fout=tempfile.mkdtemp(dir=self.tmp)) + self.assertEqual([str(w.message) for w in caught if 'original length' in str(w.message)], []) + + def test_dict_capable_trace_formats_are_unchanged(self) -> None: + # Only the PCAP trace dumper reads the frame record, so the JSON trace of + # each flow is the very file main wrote before #1507: the adapters still + # report the engine's own mapping of the frame, value types included. + # Re-measure the digests if dictdumper's output ever changes. + from pcapkit import extract + + for engine in ENGINES: + with self.subTest(engine=engine): + fout = tempfile.mkdtemp(dir=self.tmp) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + extractor = extract(fin=sample_path('tcp.pcap'), nofile=True, engine=engine, tcp=True, + trace=True, trace_fout=fout, trace_format='json') + self.addCleanup(close_extractor, extractor) + digests = {} + for flow in extractor.trace.tcp: + with open(flow.fpout, 'rb') as file: + digests[os.path.basename(flow.fpout)] = hashlib.sha256(file.read()).hexdigest() + self.assertEqual(digests, JSON_TRACE_SHA256[engine]) + + @unittest.skipUnless(HAS_SCAPY, 'scapy not installed') + def test_scapy_closes_its_reader_once(self) -> None: + # ``sniff`` closes the reader at the end of the file; a second close from + # the engine would raise on a gzip reader. + from scapy.utils import RawPcapReader + + close = RawPcapReader.close + calls = [] # type: list + + def counted(reader: 'RawPcapReader') -> None: + calls.append(reader) + close(reader) + + with mock.patch.object(RawPcapReader, 'close', counted): + self._flows('scapy', sample_path('tcp.pcap')) + self.assertEqual(len(calls), 1) + + # and if ``sniff`` raises before the end of the file, the engine closes it + import scapy.all + + from pcapkit import extract + + calls.clear() + with mock.patch.object(RawPcapReader, 'close', counted), \ + mock.patch.object(scapy.all, 'sniff', side_effect=RuntimeError('sniff failed')), \ + self.assertRaisesRegex(RuntimeError, 'sniff failed'): + extract(fin=sample_path('tcp.pcap'), nofile=True, engine='scapy') + self.assertEqual(len(calls), 1) + + @unittest.skipUnless(HAS_DPKT, 'dpkt not installed') + def test_dpkt_without_the_reader_internals_falls_back_and_warns(self) -> None: + # The engine's PCAP reader reads three private attributes of + # dpkt.pcap.Reader, and dpkt is not pinned. Under a dpkt that names them + # differently it reads with dpkt's own loop, losing only the original + # length, and says so naming the dpkt version. + import dpkt + + from pcapkit.utilities.warnings import DPKTWarning + + class Renamed(dpkt.pcap.Reader): # type: ignore[misc] + """dpkt.pcap.Reader with its private attributes under other names.""" + + def __init__(self, fileobj: 'Any') -> None: + super().__init__(fileobj) + self._state = (self._Reader__f, self._Reader__ph, self._divisor) + del self._Reader__f, self._Reader__ph, self._divisor + + def __iter__(self) -> 'Any': + file, header, divisor = self._state + while True: + buf = file.read(header.__hdr_len__) + if not buf: + break + hdr = header(buf) + yield hdr.tv_sec + (hdr.tv_usec / divisor), file.read(hdr.caplen) + + base = self._flows('default', sample_path('tcp.pcap')) + caught = [] # type: list + with mock.patch.object(dpkt.pcap, 'Reader', Renamed): + flows = self._flows('dpkt', sample_path('tcp.pcap'), caught=caught) + self.assertEqual(flows, base) + messages = [str(w.message) for w in caught if issubclass(w.category, DPKTWarning) + and 'original length' in str(w.message)] + self.assertEqual(len(messages), 1, messages) + self.assertIn(f'dpkt {dpkt.__version__}', messages[0]) + self.assertIn("'_Reader__f', '_Reader__ph', '_divisor'", messages[0]) + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class DPKTTimestampTests(unittest.TestCase): + """The timestamp carriers behave as the values they carry.""" + + def test_copies_and_arithmetic_keep_working(self) -> None: + import copy + import decimal + import pickle + import statistics + + from pcapkit.toolkit.dpkt import DecimalTimestamp, Timestamp + + for stamp in (Timestamp(1500000000.000774, decimal.Decimal('1500000000.000774'), 1_000_000, 78), + DecimalTimestamp(decimal.Decimal('1500000000.000774789'), 1_000_000_000, 78)): + with self.subTest(type=type(stamp).__name__): + # ``statistics`` rebuilds a mean through ``type(x)(value)`` + self.assertEqual(statistics.mean([stamp, stamp]), stamp) + for clone in (copy.deepcopy(stamp), pickle.loads(pickle.dumps(stamp))): + self.assertEqual(clone, stamp) + self.assertEqual((clone.exact, clone.resolution, clone.orig_len), + (stamp.exact, stamp.resolution, stamp.orig_len)) + + def test_a_frame_record_pickles_with_every_protocol(self) -> None: + import copy + import decimal + import pickle + + from pcapkit.foundation.traceflow.data.data import FrameRecord + from pcapkit.protocols.data.misc.pcap.frame import FrameInfo + + record = FrameRecord({'packet': b'rebuilt', 'Ethernet': {'type': 2048}}, packet=b'captured', + frame_info=FrameInfo(ts_sec=1, ts_usec=2, incl_len=8, orig_len=9), + time_epoch=decimal.Decimal('1.000002')) + clones = [pickle.loads(pickle.dumps(record, protocol=protocol)) + for protocol in range(pickle.HIGHEST_PROTOCOL + 1)] + [copy.deepcopy(record)] + for clone in clones: + self.assertIs(type(clone), FrameRecord) + self.assertEqual(clone, record) + self.assertEqual((clone.packet, clone.frame_info, clone.time_epoch), + (record.packet, record.frame_info, record.time_epoch)) + + +if __name__ == '__main__': + unittest.main()