From a532960ccd10933bc5d09fa75b55dc1cde3da8c2 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Sat, 10 Oct 2026 08:01:25 -0400 Subject: [PATCH] fix(toolkit): pypcapfile checks IPv4 and TCP headers as the default engine does (#1590, #1591, #1592, #1593, #1596, #1597) pcapfile 0.12.0 slices an IPv4 header at its IHL without a bounds check, never parses IPv4 options, and has no IPv6 or AH decoder, and the toolkit read a TCP Data Offset below 5 as 20 octets, so its inputs disagreed with the default engine's. - A TCP Data Offset below 5 words is no TCP (#1590). - An IPv4 header whose IHL runs past the packet is no IPv4 (#1591). - IPv4 options are judged by the default engine's own IPv4 parser, given the header alone and once per frame; a header it rejects is no IPv4 and no TCP (#1596). - ipv4_reassembly() finds IPv4 carried in IPv6, as the default engine takes a frame's first IPv4 layer; TCP inside it is still a tunnel (#1592). - TCP behind IPv4 AH headers is read past them (#1593), and behind the IPv6 extension headers the default engine dissects inside IPv4, each read by its own parser (#1597). There the TCP header must be one the default TCP parser accepts; a tunnel behind them warns. Closes #1590 Closes #1591 Closes #1592 Closes #1593 Closes #1596 Closes #1597 --- pcapkit/toolkit/pypcapfile.py | 302 ++++++++++++++++-- .../test_pypcapfile_header_checks_unit.py | 258 +++++++++++++++ tests/toolkit/test_pypcapfile_unit.py | 205 +++++++++++- 3 files changed, 739 insertions(+), 26 deletions(-) create mode 100644 tests/toolkit/test_pypcapfile_header_checks_unit.py diff --git a/pcapkit/toolkit/pypcapfile.py b/pcapkit/toolkit/pypcapfile.py index 0150f26247..e2f090c572 100644 --- a/pcapkit/toolkit/pypcapfile.py +++ b/pcapkit/toolkit/pypcapfile.py @@ -35,7 +35,16 @@ :exc:`~pcapkit.utilities.warnings.AttributeWarning` of its own, once per capture. Either warning follows the default engine's own rules for where it finds TCP, read off the raw headers, with one stated exception for a - malformed tunnelled frame (see :func:`_upper_layer`). + malformed tunnelled frame (see :func:`_upper_layer`). IPv4 carried in IPv6 + (4in6) still reaches :func:`ipv4_reassembly`, which the default engine + gives the first IPv4 layer of a frame (see :func:`_ipv4_in_ipv6`), and TCP + behind AH or an IPv6 extension header inside IPv4 is read past them (see + :func:`_transport`). + + `PyPCAPFile`_ checks neither that an IPv4 header was captured whole nor its + options, and the default engine rejects a header that fails either check, so + such a header is left out of reassembly and flow tracing alike (see + :func:`_ipv4_accepted`). Note also that `PyPCAPFile`_ decoders *replace* the payload bytes of the layer they decode. :class:`~pcapkit.foundation.engines.pypcapfile.PyPCAPFile` therefore @@ -49,6 +58,7 @@ import binascii import collections import ctypes +import functools import ipaddress import struct import sys @@ -62,7 +72,9 @@ 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.protocols.internet.ipv4 import IPv4 as Protocol_IPv4 from pcapkit.protocols.schema.internet.ipv6 import jumbo_payload_length +from pcapkit.protocols.transport.tcp import TCP as Protocol_TCP from pcapkit.utilities.exceptions import ProtocolError, UnsupportedCall, stacklevel from pcapkit.utilities.warnings import AttributeWarning, warn @@ -99,6 +111,26 @@ #: Length of the fixed IPv6 header, :rfc:`8200#section-3`. IPV6_HEADER_LEN = 40 +#: Length of the fixed part of an IP Authentication Header (AH), before its +#: Integrity Check Value, :rfc:`4302#section-2`. A Payload Length of ``0`` +#: declares a shorter header, which the default engine rejects. +AH_HEADER_LEN = 12 + +#: IP protocol numbers of the headers :func:`_past_extension_headers` reads +#: past inside IPv4, on to the protocol each one's Next Header field names, as +#: the default engine does: it dispatches IPv4's protocol through the registry +#: IPv6 uses (:attr:`Internet.__proto__ +#: `), and each of these +#: parsers then decodes what its Next Header names. That is AH and the IPv6 +#: Hop-by-Hop Options, Routing, Fragment, Destination Options, Mobility and HIP +#: headers. ESP is left out, as its Next Header is encrypted, and Shim6, whose +#: parser refuses IPv4. +IPV4_EXTENSION_HEADERS = frozenset(( + Enum_TransType.HOPOPT, Enum_TransType.IPv6_Route, Enum_TransType.IPv6_Frag, + Enum_TransType.AH, Enum_TransType.IPv6_Opts, Enum_TransType.Mobility_Header, + Enum_TransType.HIP, +)) + #: EtherTypes of the VLAN tags :func:`_network` reads past: the 802.1Q customer #: tag and the 802.1ad service tag, which QinQ stacks. These are the two the #: default engine dissects (see :attr:`Link.__proto__ @@ -381,6 +413,69 @@ def _untag(link: 'Any') -> 'Optional[tuple[int, bytes]]': return ethertype, data +@functools.lru_cache(maxsize=2) +def _default_accepts(name: 'str', header: 'bytes') -> 'bool': + """Test if the default engine's parser accepts a header. + + Args: + name: ``'IPv4'`` or ``'TCP'``, the parser to ask. + header: The whole header, options included, and nothing after it. + + Returns: + :data:`False` if the parser raises anything at all, as the default + engine then leaves the layer undissected (see + :func:`~pcapkit.utilities.decorators.beholder`); else :data:`True`. + + Note: + This is for the options, whose rules are the parser's and too many for + a header walk to follow, so it is asked only of a header that has some. + It is given the header alone and stops at its own layer, so nothing + past the header is dissected. What the parser logs and warns of, a + rejected header's :exc:`~pcapkit.utilities.exceptions.ProtocolError` + included, it logs and warns of as it does when the default engine reads + the same frame; neither the warnings filters nor the logger's level are + changed. The answer depends on the octets alone, so the last two are + remembered: the adapters each fetch a frame's layers in turn, and this + parses a frame's IPv4 and TCP headers once each between them. + + """ + protocol = Protocol_IPv4 if name == 'IPv4' else Protocol_TCP + try: + protocol(header, len(header), protocol=name) + except Exception: # as beholder does + return False + return True + + +def _ipv4_accepted(ipv4: 'IP') -> 'bool': + """Test if the default engine reads a PyPCAPFile IPv4 packet as an IPv4 layer. + + Args: + ipv4: PyPCAPFile IPv4 packet. + + Returns: + :data:`False` if its header length (IHL) runs past the captured octets + (:issue:`1591`), or its options are ones the default engine's IPv4 + parser rejects (:issue:`1596`); else :data:`True`. + + Note: + `PyPCAPFile`_ slices the header at its IHL however few octets there + are, and never checks the options. A header with none is not parsed: + all the default engine checks of one is that it is version 4, with an + IHL of at least 5 and 20 octets captured, and `PyPCAPFile`_'s decoder + refuses any other. + + .. _PyPCAPFile: https://github.com/kisom/pypcapfile + + """ + if ipv4.hl * 4 == IPV4_HEADER_LEN: + return True + options = _maybe_unhex(bytes(ipv4.opt)) + if IPV4_HEADER_LEN + len(options) < ipv4.hl * 4: + return False + return _default_accepts('IPv4', ipv4_header(ipv4)) + + def _network(packet: 'Packet', count: 'int' = -1) -> 'Optional[IP]': """Fetch the decoded IPv4 layer of a PyPCAPFile packet, if any. @@ -392,7 +487,8 @@ def _network(packet: 'Packet', count: 'int' = -1) -> 'Optional[IP]': The decoded :class:`~pcapfile.protocols.network.ip.IP` layer, or :data:`None` when the frame was not decoded that far -- which is the case for every non-IPv4 frame, `PyPCAPFile`_ having no other network - layer decoder. + layer decoder -- or the default engine rejects its header (see + :func:`_ipv4_accepted`). Warns: AttributeWarning: If an IPv4 packet behind VLAN tags will not decode, @@ -421,7 +517,7 @@ def _network(packet: 'Packet', count: 'int' = -1) -> 'Optional[IP]': payload = getattr(link, 'payload', None) if type(payload).__name__ == 'IP': - return payload + return payload if _ipv4_accepted(payload) else None if getattr(link, 'type', None) not in VLAN_TAGS: return None @@ -436,7 +532,7 @@ def _network(packet: 'Packet', count: 'int' = -1) -> 'Optional[IP]': from pcapfile.protocols.network.ip import IP # isort:skip try: - return IP(untagged[1], 0) + ipv4 = IP(untagged[1], 0) except (struct.error, AssertionError, ValueError, IndexError, KeyError) as error: key, anchor = _capture_of(packet) if not _undecoded_last or _undecoded_last[0][:2] != (key, count): @@ -444,6 +540,128 @@ def _network(packet: 'Packet', count: 'int' = -1) -> 'Optional[IP]': warn(f'Frame {count}: {Enum_LinkType.ETHERNET!r} decoding failed ({error!r}); ' 'frame left undecoded', AttributeWarning, stacklevel=stacklevel()) return None + return ipv4 if _ipv4_accepted(ipv4) else None + + +def _ipv4_in_ipv6(packet: 'Packet') -> 'Optional[IP]': + """Fetch the IPv4 layer of a PyPCAPFile frame carrying IPv4 in IPv6 (4in6). + + Args: + packet: PyPCAPFile packet. + + Returns: + The first IPv4 packet inside an IPv6 packet -- directly, behind + extension headers, or inside more IPv6 (6in6) -- decoded as + :func:`_network` decodes one behind VLAN tags, or :data:`None` if there + is none, it will not decode, or the default engine rejects its header + (see :func:`_ipv4_accepted`). The frame may be tagged (see + :func:`_untag`), and each IPv6 header is stepped over as + :func:`_upper_layer` does. + + Note: + `PyPCAPFile`_ has no IPv6 decoder, so leaves the IPv4 inside undecoded. + This is for :func:`ipv4_reassembly` alone, which the default engine + gives the first IPv4 layer of a frame however deep. TCP inside it is + tunnelled, so is still left out (see :func:`_left_out`), and nor does + an IPv4 packet that will not decode warn: `PyPCAPFile`_ never decodes + IPv6, so has nothing to warn of. + + .. _PyPCAPFile: https://github.com/kisom/pypcapfile + + """ + link = packet.packet + untagged = None if _is_raw(link) else _untag(link) + if untagged is None or untagged[0] != Enum_EtherType.Internet_Protocol_version_6: + return None + + upper = _upper_layer(6, untagged[1]) + while upper is not None and upper[0] == Enum_TransType.IPv6: + upper = _upper_layer(6, upper[1]) + if upper is None or upper[0] != Enum_TransType.IPv4: + return None + + # NOTE: imported only once there is something to decode, as in :func:`_network`. + from pcapfile.protocols.network.ip import IP # isort:skip + + try: + ipv4 = IP(upper[1], 0) + except (struct.error, AssertionError, ValueError, IndexError, KeyError): + return None + return ipv4 if _ipv4_accepted(ipv4) else None + + +@functools.lru_cache(maxsize=8) +def _default_extension_header(protocol: 'int', data: 'bytes') -> 'Optional[tuple[int, int]]': + """Read an extension header inside IPv4 with the default engine's own parser. + + Args: + protocol: The header's protocol number, one of + :data:`IPV4_EXTENSION_HEADERS` other than AH. + data: Raw octets from the header on, as far as the IPv4 packet goes. + + Returns: + The header's Next Header field and its length, or :data:`None` if the + default engine leaves it undissected: its parser raises anything at + all (see :func:`~pcapkit.utilities.decorators.beholder`), or fewer + octets were captured than the header needs. + + Note: + The parser is the one the default engine dispatches ``protocol`` to + inside IPv4, called as it calls it, so its options, routing data, + message or parameters are checked as the default engine checks them. + It stops at its own layer, so nothing past the header is dissected, and + the last few answers are remembered, as in :func:`_default_accepts`, + which this logs and warns as. + + """ + # pylint: disable-next=protected-access + klass = Protocol_IPv4._lookup_next_layer(Protocol_IPv4.__proto__, protocol) + try: + header = Protocol_IPv4._parse_next_layer( # pylint: disable=protected-access + klass, data, len(data), version=4, extension=False, alias=protocol, protocol=klass) + except Exception: # as beholder does + return None + nxt = getattr(header.info, 'next', None) # none on a header cut short, read as raw octets + if nxt is None: + return None + return int(nxt), header.length + + +def _past_extension_headers(protocol: 'int', data: 'bytes') -> 'Optional[tuple[int, bytes]]': + """Step over the headers at the head of an IPv4 payload that the default engine reads past. + + Args: + protocol: The IPv4 packet's protocol number. + data: Raw octets of its payload. + + Returns: + The protocol number after the last of the :data:`IPV4_EXTENSION_HEADERS` + and the octets from its header on -- ``protocol`` and ``data`` + themselves if ``protocol`` is none of them -- or :data:`None` if the + default engine leaves one of them undissected. An AH is + ``(Payload Length + 2) * 4`` octets long (:rfc:`4302#section-2.2`), and + no shorter than :data:`AH_HEADER_LEN`; any other is read by + :func:`_default_extension_header` (:issue:`1597`). + + Note: + Inside IPv4, the default engine decodes what an IPv6 Fragment header + names whatever its offset, so a later fragment is read past too. Over + IPv6, :func:`_upper_layer` steps over these as the extension headers + they are there. + + """ + while protocol in IPV4_EXTENSION_HEADERS: + if protocol == Enum_TransType.AH: + length = (data[1] + 2) * 4 if len(data) >= AH_HEADER_LEN else 0 + if length < AH_HEADER_LEN or len(data) < length: + return None + protocol, data = data[0], data[length:] + continue + header = _default_extension_header(protocol, data) + if header is None: + return None + protocol, data = header[0], data[header[1]:] + return protocol, data def _upper_layer(version: 'int', data: 'bytes') -> 'Optional[tuple[int, bytes]]': @@ -459,8 +677,9 @@ def _upper_layer(version: 'int', data: 'bytes') -> 'Optional[tuple[int, bytes]]' An IPv4 header must be version 4 with a header length of at least :data:`IPV4_HEADER_LEN` octets, all of them captured, and what follows - it ends at its Total Length, as in :func:`_ipv4_payload`. An IPv6 packet - ends at its Payload Length or, where that is ``0``, at the Jumbo Payload + it ends at its Total Length, as in :func:`_ipv4_payload`, and is + followed past the headers :func:`_past_extension_headers` steps over. + An IPv6 packet ends at its Payload Length or, where that is ``0``, at the Jumbo Payload Length a jumbogram's Hop-by-Hop Options header gives (:rfc:`2675`) -- lacking one, it has no payload -- and is followed past the extension headers in :data:`IPV6_EXTENSION_HEADERS`; a first fragment is read past its @@ -475,9 +694,10 @@ def _upper_layer(version: 'int', data: 'bytes') -> 'Optional[tuple[int, bytes]]' :func:`_decline_tcp` gives fires on the frames the default engine finds TCP in. This reads only as far as the upper-layer protocol number and decodes nothing, so it does not see every reason the default engine has - to leave octets undissected: an IPv4 option or an extension header - option it rejects, which leaves the header raw, or TCP options it - rejects, which in a tunnel leave the segment raw. A malformed tunnelled + to leave octets undissected: an option in a tunnelled IPv4 header or an + extension header option it rejects, which leaves the header raw, or TCP + options it rejects, which in a tunnel leave the segment raw. (The outer + IPv4 header is checked, see :func:`_ipv4_accepted`.) A malformed tunnelled frame like that is warned of all the same, though the default engine finds no TCP in it either -- a warning that errs on the side of being given, rather than a dissection of every such frame to settle it. @@ -492,7 +712,7 @@ def _upper_layer(version: 'int', data: 'bytes') -> 'Optional[tuple[int, bytes]]' if struct.unpack_from('!H', data, 6)[0] & 0x1FFF: # a later fragment return None total = struct.unpack_from('!H', data, 2)[0] - return data[9], data[ihl:max(total, ihl)] if total else data[ihl:] + return _past_extension_headers(data[9], data[ihl:max(total, ihl)] if total else data[ihl:]) if len(data) < IPV6_HEADER_LEN: return None @@ -544,9 +764,12 @@ def _left_out(packet: 'Packet', ipv4: 'Optional[IP]') -> 'Optional[str]': """ if ipv4 is not None: # read by PyPCAPFile, so only a tunnel can hide TCP # a later fragment has no upper-layer header, see ``_upper_layer`` - if ipv4.off or ipv4.p not in IP_TUNNELS or not _is_raw(ipv4.payload): + if (ipv4.off or (ipv4.p not in IP_TUNNELS and ipv4.p not in IPV4_EXTENSION_HEADERS) + or not _is_raw(ipv4.payload)): + return None + upper = _past_extension_headers(ipv4.p, _ipv4_payload(ipv4)) # type: Optional[tuple[int, bytes]] + if upper is None or upper[0] not in IP_TUNNELS: return None - upper = ipv4.p, _ipv4_payload(ipv4) # type: Optional[tuple[int, bytes]] else: link = packet.packet untagged = None if _is_raw(link) else _untag(link) @@ -665,20 +888,40 @@ def _transport(ipv4: 'IP') -> 'Optional[bytes]': Returns: The raw TCP segment, as far as the Total Length (see - :func:`_ipv4_payload`), or :data:`None` if the packet does not carry a - TCP payload long enough to hold a header. A later fragment carries none: - from its offset on, its data is a slice of the datagram's payload, with - no upper-layer header in it (:rfc:`791#section-3.2`, :issue:`1576`). + :func:`_ipv4_payload`), directly or behind AH or IPv6 extension headers + (see :func:`_past_extension_headers`, :issue:`1593`, :issue:`1597`), or + :data:`None` if the packet does + not carry a TCP header the default engine reads. A later fragment + carries none: from its offset on, its data is a slice of the datagram's + payload, with no upper-layer header in it (:rfc:`791#section-3.2`, + :issue:`1576`). Nor does a segment too short for a header, or whose Data + Offset is below 5 words (:issue:`1590`). + + Behind those, the default engine reads TCP only as its TCP parser does, so + the whole of the header its Data Offset gives must be there, and any + options must be ones that parser accepts (see + :func:`_default_accepts`). Directly over IPv4 it reads a header the + parser rejects out of the raw octets left behind instead + (:func:`pcapkit.toolkit.pcap.tcp_segment`, :issue:`1518`). """ - if ipv4.p != Enum_TransType.TCP or ipv4.off: + if (ipv4.p != Enum_TransType.TCP and ipv4.p not in IPV4_EXTENSION_HEADERS) or ipv4.off: return None if not _is_raw(ipv4.payload): return None - segment = _ipv4_payload(ipv4) + upper = _past_extension_headers(ipv4.p, _ipv4_payload(ipv4)) + if upper is None or upper[0] != Enum_TransType.TCP: + return None + segment = upper[1] if len(segment) < TCP_MIN_HEADER_LEN: return None + header = (segment[12] >> 4) * 4 # the Data Offset, in octets + if header < TCP_MIN_HEADER_LEN: + return None + if ipv4.p != Enum_TransType.TCP and (header > len(segment) or ( + header > TCP_MIN_HEADER_LEN and not _default_accepts('TCP', segment[:header]))): + return None return segment @@ -833,7 +1076,10 @@ def ipv4_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'IP_Packet[IPv4Ad * If the ``packet`` can be used for IPv4 reassembly. A packet can be reassembled if it contains an IPv4 layer (:class:`pcapfile.protocols.network.ip.IP`), behind - 802.1Q or 802.1ad VLAN tags or not, and the **DF** flag is :data:`False`. + 802.1Q or 802.1ad VLAN tags or not, and the **DF** flag is :data:`False`. The + layer is the first IPv4 packet of the frame, as for the default engine: one + carried in IPv6 counts (see :func:`_ipv4_in_ipv6`), and one whose header the + default engine rejects does not (see :func:`_ipv4_accepted`). * If the ``packet`` can be reassembled, then the :obj:`dict` mapping of data for IPv4 reassembly (:term:`reasm.ipv4.packet`) will be returned; otherwise, returns :data:`None`. @@ -846,6 +1092,8 @@ def ipv4_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'IP_Packet[IPv4Ad """ ipv4 = _network(packet, count) + if ipv4 is None: + ipv4 = _ipv4_in_ipv6(packet) if ipv4 is None: return None if ipv4.flags & IPV4_FLAG_DF: # dismiss not fragmented packet @@ -899,8 +1147,10 @@ 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, behind 802.1Q or 802.1ad - VLAN tags or not; TCP over IPv6 cannot be, as `PyPCAPFile`_ has no IPv6 decoder, + if it contains an IPv4 layer carrying a TCP segment, directly or behind AH or + IPv6 extension headers, and behind 802.1Q or 802.1ad VLAN tags or not (see + :func:`_transport`); + TCP over IPv6 cannot be, as `PyPCAPFile`_ has no IPv6 decoder, and nor can TCP tunnelled in IP, as it decodes no tunnel. * 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`. @@ -932,7 +1182,7 @@ def tcp_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'TCP_Packet | None from pcapfile.protocols.transport.tcp import TCP # isort:skip tcp = TCP(segment) - hdr_len = max(tcp.data_offset, TCP_MIN_HEADER_LEN) + hdr_len = tcp.data_offset # at least TCP_MIN_HEADER_LEN, see ``_transport`` payload = segment[hdr_len:] return TCP_Packet( @@ -970,8 +1220,10 @@ 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, behind 802.1Q or 802.1ad - VLAN tags or not; TCP over IPv6 cannot be, as `PyPCAPFile`_ has no IPv6 decoder, + if it contains an IPv4 layer carrying a TCP segment, directly or behind AH or + IPv6 extension headers, and behind 802.1Q or 802.1ad VLAN tags or not (see + :func:`_transport`); + TCP over IPv6 cannot be, as `PyPCAPFile`_ has no IPv6 decoder, and nor can TCP tunnelled in IP, as it decodes no tunnel. * 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`. @@ -1003,7 +1255,7 @@ def tcp_traceflow(packet: 'Packet', *, data_link: 'Enum_LinkType', from pcapfile.protocols.transport.tcp import TCP # isort:skip tcp = TCP(segment) - hdr_len = max(tcp.data_offset, TCP_MIN_HEADER_LEN) + hdr_len = tcp.data_offset # at least TCP_MIN_HEADER_LEN, see ``_transport`` return TF_TCP_Packet( # type: ignore[type-var] protocol=data_link, # data link type from savefile header index=count, # frame number diff --git a/tests/toolkit/test_pypcapfile_header_checks_unit.py b/tests/toolkit/test_pypcapfile_header_checks_unit.py new file mode 100644 index 0000000000..c9cfc4b4c4 --- /dev/null +++ b/tests/toolkit/test_pypcapfile_header_checks_unit.py @@ -0,0 +1,258 @@ +"""The header checks :mod:`pcapkit.toolkit.pypcapfile` makes as the default engine does. + +`pypcapfile`_ slices an IPv4 header at its IHL however few octets were captured, +never parses its options, has no IPv6 decoder and no AH decoder, and reads a TCP +Data Offset below 5 words as if it were 5. The toolkit reads such frames as the +default engine does instead: GH#1590 (Data Offset below 5), GH#1591 (IHL past +the packet), GH#1592 (IPv4 in IPv6), GH#1593 (TCP behind an IPv4 AH header) and +GH#1596 (IPv4 options the default engine rejects). + +These tests use stand-ins for `pypcapfile`_'s decoders, so run without it. The +ones that read real captures with both engines need it, so live in +:mod:`tests.toolkit.test_pypcapfile_unit`, which the PyPCAPFile CI cell runs. + +.. _pypcapfile: https://github.com/kisom/pypcapfile + +""" +from __future__ import annotations + +import unittest +import warnings +from unittest import mock + +from tests._support import reimport_once_per_class +from tests.foundation import _roundtrip as wire +from tests.toolkit import test_pypcapfile_unit as base + + +@unittest.skipUnless(base.HAS_RUNTIME, 'runtime dependencies not installed') +class PyPCAPFileHeaderCheckTests(unittest.TestCase): + """Stand-ins for the decoded layers, so `pypcapfile`_ is not needed.""" + + decline = base.PyPCAPFileTCPOverIPv6Tests.decline + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_data_offset_below_5_is_no_tcp(self) -> None: + # C.f. #1590: no TCP input, and no warning either. + for words in range(5): + with self.subTest(words=words): + segment = base.data_offset(base.make_tcp(b'data'), words) + frame = base.FakeEthernet(base.FakeIP(segment)) + self.assertEqual(self.decline(base.make_packet(frame)), []) + + def test_ipv4_header_past_the_packet_is_no_ipv4(self) -> None: + # C.f. #1591: pypcapfile slices the 60-octet header of a 40-octet packet + # as 20 octets of options and no payload. + from pcapkit.toolkit.pypcapfile import ipv4_reassembly + + for label, layer in ( + ('udp', base.FakeIP(b'', p=17, hl=15, opt=b'\x01' * 20, len=40)), + ('tcp', base.FakeIP(b'', hl=15, opt=base.make_tcp(b''), len=40)), + ('one option word short', base.FakeIP(b'', p=17, hl=7, opt=b'\x01' * 4, len=24)), + ): + with self.subTest(case=label): + packet = base.make_packet(base.FakeEthernet(layer)) + self.assertIsNone(ipv4_reassembly(packet, count=1)) + self.assertEqual(self.decline(packet), []) + + def test_ipv4_options_the_default_engine_rejects_are_no_ipv4(self) -> None: + # C.f. #1596: a fragment with such options is no IPv4 input, a segment + # behind them no TCP input; one with options it accepts is both. + from pcapkit.toolkit.pypcapfile import ipv4_reassembly + + for options in base.BAD_IPV4_OPTIONS: + with self.subTest(options=options.hex()): + packet = base.make_packet(base.FakeEthernet(base.FakeIP(b'x' * 16, p=17, hl=6, + opt=options))) + self.assertIsNone(ipv4_reassembly(packet, count=1)) + tcp = base.make_packet(base.FakeEthernet(base.FakeIP(base.make_tcp(b'data'), hl=6, + opt=options))) + self.assertEqual(self.decline(tcp), []) + for options in base.GOOD_IPV4_OPTIONS: + with self.subTest(options=options.hex()): + packet = base.make_packet(base.FakeEthernet(base.FakeIP(b'x' * 16, p=17, hl=6, + opt=options))) + data = ipv4_reassembly(packet, count=1) + self.assertIsNotNone(data) + self.assertEqual(data.header[20:], options) + + def test_options_are_parsed_once_per_frame(self) -> None: + # The adapters each fetch a frame's network layer; the default engine's + # parser is asked once between them, and not at all without options. + from pcapkit.const.reg.linktype import LinkType + from pcapkit.toolkit import pypcapfile + + pypcapfile._default_accepts.cache_clear() + real = pypcapfile.Protocol_IPv4 + frames = [base.FakeEthernet(base.FakeIP(b'x' * 16, p=17, hl=6, opt=options)) + for options in (*base.GOOD_IPV4_OPTIONS, base.BAD_IPV4_OPTIONS[0])] + frames.append(base.FakeEthernet(base.FakeIP(b'x' * 16, p=17))) + with mock.patch.object(pypcapfile, 'Protocol_IPv4', side_effect=real) as parser: + for count, frame in enumerate(frames, start=1): + packet = base.make_packet(frame) + pypcapfile.ipv4_reassembly(packet, count=count) + pypcapfile.tcp_reassembly(packet, count=count) + pypcapfile.tcp_traceflow(packet, data_link=LinkType.ETHERNET, count=count) + self.assertEqual([call.args[0][20:] for call in parser.call_args_list], + [*base.GOOD_IPV4_OPTIONS, base.BAD_IPV4_OPTIONS[0]]) + + def test_parsing_options_leaves_the_warnings_filters_and_the_logger_level_alone(self) -> None: + import logging + + from pcapkit.toolkit.pypcapfile import ipv4_reassembly + + logger = logging.getLogger('pcapkit') + level, filters = logger.level, list(warnings.filters) + for count, options in enumerate((*base.BAD_IPV4_OPTIONS, *base.GOOD_IPV4_OPTIONS), start=1): + frame = base.FakeEthernet(base.FakeIP(b'x' * 16, p=17, hl=6, opt=options)) + ipv4_reassembly(base.make_packet(frame), count=count) + self.assertEqual(warnings.filters, filters) + self.assertEqual(logger.level, level) + + def test_ipv4_in_ipv6_is_found(self) -> None: + # C.f. #1592: the first IPv4 packet inside IPv6, however it is wrapped, + # goes to pypcapfile's own IPv4 decoder. + from pcapkit.toolkit.pypcapfile import _ipv4_in_ipv6 + + inner = base.udp_fragment() + for label, frame in ( + ('4in6', base.ipv6_frame(wire.ipv6(inner, nxt=4))), + ('802.1Q, 4in6', base.tagged_frame(wire.ipv6(inner, nxt=4), 0x86DD, base.C_TAG)), + ('hop-by-hop', base.ipv6_frame(wire.ipv6(bytes([4, 0, 1, 4, 0, 0, 0, 0]) + inner, nxt=0))), + ('first fragment', base.ipv6_frame(wire.ipv6_fragment(inner, mf=True, nxt=4))), + ('ah', base.ipv6_frame(wire.ipv6(base.ah(4) + inner, nxt=51))), + ('4in6in6', base.ipv6_frame(wire.ipv6(wire.ipv6(inner, nxt=4), nxt=41))), + ('ethernet padding', base.ipv6_frame(wire.ipv6(inner, nxt=4) + bytes(6))), + ): + with self.subTest(case=label), mock.patch.dict('sys.modules', base.fake_ip_decoder()) as modules: + IP = modules['pcapfile.protocols.network.ip'].IP + self.assertIsInstance(_ipv4_in_ipv6(base.make_packet(frame)), IP) + self.assertEqual(IP.calls, [(inner, 0)]) + + def test_no_ipv4_in_ipv6_where_the_default_engine_finds_none(self) -> None: + from pcapkit.toolkit.pypcapfile import _ipv4_in_ipv6 + + inner = base.udp_fragment() + for label, frame, decoded in ( + ('ipv4 over ethernet', base.FakeEthernet(base.FakeIP(inner, p=4)), False), + ('ipv6, udp', base.ipv6_frame(wire.ipv6(wire.udp(b'x'), nxt=17)), False), + ('ipv6 later fragment', base.ipv6_frame(wire.ipv6_fragment(inner, offset=8, nxt=4)), False), + ('ipv6 truncated', base.ipv6_frame(wire.ipv6(inner, nxt=4)[:39]), False), + ('esp', base.ipv6_frame(wire.ipv6(bytes(8) + inner, nxt=50)), False), + ('802.1Q, no ipv6 header', base.tagged_frame(b'', 0x86DD, base.C_TAG), False), + ('undecoded frame', b'\x00' * 20, False), + # pypcapfile never decodes IPv6, so has nothing to warn of + ('inner ihl 4', base.ipv6_frame(wire.ipv6(base.ihl(inner, 4), nxt=4)), True), + ('inner ihl past the packet', base.ipv6_frame(wire.ipv6(base.ihl(inner, 15), nxt=4)), True), + ): + with self.subTest(case=label), warnings.catch_warnings(record=True) as caught, \ + mock.patch.dict('sys.modules', base.fake_ip_decoder()) as modules: + warnings.simplefilter('always') + IP = modules['pcapfile.protocols.network.ip'].IP + self.assertIsNone(_ipv4_in_ipv6(base.make_packet(frame))) + self.assertEqual(bool(IP.calls), decoded) + self.assertEqual(caught, []) + + def test_transport_reads_tcp_behind_ah(self) -> None: + # C.f. #1593: the default engine dissects AH over IPv4, and the TCP behind it. + from pcapkit.toolkit.pypcapfile import _transport + + segment = base.make_tcp(b'data') + options = base.make_tcp(b'data', options=b'\x02\x04\x05\xb4') # MSS + for label, payload, want in ( + ('ah', base.ah(6) + segment, segment), + ('ah of the fixed part only', base.ah(6, 1) + segment, segment), + ('two ah', base.ah(51) + base.ah(6, 1) + segment, segment), + ('ah, tcp options', base.ah(6) + options, options), + ): + with self.subTest(case=label): + self.assertEqual(_transport(base.FakeIP(payload, p=51)), want) + + def test_no_tcp_behind_ah_where_the_default_engine_finds_none(self) -> None: + # Behind AH there is no fall-back to the raw octets (#1518), so a TCP + # header its parser rejects is no TCP -- and no TCP over IPv6 either. + from pcapkit.toolkit.pypcapfile import _transport + + segment = base.make_tcp(b'data') + for label, payload, fields in ( + ('payload length 0', base.ah(6, 0) + segment, {}), + ('ah cut short', base.ah(6)[:15], {}), + ('ah under the fixed part', base.ah(6)[:11], {}), + ('ah, udp', base.ah(17) + wire.udp(b'x'), {}), + ('ah, tcp truncated', base.ah(6) + segment[:19], {}), + ('ah, data offset 4', base.ah(6) + base.data_offset(segment, 4), {}), + ('ah, data offset past the capture', base.ah(6) + base.data_offset(segment, 9), {}), + ('ah, mss option of length 3', base.ah(6) + base.make_tcp(b'', options=b'\x02\x03\x00\x00'), {}), + ('later fragment', base.ah(6) + segment, {'off': 3}), + ): + with self.subTest(case=label): + layer = base.FakeIP(payload, p=51, **fields) + self.assertIsNone(_transport(layer)) + self.assertEqual(self.decline(base.make_packet(base.FakeEthernet(layer))), []) + + def test_transport_reads_tcp_behind_ipv6_extension_headers(self) -> None: + # C.f. #1597: inside IPv4 too, the default engine dissects these headers + # and what each one's Next Header names. + from pcapkit.toolkit.pypcapfile import _transport + + segment = base.make_tcp(b'data') + cases = [(label, protocol, header) for label, protocol, header in base.IPV4_EXTENSION_HEADERS] + cases += [('hop-by-hop, destination options', 0, base.options_header(60) + base.options_header(6)), + ('destination options, ah', 60, base.options_header(51) + base.ah(6)), + ('ah, hop-by-hop', 51, base.ah(0) + base.options_header(6))] + for label, protocol, header in cases: + with self.subTest(case=label): + self.assertEqual(_transport(base.FakeIP(header + segment, p=protocol)), segment) + + def test_no_tcp_behind_ipv6_extension_headers_where_the_default_engine_finds_none(self) -> None: + from pcapkit.toolkit.pypcapfile import _transport + + segment = base.make_tcp(b'data') + for label, protocol, payload in ( + ('padn of length 9', 60, base.options_header(6, b'\x01\x09' + bytes(4)) + segment), + ('header length past the packet', 60, bytes([6, 9]) + bytes(6) + segment), + ('cut short', 0, base.options_header(6)[:4]), + # read as raw octets, as fewer were captured than the header needs + ('routing header cut short', 43, bytes([6, 0, 0, 0])), + ('routing type 3, under its addresses', 43, bytes([6, 0, 3, 1]) + bytes(4) + segment), + ('hip header length 3', 139, bytes([6, 3, 0x01, 0x21]) + bytes(28) + segment), + ('shim6, whose parser refuses ipv4', 140, bytes([6, 0, 0x80, 0]) + bytes(4) + segment), + ('no next header', 59, segment), + ('tcp option of length 3', 60, + base.options_header(6) + base.make_tcp(b'', options=b'\x02\x03\x00\x00')), + ): + with self.subTest(case=label): + layer = base.FakeIP(payload, p=protocol) + self.assertIsNone(_transport(layer)) + self.assertEqual(self.decline(base.make_packet(base.FakeEthernet(layer))), []) + + def test_extension_headers_are_parsed_once_per_frame(self) -> None: + from pcapkit.const.reg.linktype import LinkType + from pcapkit.toolkit import pypcapfile + + pypcapfile._default_extension_header.cache_clear() + layer = base.FakeIP(base.options_header(60) + base.options_header(17) + wire.udp(b'x'), p=0) + packet = base.make_packet(base.FakeEthernet(layer)) + pypcapfile.ipv4_reassembly(packet, count=1) + pypcapfile.tcp_reassembly(packet, count=1) + pypcapfile.tcp_traceflow(packet, data_link=LinkType.ETHERNET, count=1) + self.assertEqual(pypcapfile._default_extension_header.cache_info().misses, 2) + + def test_tunnelled_tcp_behind_ah_warns(self) -> None: + tcp4 = wire.ipv4(base.TCP_SEGMENT, proto=6) + for label, frame in ( + ('4in4 behind ah', base.tunnel(base.ah(4) + tcp4, 51)), + ('6in4 behind ah', base.tunnel(base.ah(41) + wire.ipv6(base.TCP_SEGMENT, nxt=6), 51)), + ('4in4, inner ah', base.tunnel(wire.ipv4(base.ah(6) + base.TCP_SEGMENT, proto=51), 4)), + ('4in4 behind destination options', base.tunnel(base.options_header(4) + tcp4, 60)), + ): + with self.subTest(case=label): + messages = self.decline(base.make_packet(frame)) + self.assertEqual(len(messages), 1, messages) + self.assertTrue(messages[0].startswith('Frame 1: TCP tunnelled in IP'), messages) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/toolkit/test_pypcapfile_unit.py b/tests/toolkit/test_pypcapfile_unit.py index 0a345948bd..a485b59071 100644 --- a/tests/toolkit/test_pypcapfile_unit.py +++ b/tests/toolkit/test_pypcapfile_unit.py @@ -199,6 +199,9 @@ def __init__(self, packet: bytes, layers: int = 0) -> None: if packet[0] >> 4 != 4 or packet[0] & 0x0F <= 4: raise AssertionError('not an IPv4 packet.') struct.unpack('!BBHHHBBHII', packet[:20]) + # the header length and options, sliced without a bounds check, as the real one does + self.hl = packet[0] & 0x0F + self.opt = packet[20:self.hl * 4] modules = {name: types.ModuleType(name) for name in ( 'pcapfile', 'pcapfile.protocols', 'pcapfile.protocols.network', @@ -225,6 +228,52 @@ def tunnel(packet: bytes, protocol: int, **fields) -> FakeEthernet: return FakeEthernet(FakeIP(binascii.hexlify(packet), p=protocol, **fields)) +#: IPv4 options the default engine's IPv4 parser rejects: a Record Route of +#: length 1, one whose length runs past the option area, and a Router Alert of +#: length 2. +BAD_IPV4_OPTIONS = (b'\x07\x01\x00\x00', b'\x07\x08\x00\x00', b'\x94\x02\x00\x00') + +#: IPv4 options it accepts: four No Operations, a Router Alert, and an End of +#: Option List followed by octets it leaves as padding. +GOOD_IPV4_OPTIONS = (b'\x01' * 4, b'\x94\x04\x00\x00', b'\x00\xff\xff\xff') + + +def ah(nxt: int, words: int = 4) -> bytes: + """An IP Authentication Header whose Payload Length is ``words``, then next header ``nxt``.""" + icv = b'\xaa' * max((words + 2) * 4 - 12, 0) + return bytes([nxt, words, 0, 0]) + (0x100).to_bytes(4, 'big') + (1).to_bytes(4, 'big') + icv + + +def options_header(nxt: int, padn: bytes = b'\x01\x04\x00\x00\x00\x00') -> bytes: + """An IPv6 Hop-by-Hop or Destination Options header of 8 octets, then next header ``nxt``.""" + return bytes([nxt, 0]) + padn + + +#: IPv6 extension headers the default engine reads past inside IPv4, each with +#: its protocol number, then TCP: Hop-by-Hop and Destination Options, Routing +#: (type 0, then type 2), a Fragment header (a later fragment), Mobility (a +#: Binding Refresh Request) and HIP (an I1, with no parameters). +IPV4_EXTENSION_HEADERS = ( + ('hop-by-hop', 0, options_header(6)), + ('destination options', 60, options_header(6)), + ('routing type 0', 43, bytes([6, 0, 0, 0]) + bytes(4)), + ('routing type 2', 43, bytes([6, 2, 2, 1]) + bytes(20)), + ('later fragment', 44, bytes([6, 0, 0, 0x10]) + (1).to_bytes(4, 'big')), + ('mobility', 135, bytes([6, 1, 0, 0]) + bytes(12)), + ('hip', 139, bytes([6, 4, 0x01, 0x21]) + bytes(36)), +) + + +def ihl(packet: bytes, words: int) -> bytes: + """``packet``, an IPv4 packet, with its IHL set to ``words``.""" + return bytes([0x40 | words]) + packet[1:] + + +def udp_fragment(**fields) -> bytes: + """A first IPv4 fragment of a UDP datagram, more to come.""" + return wire.ipv4(wire.udp(b'u' * 24)[:16], proto=17, mf=True, **fields) + + @unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') class PyPCAPFileToolkitTests(unittest.TestCase): def setUp(self) -> None: @@ -737,6 +786,9 @@ def test_no_warning_where_the_default_engine_finds_no_tcp(self) -> None: tunnel(total_length(wire.ipv4(make_tcp(b'', options=b'\x02\x04\x05\xb4'), proto=6), 40) + bytes(8), 4)), ('tunnelled data offset 4', tunnel(wire.ipv4(data_offset(TCP_SEGMENT, 4), proto=6), 4)), + # nor is a tunnel whose outer IPv4 header the default engine rejects (#1591, #1596) + ('outer ipv4 option', tunnel(tcp4, 4, hl=6, opt=b'\x94\x02\x00\x00')), # router alert, length 2 + ('outer ipv4 header overruns the packet', tunnel(tcp4, 4, hl=6, opt=b'\x94\x04')), # nor is one whose Data Offset is under 5 words, over IPv6 alone too ('ipv6 data offset 4', ipv6_frame(wire.ipv6(data_offset(TCP_SEGMENT, 4), nxt=6))), ('802.1Q, ipv6 data offset 0', tagged_frame(wire.ipv6(data_offset(TCP_SEGMENT, 0), nxt=6), @@ -816,7 +868,7 @@ def test_malformed_tunnelled_tcp_the_header_walk_cannot_see_into_still_warns(sel tunnel(wire.ipv4(make_tcp(b'x', options=b'\x02\x00\x05\xb4'), proto=6), 4)), ('tunnelled tcp option past its area', ipv6_frame(wire.ipv6(wire.ipv6(make_tcp(b'x', options=b'\x08\x0a\x00\x00'), nxt=6), nxt=41))), - ('outer ipv4 option', tunnel(tcp4, 4, hl=6, opt=b'\x94\x02\x00\x00')), # router alert, length 2 + ('inner ipv4 option', tunnel(make_ipv4(TCP_SEGMENT, options=b'\x94\x02\x00\x00'), 4)), ): with self.subTest(case=label): messages = self.decline(make_packet(frame)) @@ -1328,5 +1380,156 @@ def test_engine_warns_once_per_capture_on_tunnelled_tcp(self) -> None: close_extractor(extractor) +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +@unittest.skipUnless(HAS_PYPCAPFILE, 'pypcapfile not installed or not importable') +class PyPCAPFileHeaderCheckAgreementTests(unittest.TestCase): + """Frames whose headers PyPCAPFile reads but does not check, against the default engine. + + C.f. #1590 (Data Offset below 5), #1591 (IHL past the packet), #1592 (IPv4 in + IPv6), #1593 (TCP behind AH over IPv4), #1596 (IPv4 options) and #1597 (TCP + behind IPv6 extension headers in IPv4). + + """ + + ASPECTS = PyPCAPFileToolkitAgainstRealDecodersTests.ASPECTS + read = PyPCAPFileToolkitAgainstRealDecodersTests.read + read_both = PyPCAPFileToolkitAgainstRealDecodersTests.read_both + + def setUp(self) -> None: + reimport_once_per_class(self) + + def agree(self, frames, **want) -> dict: + """:meth:`read_both` ``frames``; check each aspect agrees, and pypcapfile's inputs are ``want``.""" + mine, theirs = self.read_both(frames) + for kind, numbers in want.items(): + self.assertEqual(sorted(mine[kind]), numbers, kind) + for aspect in self.ASPECTS: + with self.subTest(aspect=aspect): + self.assertEqual(mine[aspect], theirs[aspect]) + return mine + + def test_data_offset_below_5_agrees(self) -> None: + segment = wire.tcp(b'data', seq=1) + self.agree([ + wire.ethernet(wire.ipv4(data_offset(segment, 4), proto=6), 0x0800), + tagged(wire.ipv4(data_offset(segment, 0), proto=6), 0x0800, C_TAG), + wire.ethernet(wire.ipv4(segment, proto=6), 0x0800), + ], tcp=[3], trace=[3]) + + def test_ipv4_header_past_the_packet_agrees(self) -> None: + self.agree([ + wire.ethernet(ihl(wire.ipv4(wire.tcp(b'', seq=1), proto=6), 15), 0x0800), + tagged(ihl(wire.ipv4(wire.tcp(b'', seq=1), proto=6), 15), 0x0800, C_TAG), + wire.ethernet(ihl(wire.ipv4(b'', proto=17), 6), 0x0800), + wire.ethernet(wire.ipv4(b'', proto=17), 0x0800), + ], ipv4=[4], tcp=[]) + + def test_ipv4_in_ipv6_agrees(self) -> None: + datagram = wire.udp(b'u' * 24) + first = wire.ipv4(datagram[:16], proto=17, mf=True, ident=5) + frames = [ + # one datagram, in two fragments, each in IPv6 + wire.ethernet(wire.ipv6(first, nxt=4), 0x86DD), + tagged(wire.ipv6(wire.ipv4(datagram[16:], proto=17, offset=16, ident=5), nxt=4), + 0x86DD, C_TAG), + wire.ethernet(wire.ipv6(bytes([4, 0, 1, 4, 0, 0, 0, 0]) + udp_fragment(), nxt=0), 0x86DD), + wire.ethernet(wire.ipv6(wire.ipv6(udp_fragment(), nxt=4), nxt=41), 0x86DD), + wire.ethernet(wire.ipv6(udp_fragment(), nxt=4), 0x86DD) + bytes(9), + wire.ethernet(wire.ipv6(make_ipv4(b'u' * 8, protocol=17, flags=1, + options=b'\x94\x04\x00\x00'), nxt=4), 0x86DD), + # none: DF set, a later IPv6 fragment, an IHL past the packet, options it rejects + wire.ethernet(wire.ipv6(wire.ipv4(wire.udp(b'x'), proto=17, df=True), nxt=4), 0x86DD), + wire.ethernet(wire.ipv6_fragment(udp_fragment(), offset=8, nxt=4), 0x86DD), + wire.ethernet(wire.ipv6(ihl(udp_fragment(), 15), nxt=4), 0x86DD), + wire.ethernet(wire.ipv6(make_ipv4(b'u' * 8, protocol=17, flags=1, + options=BAD_IPV4_OPTIONS[0]), nxt=4), 0x86DD), + ] + mine = self.agree(frames, ipv4=[1, 2, 3, 4, 5, 6]) + self.assertIn(datagram, [item['payload'] for item in mine['ipv4 datagrams'] if item['completed']]) + + def test_tcp_behind_ah_agrees(self) -> None: + def over_ah(segment: bytes, header: bytes = ah(6), **fields) -> bytes: + return wire.ethernet(wire.ipv4(header + segment, proto=51, **fields), 0x0800) + + frames = [ + # a flow behind AH, then behind two, tagged, and with Ethernet padding + over_ah(wire.tcp(b'', seq=0, syn=True)), + over_ah(wire.tcp(b'hello', seq=1, ack=1), ah(51) + ah(6, 1)), + tagged(wire.ipv4(ah(6) + wire.tcp(b'!', seq=6, ack=1), proto=51), 0x0800, C_TAG), + over_ah(wire.tcp(b'', seq=7, ack=1, fin=True)) + bytes(7), + over_ah(make_tcp(b'opts', options=b'\x02\x04\x05\xb4')), + # none: a Payload Length of 0, a Data Offset past the capture or below + # 5, an option the TCP parser rejects, a later fragment + over_ah(wire.tcp(b'x', seq=0), ah(6, 0)), + over_ah(data_offset(wire.tcp(b'x', seq=0), 9)), + over_ah(data_offset(wire.tcp(b'x', seq=0), 4)), + over_ah(make_tcp(b'x', options=b'\x02\x03\x00\x00')), + over_ah(wire.tcp(b'x', seq=0), offset=8), + ] + mine = self.agree(frames, tcp=[1, 2, 3, 4, 5], trace=[1, 2, 3, 4, 5]) + self.assertEqual(mine['tcp'][2]['payload'], b'hello') + + def test_tcp_behind_ipv6_extension_headers_in_ipv4_agrees(self) -> None: + # C.f. #1597: the default engine reads IPv4's protocol as IPv6 reads a + # Next Header, so dissects these headers there too, and the TCP after them. + def over(protocol: int, header: bytes, segment: bytes) -> bytes: + return wire.ethernet(wire.ipv4(header + segment, proto=protocol), 0x0800) + + frames = [over(protocol, header, wire.tcp(b'data', seq=number, sport=1000 + number)) + for number, (_, protocol, header) in enumerate(IPV4_EXTENSION_HEADERS)] + frames += [ + # two, then AH after one, tagged + over(0, options_header(60) + options_header(6), wire.tcp(b'data', seq=7)), + over(60, options_header(51) + ah(6), wire.tcp(b'data', seq=8)), + tagged(wire.ipv4(options_header(6) + wire.tcp(b'data', seq=9), proto=60), 0x0800, C_TAG), + # none: an option the default engine rejects, a header cut short, a HIP + # Header Length under 4, Shim6, and a TCP option behind it the TCP parser rejects + over(60, options_header(6, b'\x01\x09' + bytes(4)), wire.tcp(b'data', seq=10)), + wire.ethernet(wire.ipv4(options_header(6)[:4], proto=60), 0x0800), + wire.ethernet(wire.ipv4(bytes([6, 0, 0, 0]), proto=43), 0x0800), + over(139, bytes([6, 3, 0x01, 0x21]) + bytes(28), wire.tcp(b'data', seq=11)), + over(140, bytes([6, 0, 0x80, 0]) + bytes(4), wire.tcp(b'data', seq=12)), + over(60, options_header(6), make_tcp(b'x', options=b'\x02\x03\x00\x00')), + ] + self.agree(frames, tcp=list(range(1, 11)), trace=list(range(1, 11))) + + def test_ipv4_options_agree(self) -> None: + frames = [] + for options in (*GOOD_IPV4_OPTIONS, *BAD_IPV4_OPTIONS): + segment = make_tcp(b'data', src_port=1000 + len(frames)) + frames += [ + wire.ethernet(make_ipv4(segment, options=options), 0x0800), + tagged(make_ipv4(segment, options=options), 0x0800, C_TAG), + wire.ethernet(make_ipv4(b'x' * 16, protocol=17, flags=1, options=options), 0x0800), + ] + self.agree(frames, ipv4=list(range(1, 10)), tcp=[1, 2, 4, 5, 7, 8]) + + def test_engine_warns_of_a_tunnel_behind_ah(self) -> None: + from pcapkit import extract + + frames = ( + # a TCP header behind AH that the TCP parser rejects is no TCP at all + wire.ethernet(wire.ipv4(ah(6) + data_offset(TCP_SEGMENT, 9), proto=51), 0x0800), + wire.ethernet(wire.ipv4(ah(4) + wire.ipv4(TCP_SEGMENT, proto=6), proto=51), 0x0800), + ) + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, 'ah.pcap') + with open(path, 'wb') as file: + file.write(wire.pcap([(1, number, frame) for number, frame in enumerate(frames)])) + with 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, 'trace'), + trace_format='json') + try: + self.assertEqual(extractor._exnam, 'pypcapfile') + messages = [str(item.message) for item in caught if 'left out' in str(item.message)] + self.assertEqual(len(messages), 1, messages) + self.assertTrue(messages[0].startswith('Frame 2: TCP tunnelled in IP'), messages) + self.assertEqual(len(extractor.reassembly.tcp), 0) + finally: + close_extractor(extractor) + + if __name__ == '__main__': unittest.main()