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()