From da66f152edcf22f45b730f24c43fb44490ea6170 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Fri, 9 Oct 2026 18:56:52 -0400 Subject: [PATCH] fix(toolkit): the pypcapfile engine warns that it leaves TCP over IPv6 out (#1513) pypcapfile 0.12.0 has no IPv6 decoder: Ethernet.payload_type maps only 0x0800, so an IPv6 frame's payload stays hex-encoded and _network() returns None. tcp_reassembly and tcp_traceflow dropped those frames without a word. - Both adapters now warn (AttributeWarning) on the first frame of a capture that carries TCP over IPv6, directly or behind the extension headers the default engine steps over (RFC 8200's less ESP, plus Mobility, HIP and Shim6), once per capture rather than per frame. - Captures are keyed by the address of their savefile header, each pinned by an anchor; the last 32 warned about are remembered. - The toolkit, engine docstring and engines docs page state the loss; VLAN, QinQ and tunnelled frames stay out of reach, without a warning. - test_engine_agreement_runtime: the Gap(1513) row is gone. Where the engine warns, its tcp/trace inputs and TCP datagrams are compared with default's over IPv4 only, and its flows skip. Closes #1513 --- .../pcapkit/foundation/engines/index.rst | 4 +- pcapkit/foundation/engines/pypcapfile.py | 7 +- pcapkit/toolkit/pypcapfile.py | 187 +++++++++++++++- .../engines/test_engine_agreement_runtime.py | 49 +++-- tests/toolkit/test_pypcapfile_unit.py | 199 +++++++++++++++++- 5 files changed, 427 insertions(+), 19 deletions(-) diff --git a/docs/source/pcapkit/foundation/engines/index.rst b/docs/source/pcapkit/foundation/engines/index.rst index 6d15766bd4..e7172cec71 100644 --- a/docs/source/pcapkit/foundation/engines/index.rst +++ b/docs/source/pcapkit/foundation/engines/index.rst @@ -92,7 +92,9 @@ offers, and the ones that cannot say so rather than quietly doing less: | :class:`~pcapkit.foundation.engines.pypcapfile.PyPCAPFile` | no IPv6 decoder, hence no IPv6 reassembly -- disabled with an | | | :class:`~pcapkit.utilities.warnings.AttributeWarning`, and | | | :func:`~pcapkit.toolkit.pypcapfile.ipv6_reassembly` raises; | -| | PCAP savefiles only, otherwise | +| | TCP over IPv6 is left out of TCP reassembly and flow tracing, | +| | with an :class:`~pcapkit.utilities.warnings.AttributeWarning` | +| | once per capture; PCAP savefiles only, otherwise | | | :exc:`~pcapkit.utilities.exceptions.FormatError` | +-----------------------------------------------------------------+---------------------------------------------------------------+ | :class:`~pcapkit.foundation.engines.dpkt.DPKT`, | no PCAP output, since each frame reaches the writer as a | diff --git a/pcapkit/foundation/engines/pypcapfile.py b/pcapkit/foundation/engines/pypcapfile.py index 0826ad98a0..404c56ba6f 100644 --- a/pcapkit/foundation/engines/pypcapfile.py +++ b/pcapkit/foundation/engines/pypcapfile.py @@ -40,9 +40,12 @@ class PyPCAPFile(EngineBase['PCAPFilePacket']): no PCAP-NG support. Consequently this engine * reads PCAP savefiles only, raising - :exc:`~pcapkit.utilities.exceptions.FormatError` on PCAP-NG, and + :exc:`~pcapkit.utilities.exceptions.FormatError` on PCAP-NG, * disables IPv6 reassembly, warning as it does so, while leaving IPv4 and - TCP reassembly and TCP flow tracing in place. + TCP reassembly and TCP flow tracing in place, and + * leaves TCP over IPv6 out of TCP reassembly and flow tracing, with one + :exc:`~pcapkit.utilities.warnings.AttributeWarning` per capture -- see + :mod:`pcapkit.toolkit.pypcapfile`. The engine stops decoding at the network layer rather than descending into the transport layer. `PyPCAPFile`_ decoders *replace* the payload bytes of diff --git a/pcapkit/toolkit/pypcapfile.py b/pcapkit/toolkit/pypcapfile.py index bdcce84309..9d69fe3b91 100644 --- a/pcapkit/toolkit/pypcapfile.py +++ b/pcapkit/toolkit/pypcapfile.py @@ -19,6 +19,17 @@ instead of quietly returning :data:`None`, which would be indistinguishable from "this frame carries no IPv6 fragment". + For the same reason :func:`tcp_reassembly` and :func:`tcp_traceflow` see TCP + over IPv4 only, and decline a frame carrying TCP over IPv6. That loss is not + silent: it is reported with one + :exc:`~pcapkit.utilities.warnings.AttributeWarning` per capture, on its first + such frame, rather than one per frame. + + That warning covers IPv6 straight over Ethernet only. `PyPCAPFile`_ decodes + no VLAN tag (802.1Q), no QinQ (802.1ad) and no tunnel (e.g. IPv6 in IPv4, or + GRE), so TCP inside any of them is left out without a warning, over IPv6 and + IPv4 alike. + Note also that `PyPCAPFile`_ decoders *replace* the payload bytes of the layer they decode. :class:`~pcapkit.foundation.engines.pypcapfile.PyPCAPFile` therefore stops at the network layer, which keeps the transport segment verbatim so that @@ -29,19 +40,25 @@ """ import binascii +import collections +import ctypes import ipaddress import struct import sys import textwrap from typing import TYPE_CHECKING +from pcapkit.const.ipv6.extension_header import ExtensionHeader as Enum_ExtensionHeader +from pcapkit.const.reg.ethertype import EtherType as Enum_EtherType 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.tcp import Packet as TF_TCP_Packet -from pcapkit.utilities.exceptions import ProtocolError, UnsupportedCall +from pcapkit.utilities.exceptions import ProtocolError, UnsupportedCall, stacklevel +from pcapkit.utilities.warnings import AttributeWarning, warn if TYPE_CHECKING: + from collections import OrderedDict from ipaddress import IPv4Address, IPv6Address from typing import Any, Optional @@ -69,6 +86,32 @@ #: IPv4 **MF** (more fragments) bit, within the three-bit flags field. IPV4_FLAG_MF = 0b001 +#: Length of the fixed IPv6 header, :rfc:`8200#section-3`. +IPV6_HEADER_LEN = 40 + +#: IPv6 extension headers that :func:`_ipv6_carries_tcp` steps over on its way +#: to the upper-layer protocol, as the default engine does: the ones +#: :rfc:`8200#section-4.1` orders before it, less ESP, whose payload is +#: encrypted, plus the Mobility, HIP and Shim6 headers. Those three are sized in +#: 8-octet units like most of the rest, and the default engine dissects whatever +#: protocol their next header field names. The experimental 253 and 254 are left +#: out, because the default engine dissects nothing after them. +IPV6_EXTENSION_HEADERS = frozenset(( + Enum_ExtensionHeader.HOPOPT, Enum_ExtensionHeader.IPv6_Opts, Enum_ExtensionHeader.IPv6_Route, + Enum_ExtensionHeader.IPv6_Frag, Enum_ExtensionHeader.AH, Enum_ExtensionHeader.Mobility_Header, + Enum_ExtensionHeader.HIP, Enum_ExtensionHeader.Shim6, +)) + +#: How many captures :func:`_decline_ipv6_tcp` remembers having warned about. +#: Each one pins a packet of its own in memory (see :func:`_capture_of`), so the +#: memory is bounded and the oldest is forgotten first. Up to this many captures +#: read at the same time warn exactly once each. +IPV6_TCP_WARNED_LIMIT = 32 + +#: The captures :func:`_decline_ipv6_tcp` has warned about, oldest first. Each +#: key :func:`_capture_of` gives maps to the anchor that keeps that key unique. +_ipv6_tcp_warned = collections.OrderedDict() # type: OrderedDict[int, Any] + def packet2timestamp(packet: 'Packet') -> 'float': """Calculate the timestamp of a PyPCAPFile packet. @@ -284,6 +327,124 @@ def _network(packet: 'Packet') -> 'Optional[IP]': return payload +def _ipv6_carries_tcp(packet: 'Packet') -> 'bool': + """Test if a PyPCAPFile frame is an IPv6 packet carrying TCP. + + Args: + packet: PyPCAPFile packet. + + Returns: + :data:`True` if the frame is IPv6 straight over Ethernet and its + upper-layer protocol is TCP, whether directly or behind the extension + headers in :data:`IPV6_EXTENSION_HEADERS`. A VLAN-tagged, QinQ or + tunnelled frame is not looked into, so it never counts. + A first fragment is read past its Fragment header as if that header + were not there. A later fragment never counts, since from its offset on + it carries a slice of the payload and no header (:rfc:`8200#section-4.5`). + An ESP payload cannot be seen into, so it does not count, and nor does a + truncated header chain or, as for IPv4 (see :func:`_transport`), a TCP + segment too short to hold a header. + + Note: + `PyPCAPFile`_ leaves an IPv6 frame's payload undecoded and hex-encoded + (see :func:`_maybe_unhex`), so this reads the raw header chain. It + reads only as far as the upper-layer protocol number and decodes + nothing. + + Each of these rules is the default engine's own, so that the warning + fires on the frames the default engine finds TCP in. + + .. _PyPCAPFile: https://github.com/kisom/pypcapfile + + """ + link = packet.packet + if _is_raw(link) or getattr(link, 'type', None) != Enum_EtherType.Internet_Protocol_version_6: + return False + + payload = getattr(link, 'payload', None) # type: Any + if not _is_raw(payload): + return False + + data = _maybe_unhex(bytes(payload)) + if len(data) < IPV6_HEADER_LEN: + return False + + nxt, offset = data[6], IPV6_HEADER_LEN + while nxt in IPV6_EXTENSION_HEADERS: + if len(data) < offset + 8: + return False + if nxt == Enum_ExtensionHeader.IPv6_Frag: # fixed length, second octet reserved + if struct.unpack_from('!H', data, offset + 2)[0] >> 3: # a later fragment + return False + length = 8 + elif nxt == Enum_ExtensionHeader.AH: # length in 4-octet units, less two + length = (data[offset + 1] + 2) * 4 + else: # length in 8-octet units, less one + length = (data[offset + 1] + 1) * 8 + nxt, offset = data[offset], offset + length + return nxt == Enum_TransType.TCP and len(data) >= offset + TCP_MIN_HEADER_LEN + + +def _capture_of(packet: 'Packet') -> 'tuple[int, Any]': + """Identify the capture a PyPCAPFile packet was read from. + + Args: + packet: PyPCAPFile packet. + + Returns: + A key, equal for every packet of one capture, and an anchor: while the + anchor is alive, no packet of another capture has the same key. + + Note: + :attr:`pcap_packet.header ` is a + :mod:`ctypes` pointer to the savefile header. Each read of the field + builds a new pointer object, but every one of them points at the same + header, so the header's address is the key. The pointer object read + here is the anchor: it keeps its packet alive and, through that packet, + the header itself, so no other capture's header can be allocated at + that address while it is held. A header that is not a :mod:`ctypes` + pointer is keyed by its identity instead, and anchors itself. + + """ + anchor = packet.header + try: + return ctypes.addressof(anchor.contents), anchor + except (AttributeError, TypeError, ValueError): + return id(anchor), anchor + + +def _decline_ipv6_tcp(packet: 'Packet', count: 'int') -> 'None': + """Warn that a frame carrying TCP over IPv6 is left out, once per capture. + + Args: + packet: PyPCAPFile packet, which :func:`_network` found no IPv4 layer in. + count: Packet index. + + Warns: + AttributeWarning: If ``packet`` carries TCP over IPv6 and is the first + such frame of its capture to reach :func:`tcp_reassembly` or + :func:`tcp_traceflow`. Later frames of the same capture are left out + without another warning, and every other capture warns once of its + own, however their frames interleave. Only the + :data:`IPV6_TCP_WARNED_LIMIT` captures warned about most recently + are remembered, so with more than that read at once a forgotten one + warns again: the warning can repeat, but is never omitted. + + """ + key, anchor = _capture_of(packet) + if key in _ipv6_tcp_warned: + return + if not _ipv6_carries_tcp(packet): + return + + _ipv6_tcp_warned[key] = anchor + while len(_ipv6_tcp_warned) > IPV6_TCP_WARNED_LIMIT: + _ipv6_tcp_warned.popitem(last=False) + warn(f"Frame {count}: TCP over IPv6 is left out of TCP reassembly and flow tracing, as " + "'pypcapfile' has no IPv6 decoder; later such frames of this capture are left out " + 'without another warning', AttributeWarning, stacklevel=stacklevel()) + + def _transport(ipv4: 'IP') -> 'Optional[bytes]': """Fetch the verbatim TCP segment carried by a PyPCAPFile IPv4 packet. @@ -525,7 +686,8 @@ def tcp_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'TCP_Packet | None Data for TCP reassembly. * If the ``packet`` can be used for TCP reassembly. A packet can be reassembled - if it contains an IPv4 layer carrying a TCP segment. + if it contains an IPv4 layer carrying a TCP segment; TCP over IPv6 cannot + be, as `PyPCAPFile`_ has no IPv6 decoder. * If the ``packet`` can be reassembled, then the :obj:`dict` mapping of data for TCP reassembly (:term:`reasm.tcp.packet`) will be returned; otherwise, returns :data:`None`. @@ -533,12 +695,21 @@ def tcp_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'TCP_Packet | None ProtocolError: If ``ipv4.src`` or ``ipv4.dst`` cannot be parsed as an IPv4 address. + Warns: + AttributeWarning: If ``packet`` carries TCP over IPv6, which is left out, + and is the first such frame of its capture to reach either + :func:`tcp_reassembly` or :func:`tcp_traceflow`. The rest of that + capture's such frames are left out without another warning. + See Also: :class:`pcapkit.foundation.reassembly.tcp.TCP` + .. _PyPCAPFile: https://github.com/kisom/pypcapfile + """ ipv4 = _network(packet) if ipv4 is None: + _decline_ipv6_tcp(packet, count) return None segment = _transport(ipv4) @@ -588,7 +759,8 @@ def tcp_traceflow(packet: 'Packet', *, data_link: 'Enum_LinkType', Data for TCP flow tracing. * If the ``packet`` can be used for TCP flow tracing. A packet can be traced - if it contains an IPv4 layer carrying a TCP segment. + if it contains an IPv4 layer carrying a TCP segment; TCP over IPv6 cannot + be, as `PyPCAPFile`_ has no IPv6 decoder. * If the ``packet`` can be traced, then the :obj:`dict` mapping of data for TCP flow tracing (:term:`trace.tcp.packet`) will be returned; otherwise, returns :data:`None`. @@ -596,12 +768,21 @@ def tcp_traceflow(packet: 'Packet', *, data_link: 'Enum_LinkType', ProtocolError: If ``ipv4.src`` or ``ipv4.dst`` cannot be parsed as an IPv4 address. + Warns: + AttributeWarning: If ``packet`` carries TCP over IPv6, which is left out, + and is the first such frame of its capture to reach either + :func:`tcp_reassembly` or :func:`tcp_traceflow`. The rest of that + capture's such frames are left out without another warning. + See Also: :class:`pcapkit.foundation.traceflow.tcp.TCP` + .. _PyPCAPFile: https://github.com/kisom/pypcapfile + """ ipv4 = _network(packet) if ipv4 is None: + _decline_ipv6_tcp(packet, count) return None segment = _transport(ipv4) diff --git a/tests/foundation/engines/test_engine_agreement_runtime.py b/tests/foundation/engines/test_engine_agreement_runtime.py index 8cbfb67ee9..0689151950 100644 --- a/tests/foundation/engines/test_engine_agreement_runtime.py +++ b/tests/foundation/engines/test_engine_agreement_runtime.py @@ -43,7 +43,11 @@ with a :exc:`~pcapkit.utilities.exceptions.FormatError`, and an engine that declines reassembly or flow tracing, or one protocol of it, warns and leaves it unset (``pypcap`` and ``pcap_ct`` decline both, ``pyshark`` reassembly, -``pypcapfile`` IPv6). Any other exception is an ``ERROR``. +``pypcapfile`` IPv6). ``pypcapfile`` also warns when a capture carries TCP over +IPv6, which it leaves out of TCP reassembly and flow tracing: on such a capture +its ``tcp`` and ``trace`` inputs and TCP ``datagrams`` are compared with the +``default`` engine's over IPv4 only, and its ``flows`` skip. Any other exception +is an ``ERROR``. This module reads generated captures, so it is in the fixture-dependent tier (see :mod:`tests._tiers`) and runs after ``make samples``. @@ -53,6 +57,7 @@ from __future__ import annotations import importlib.util +import ipaddress import os import tempfile import unittest @@ -140,9 +145,11 @@ def __init__(self, engine: 'str', capture: 'str') -> None: from pcapkit import extract from pcapkit.foundation.reassembly.reassembly import ReassemblyBase from pcapkit.foundation.traceflow.traceflow import TraceFlowBase - from pcapkit.utilities.warnings import EngineWarning + from pcapkit.utilities.warnings import AttributeWarning, EngineWarning self.fallback = None # type: Optional[str] + #: The ``pypcapfile`` toolkit's warning that it leaves TCP over IPv6 out. + self.tcp_declined = None # type: Optional[str] self.inputs = {'ipv4': {}, 'ipv6': {}, 'tcp': {}, 'trace': {}} # type: dict[str, dict[int, Any]] inputs = self.inputs reassemble, trace = ReassemblyBase.__call__, TraceFlowBase.__call__ @@ -158,7 +165,8 @@ def spy_trace(this: 'Any', packet: 'Any') -> 'None': with warnings.catch_warnings(), tempfile.TemporaryDirectory() as tracedir, \ mock.patch.object(ReassemblyBase, '__call__', spy_reassembly), \ mock.patch.object(TraceFlowBase, '__call__', spy_trace), \ - mock.patch('pcapkit.foundation.extraction.warn') as warn: + mock.patch('pcapkit.foundation.extraction.warn') as warn, \ + mock.patch('pcapkit.toolkit.pypcapfile.warn') as toolkit_warn: warnings.simplefilter('ignore') fallen = [] # type: list[str] try: @@ -176,6 +184,7 @@ def spy_trace(this: 'Any', packet: 'Any') -> 'None': if fallen or extractor._exnam != engine: # pylint: disable=protected-access self.fallback = '; '.join(fallen) or f'ran as {extractor._exnam}' # pylint: disable=protected-access return + self.tcp_declined = '; '.join(self._fallen(toolkit_warn, AttributeWarning)) or None self.length = extractor.length self.records = [_record(engine, frame) for frame in extractor.frame] # An engine that declines a protocol says so with a warning and leaves @@ -255,6 +264,25 @@ def _first_difference(got: 'dict[int, Any]', want: 'dict[int, Any]') -> 'Optiona return None +def _over_ipv4(record: 'dict[str, Any]') -> 'bool': + """Whether a TCP trace or reassembly input, or a TCP datagram, is over IPv4.""" + if 'src' in record: + address = record['src'] + elif 'bufid' in record: + address = record['bufid'][0] + else: + address = record['id'].src[0] + return isinstance(address, ipaddress.IPv4Address) + + +def _ipv4_only(base: '_Run', aspect: 'str') -> 'Any': + """``base``'s ``tcp`` or ``trace`` inputs, or its TCP datagrams, over IPv4 only.""" + if aspect == 'datagrams': + return [datagram for datagram in base.datagrams['tcp'] if _over_ipv4(datagram)] + return {number: records for number, records in base.inputs[aspect].items() + if _over_ipv4(records[0])} + + def compare(engine: 'str', capture: 'str', aspect: 'str') -> 'Outcome': run = _run(engine, capture) if run.fallback is not None: @@ -263,6 +291,8 @@ def compare(engine: 'str', capture: 'str', aspect: 'str') -> 'Outcome': if 'reads PCAP savefiles only' in run.error: # a documented refusal, not a reading return Outcome('SKIP', f'{engine} refuses {capture}: {run.error}') return Outcome('ERROR', run.error) + if run.tcp_declined is not None and aspect == 'flows': # the rest narrow to IPv4 below + return Outcome('SKIP', f'{engine} warns: {run.tcp_declined}') base = _run('default', capture) if aspect in ('ipv4', 'ipv6', 'tcp') and run.datagrams[aspect] is None: return Outcome('SKIP', f'{engine} declines {aspect} reassembly') @@ -285,6 +315,8 @@ def compare(engine: 'str', capture: 'str', aspect: 'str') -> 'Outcome': return Outcome('OK') if aspect in run.inputs: mine, theirs = run.inputs[aspect], base.inputs[aspect] + if run.tcp_declined is not None and aspect in ('tcp', 'trace'): + theirs = _ipv4_only(base, aspect) if aspect == 'trace' and engine in NO_TCP_OCTETS: mine, theirs = _without_tcp_octets(mine), _without_tcp_octets(theirs) diff = _first_difference(mine, theirs) @@ -295,6 +327,8 @@ def compare(engine: 'str', capture: 'str', aspect: 'str') -> 'Outcome': return Outcome('SKIP', f'{engine} declines reassembly') for kind in kinds: mine, theirs = run.datagrams[kind], base.datagrams[kind] + if kind == 'tcp' and run.tcp_declined is not None: + theirs = _ipv4_only(base, 'datagrams') if mine != theirs: if len(mine) != len(theirs): return Outcome('DATAGRAMS', f'{kind}: {len(mine)} datagrams != {len(theirs)}') @@ -340,15 +374,6 @@ class TestEngineAgreement(harness.RoundTripBase): tuple(f'{capture}/pypcapfile/{aspect}' for capture in ('big_endian.pcap', 'big_endian_nanosecond.pcap') for aspect in ('frames', 'timestamps', 'ipv4', 'datagrams'))), - Gap(1513, 'the pypcapfile engine leaves TCP over IPv6 out of TCP reassembly and flow ' - 'tracing, without a warning: _network returns None for any frame that is not ' - 'IPv4 (pcapkit/toolkit/pypcapfile.py:262), so tcp_reassembly (:540) and ' - 'tcp_traceflow (:603) skip it; only IPv6 reassembly is declined with a warning ' - '(pcapkit/foundation/engines/pypcapfile.py:229)', - ('INPUTS', 'DATAGRAMS', 'FLOWS'), ('missing, default has', 'tcp: ', 'flows ['), - tuple(f'{capture}/pypcapfile/{aspect}' - for capture in ('http6.cap', 'stream.pcap', 'tcp.pcap', 'test.pcap') - for aspect in ('tcp', 'trace', 'datagrams', 'flows'))), Gap(1501, 'the dpkt engine hands reassembly the Decimal timestamp dpkt reads from a ' 'nanosecond PCAP (pcapkit/foundation/engines/dpkt.py:309), where every other ' 'engine hands it a float (pcapkit/toolkit/pcap.py:73)', diff --git a/tests/toolkit/test_pypcapfile_unit.py b/tests/toolkit/test_pypcapfile_unit.py index dc3630a37a..9ea26a00f3 100644 --- a/tests/toolkit/test_pypcapfile_unit.py +++ b/tests/toolkit/test_pypcapfile_unit.py @@ -14,13 +14,18 @@ from __future__ import annotations import binascii +import ctypes import importlib.util import ipaddress +import os import struct +import tempfile import types import unittest +import warnings -from tests._support import reimport_once_per_class +from tests._support import close_extractor, reimport_once_per_class +from tests.foundation import _roundtrip as wire RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) @@ -135,6 +140,17 @@ def make_packet(layer, *, timestamp: int = 1511106545, timestamp_us: int = 47171 ) +def ipv6_frame(packet: bytes) -> FakeEthernet: + """An Ethernet frame carrying IPv6, as PyPCAPFile leaves it: payload hex-encoded.""" + frame = FakeEthernet(binascii.hexlify(packet)) + frame.type = 0x86DD + return frame + + +#: A TCP segment with a full header and a payload. +TCP_SEGMENT = wire.tcp(b'payload', seq=1, syn=True) + + @unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') class PyPCAPFileToolkitTests(unittest.TestCase): def setUp(self) -> None: @@ -382,6 +398,151 @@ def test_tcp_adapters_decline_non_tcp_and_truncated_segments(self) -> None: self.assertIsNone(tcp_traceflow(packet, data_link=LinkType.ETHERNET, count=1)) +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class PyPCAPFileTCPOverIPv6Tests(unittest.TestCase): + """TCP over IPv6 is left out, with one warning per capture. C.f. #1513.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def decline(self, *packets) -> list[str]: + """Feed ``packets`` to both TCP adapters, frame 1 first; return their warnings.""" + from pcapkit.const.reg.linktype import LinkType + from pcapkit.toolkit.pypcapfile import tcp_reassembly, tcp_traceflow + from pcapkit.utilities.warnings import AttributeWarning + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + for count, packet in enumerate(packets, start=1): + self.assertIsNone(tcp_reassembly(packet, count=count)) + self.assertIsNone(tcp_traceflow(packet, data_link=LinkType.ETHERNET, count=count)) + return [str(item.message) for item in caught if issubclass(item.category, AttributeWarning)] + + def test_tcp_over_ipv6_warns_once_per_capture(self) -> None: + first, second = (make_packet(ipv6_frame(wire.ipv6(TCP_SEGMENT, nxt=6))) for _ in range(2)) + second.header = first.header # both frames of one capture + + with self.assertLogs('pcapkit', level='WARNING') as logs: + messages = self.decline(first, second) + self.assertEqual(len(messages), 1, messages) + self.assertTrue(messages[0].startswith('Frame 1: TCP over IPv6 is left out'), messages) + self.assertIn("'pypcapfile' has no IPv6 decoder", messages[0]) + self.assertEqual(len(logs.records), 1) + + def test_interleaved_captures_each_warn_once(self) -> None: + one, other, one_again, other_again = ( + make_packet(ipv6_frame(wire.ipv6(TCP_SEGMENT, nxt=6))) for _ in range(4)) + one_again.header, other_again.header = one.header, other.header + + messages = self.decline(one, other, one_again, other_again) + self.assertEqual([message.split(':')[0] for message in messages], ['Frame 1', 'Frame 2']) + + def test_the_oldest_capture_is_forgotten_beyond_the_limit(self) -> None: + from unittest import mock + + from pcapkit.toolkit import pypcapfile as toolkit + + first, second, third, first_again, third_again = ( + make_packet(ipv6_frame(wire.ipv6(TCP_SEGMENT, nxt=6))) for _ in range(5)) + first_again.header, third_again.header = first.header, third.header + + # Two remembered at most: the third capture pushes the first out, so the + # first warns a second time -- repeated, never omitted -- while the + # third is still remembered. + with mock.patch.object(toolkit, 'IPV6_TCP_WARNED_LIMIT', 2), \ + mock.patch.object(toolkit, '_ipv6_tcp_warned', type(toolkit._ipv6_tcp_warned)()): + messages = self.decline(first, second, third, first_again, third_again) + self.assertEqual(len(toolkit._ipv6_tcp_warned), 2) + self.assertEqual([message.split(':')[0] for message in messages], + ['Frame 1', 'Frame 2', 'Frame 3', 'Frame 4']) + + def test_tcp_behind_extension_headers_warns(self) -> None: + hop_by_hop = bytes([6, 0]) + b'\x01\x04' + bytes(4) # PadN, 8 octets + long_hop_by_hop = bytes([6, 1]) + b'\x01\x0c' + bytes(12) # PadN, 16 octets + options_then_routing = bytes([43, 0]) + b'\x01\x04' + bytes(4) + bytes([6, 0]) + bytes(6) + auth = struct.pack('!BBHII', 6, 4, 0, 0x100, 1) + bytes(12) # (4 + 2) * 4 octets + mobility = bytes([6, 1]) + bytes(14) # payload proto, 16 octets + hip = bytes([6, 4, 0x01, 0x21]) + bytes(36) # next header, 40 octets + shim6 = bytes([6, 0]) + bytes(6) # next header, 8 octets + for label, packet in ( + ('hop-by-hop', wire.ipv6(hop_by_hop + TCP_SEGMENT, nxt=0)), + ('16-octet hop-by-hop', wire.ipv6(long_hop_by_hop + TCP_SEGMENT, nxt=0)), + ('destination options, routing', wire.ipv6(options_then_routing + TCP_SEGMENT, nxt=60)), + ('authentication header', wire.ipv6(auth + TCP_SEGMENT, nxt=51)), + ('first fragment', wire.ipv6_fragment(TCP_SEGMENT, mf=True, nxt=6)), + ('mobility', wire.ipv6(mobility + TCP_SEGMENT, nxt=135)), + ('hip', wire.ipv6(hip + TCP_SEGMENT, nxt=139)), + ('shim6', wire.ipv6(shim6 + TCP_SEGMENT, nxt=140)), + ('hop-by-hop, mobility', wire.ipv6(bytes([135, 0]) + bytes(6) + mobility + TCP_SEGMENT, + nxt=0)), + ('first fragment, shim6', wire.ipv6_fragment(shim6 + TCP_SEGMENT, mf=True, nxt=140)), + ): + with self.subTest(case=label): + messages = self.decline(make_packet(ipv6_frame(packet))) + self.assertEqual(len(messages), 1, messages) + + def test_no_warning_without_tcp_over_ipv6(self) -> None: + arp = FakeEthernet(binascii.hexlify(b'\x00\x01\x08\x00' + bytes(24))) + arp.type = 0x0806 + decoded = ipv6_frame(b'') + decoded.payload = object() # a decoder PyPCAPFile does not have yet + for label, layer in ( + ('undecoded link', b'raw'), + ('arp', arp), + ('ipv6 payload decoded', decoded), + ('icmpv6', ipv6_frame(wire.ipv6(b'\x80\x00' + bytes(6), nxt=58))), + ('udp', ipv6_frame(wire.ipv6(wire.udp(b'x'), nxt=17))), + ('esp', ipv6_frame(wire.ipv6(b'\x00\x00\x01\x00' + bytes(4) + TCP_SEGMENT, nxt=50))), + # the header's own length runs to the end of the packet, so nothing + # follows it -- as in options-internet.pcap's HIP frames + ('hip filling the packet', ipv6_frame(wire.ipv6(bytes([6, 6, 0x01, 0x21]) + bytes(52), + nxt=139))), + ('mobility, no next header', ipv6_frame(wire.ipv6(bytes([59, 0]) + bytes(6) + + TCP_SEGMENT, nxt=135))), + ('experimental 253', ipv6_frame(wire.ipv6(bytes([6, 0]) + bytes(6) + TCP_SEGMENT, + nxt=253))), + ('truncated ipv6 header', ipv6_frame(wire.ipv6(b'', nxt=6)[:39])), + ('truncated extension header', ipv6_frame(wire.ipv6(b'\x06\x00\x01', nxt=0))), + ('truncated tcp header', ipv6_frame(wire.ipv6(TCP_SEGMENT[:19], nxt=6))), + ('extension header overruns the frame', + ipv6_frame(wire.ipv6(bytes([6, 200]) + bytes(6) + TCP_SEGMENT, nxt=0))), + # a later fragment's data is a slice of the payload, not a header, + # even when it happens to look like one + ('later fragment', ipv6_frame(wire.ipv6_fragment(TCP_SEGMENT, offset=8, nxt=6))), + ('later fragment, shim6', + ipv6_frame(wire.ipv6_fragment(bytes([6, 0]) + bytes(6) + TCP_SEGMENT, offset=8, + nxt=140))), + ('first fragment, then a later one', ipv6_frame(wire.ipv6_fragment( + struct.pack('!BBHI', 6, 0, 8, 7) + TCP_SEGMENT, mf=True, nxt=44))), + ): + with self.subTest(case=label): + self.assertEqual(self.decline(make_packet(layer)), []) + + def test_capture_of_keys_a_ctypes_header_by_its_address(self) -> None: + from pcapkit.toolkit.pypcapfile import _capture_of + + header, other = ctypes.c_int(1), ctypes.c_int(2) + # Every read of a ``ctypes`` pointer field builds a new pointer object, + # so the key must not be the pointer's identity. + one, same = make_packet(b''), make_packet(b'') + one.header, same.header = ctypes.pointer(header), ctypes.pointer(header) + another = make_packet(b'') + another.header = ctypes.pointer(other) + + key, anchor = _capture_of(one) + self.assertIs(anchor, one.header) + self.assertEqual(key, ctypes.addressof(header)) + self.assertEqual(_capture_of(same)[0], key) + self.assertNotEqual(_capture_of(another)[0], key) + + for label, value in (('null pointer', ctypes.POINTER(ctypes.c_int)()), + ('stand-in list', [types.SimpleNamespace(ns_resolution=False)])): + with self.subTest(case=label): + packet = make_packet(b'') + packet.header = value + self.assertEqual(_capture_of(packet), (id(value), value)) + + @unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') @unittest.skipUnless(HAS_PYPCAPFILE, 'pypcapfile not installed or not importable') class PyPCAPFileToolkitAgainstRealDecodersTests(unittest.TestCase): @@ -524,6 +685,42 @@ def test_ethernet2dict_reports_the_default_engines_mac_format(self) -> None: self.assertEqual(info['dst'], '01:00:5e:01:03:03') self.assertEqual(info['src'], '00:0c:29:3f:1a:07') + def test_engine_warns_once_per_capture_on_tcp_over_ipv6(self) -> None: + from pcapkit import extract + + # One IPv4 TCP frame, then two IPv6 TCP frames and an IPv6 UDP one. The + # engine builds a new ``pcap_packet`` per frame, so this also checks that + # the real savefile header keys one capture -- see ``_capture_of``. + frames = ( + wire.ethernet(wire.ipv4(TCP_SEGMENT, proto=6), 0x0800), + wire.ethernet(wire.ipv6(TCP_SEGMENT, nxt=6), 0x86DD), + wire.ethernet(wire.ipv6(wire.tcp(b'more', seq=8, ack=1), nxt=6), 0x86DD), + wire.ethernet(wire.ipv6(wire.udp(b'x'), nxt=17), 0x86DD), + ) + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, 'mixed.pcap') + with open(path, 'wb') as file: + file.write(wire.pcap([(1, number, frame) for number, frame in enumerate(frames)])) + + for read in ('first', 'second'): # reading it again is another capture + with self.subTest(read=read), warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + extractor = extract(fin=path, nofile=True, engine='pypcapfile', + reassembly=True, tcp=True, trace=True, + trace_fout=os.path.join(tmp, read), trace_format='json') + try: + messages = [str(item.message) for item in caught + if 'TCP over IPv6' in str(item.message)] + self.assertEqual(extractor._exnam, 'pypcapfile') + self.assertEqual(len(messages), 1, messages) + self.assertTrue(messages[0].startswith('Frame 2: '), messages) + # ...while the IPv4 frame is still reassembled and traced + self.assertEqual([tuple(flow.index) for flow in extractor.trace.tcp], + [(1,)]) + self.assertEqual(len(extractor.reassembly.tcp), 1) + finally: + close_extractor(extractor) + if __name__ == '__main__': unittest.main()