diff --git a/docs/source/pcapkit/foundation/engines/index.rst b/docs/source/pcapkit/foundation/engines/index.rst index e7172cec71..efbf1110dd 100644 --- a/docs/source/pcapkit/foundation/engines/index.rst +++ b/docs/source/pcapkit/foundation/engines/index.rst @@ -92,9 +92,11 @@ 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; | -| | 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 | +| | TCP over IPv6, and TCP tunnelled in IP (e.g. 6in4, 4in4), are | +| | left out of TCP reassembly and flow tracing, each with an | +| | :class:`~pcapkit.utilities.warnings.AttributeWarning` once | +| | per capture; IPv4 behind 802.1Q/802.1ad VLAN tags is read as | +| | it is untagged; 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/docs/source/pcapkit/toolkit/3rdparty.rst b/docs/source/pcapkit/toolkit/3rdparty.rst index c46c680642..bbc3bece1f 100644 --- a/docs/source/pcapkit/toolkit/3rdparty.rst +++ b/docs/source/pcapkit/toolkit/3rdparty.rst @@ -184,6 +184,13 @@ cannot be used for it. :data:`None` -- which would be indistinguishable from "this frame carries no IPv6 fragment". + Nor does it decode a VLAN tag or a tunnel. The adapters read an IPv4 packet + behind 802.1Q and 802.1ad tags as `PyPCAPFile`_ reads an untagged one, while + TCP over IPv6 and TCP tunnelled in IP (e.g. 6in4, 4in4, 6in6) are left out of + :func:`~pcapkit.toolkit.pypcapfile.tcp_reassembly` and + :func:`~pcapkit.toolkit.pypcapfile.tcp_traceflow`, with an + :class:`~pcapkit.utilities.warnings.AttributeWarning` once per capture for each. + .. autofunction:: pcapkit.toolkit.pypcapfile.ipv4_reassembly .. autofunction:: pcapkit.toolkit.pypcapfile.ipv6_reassembly diff --git a/pcapkit/toolkit/pypcapfile.py b/pcapkit/toolkit/pypcapfile.py index 9d69fe3b91..0150f26247 100644 --- a/pcapkit/toolkit/pypcapfile.py +++ b/pcapkit/toolkit/pypcapfile.py @@ -25,10 +25,17 @@ :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. + `PyPCAPFile`_ decodes no VLAN tag either, so :func:`_network` reads past the + 802.1Q and 802.1ad tags the default engine steps over (see + :data:`VLAN_TAGS`), however deeply stacked, and decodes the IPv4 packet + behind them with `PyPCAPFile`_'s own decoder. TCP and IPv4 reassembly and TCP + flow tracing therefore see a tagged IPv4 frame as they see an untagged one. + Nor does it decode a tunnel, and TCP carried in IPv4 or IPv6 that is itself + carried in IP (e.g. 4in4, 6in4, 6in6) is left out with an + :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`). Note also that `PyPCAPFile`_ decoders *replace* the payload bytes of the layer they decode. :class:`~pcapkit.foundation.engines.pypcapfile.PyPCAPFile` therefore @@ -50,23 +57,23 @@ 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.linktype import LinkType as Enum_LinkType 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.protocols.schema.internet.ipv6 import jumbo_payload_length from pcapkit.utilities.exceptions import ProtocolError, UnsupportedCall, stacklevel from pcapkit.utilities.warnings import AttributeWarning, warn if TYPE_CHECKING: - from collections import OrderedDict + from collections import OrderedDict, deque from ipaddress import IPv4Address, IPv6Address from typing import Any, Optional from pcapfile.protocols.network.ip import IP from pcapfile.structs import pcap_packet as Packet - from pcapkit.const.reg.linktype import LinkType as Enum_LinkType - __all__ = [ 'packet2timestamp', 'ipv4_header', 'packet2chain', 'packet2dict', 'ipv4_reassembly', 'ipv6_reassembly', 'tcp_reassembly', 'tcp_traceflow', @@ -86,10 +93,33 @@ #: IPv4 **MF** (more fragments) bit, within the three-bit flags field. IPV4_FLAG_MF = 0b001 +#: Length of the fixed IPv4 header, i.e. with no options, :rfc:`791#section-3.1`. +IPV4_HEADER_LEN = 20 + #: 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 +#: 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__ +#: `). +VLAN_TAGS = frozenset(( + Enum_EtherType.Customer_VLAN_Tag_Type, + Enum_EtherType.IEEE_Std_802_1Q_Service_VLAN_tag_identifier, +)) + +#: Length of a VLAN tag (IEEE 802.1Q): two octets of tag control information, +#: then the EtherType of what follows the tag. +VLAN_TAG_LEN = 4 + +#: IP protocol numbers of the IP-in-IP tunnels the default engine dissects +#: (see :attr:`Internet.__proto__ +#: `), each mapped to the +#: IP version it carries: IPv4 (protocol 4, :rfc:`2003`) and IPv6 (protocol 41, +#: :rfc:`2473` and :rfc:`4213`). +IP_TUNNELS = {Enum_TransType.IPv4: 4, Enum_TransType.IPv6: 6} + +#: IPv6 extension headers that :func:`_upper_layer` 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 @@ -102,15 +132,32 @@ 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] +#: The warning :func:`_decline_tcp` gives for each kind of TCP that +#: :func:`_left_out` finds this engine cannot read. +TCP_LEFT_OUT = { + 'ipv6': "TCP over IPv6 is left out of TCP reassembly and flow tracing, as 'pypcapfile' " + 'has no IPv6 decoder', + 'tunnel': 'TCP tunnelled in IP (IPv4 or IPv6 carried in IPv4 or IPv6) is left out of TCP ' + "reassembly and flow tracing, as 'pypcapfile' decodes no tunnel", +} + +#: How many warnings :func:`_decline_tcp` remembers having given, each for one +#: kind in :data:`TCP_LEFT_OUT` and one capture. 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 read at the same time warn exactly +#: once each. +TCP_WARNED_LIMIT = 32 + +#: The warnings :func:`_decline_tcp` has given, oldest first. Each is keyed by +#: the key :func:`_capture_of` gives and the kind, and maps to the anchor that +#: keeps that key unique. +_tcp_warned = collections.OrderedDict() # type: OrderedDict[tuple[int, str], Any] + +#: The frame :func:`_network` last warned it could not decode: the key +#: :func:`_capture_of` gives, the packet index and the anchor that keeps the key +#: unique. The adapters each fetch the network layer of a frame in turn, so this +#: is what holds them to one warning per frame between them. +_undecoded_last = collections.deque(maxlen=1) # type: deque[tuple[int, int, Any]] def packet2timestamp(packet: 'Packet') -> 'float': @@ -302,11 +349,44 @@ def _is_raw(layer: 'Any') -> 'bool': return isinstance(layer, (bytes, bytearray, memoryview)) -def _network(packet: 'Packet') -> 'Optional[IP]': +def _untag(link: 'Any') -> 'Optional[tuple[int, bytes]]': + """Read past the VLAN tags of an Ethernet frame whose payload is undecoded. + + Args: + link: Decoded PyPCAPFile link layer. + + Returns: + The EtherType after the last of the tags in :data:`VLAN_TAGS` and the + raw octets it heads -- for an untagged frame, the frame's own EtherType + and payload. :data:`None` if `PyPCAPFile`_ decoded the payload, or a tag + is cut short. + + Note: + `PyPCAPFile`_ leaves a payload it has no decoder for hex-encoded (see + :func:`_maybe_unhex`), so this reads the raw octets and decodes nothing. + + .. _PyPCAPFile: https://github.com/kisom/pypcapfile + + """ + ethertype = getattr(link, 'type', None) # type: Optional[int] + payload = getattr(link, 'payload', None) # type: Any + if ethertype is None or not _is_raw(payload): + return None + + data = _maybe_unhex(bytes(payload)) + while ethertype in VLAN_TAGS: + if len(data) < VLAN_TAG_LEN: + return None + ethertype, data = int.from_bytes(data[2:4], 'big'), data[VLAN_TAG_LEN:] + return ethertype, data + + +def _network(packet: 'Packet', count: 'int' = -1) -> 'Optional[IP]': """Fetch the decoded IPv4 layer of a PyPCAPFile packet, if any. Args: packet: PyPCAPFile packet. + count: Packet index, for the warning below. Returns: The decoded :class:`~pcapfile.protocols.network.ip.IP` layer, or @@ -314,6 +394,24 @@ def _network(packet: 'Packet') -> 'Optional[IP]': case for every non-IPv4 frame, `PyPCAPFile`_ having no other network layer decoder. + Warns: + AttributeWarning: If an IPv4 packet behind VLAN tags will not decode, + in the words, and with the category, + :meth:`PyPCAPFile._decode ` + uses for an untagged one -- once per frame, however many adapters + fetch its network layer. + + Note: + `PyPCAPFile`_ decodes no VLAN tag, so it leaves an IPv4 packet behind + one undecoded. That packet is decoded here, past every tag that + :func:`_untag` reads past, with the same + :class:`~pcapfile.protocols.network.ip.IP` decoder and the same depth + :class:`~pcapfile.protocols.linklayer.ethernet.Ethernet` uses for an + untagged frame, so the two decode alike; one that will not decode is + left out with a warning, as an untagged one is. The engine warns of an + untagged frame as it reads it, and this of a tagged one as reassembly or + flow tracing does. + .. _PyPCAPFile: https://github.com/kisom/pypcapfile """ @@ -322,67 +420,152 @@ def _network(packet: 'Packet') -> 'Optional[IP]': return None payload = getattr(link, 'payload', None) - if type(payload).__name__ != 'IP': + if type(payload).__name__ == 'IP': + return payload + if getattr(link, 'type', None) not in VLAN_TAGS: return None - return payload + untagged = _untag(link) + if untagged is None or untagged[0] != Enum_EtherType.Internet_Protocol_version_4: + return None -def _ipv6_carries_tcp(packet: 'Packet') -> 'bool': - """Test if a PyPCAPFile frame is an IPv6 packet carrying TCP. + # NOTE: imported only once there is something to decode, as in + # :func:`tcp_reassembly`; ``layers=0`` stops at the network layer, as the + # engine does (see :attr:`PyPCAPFile.LAYERS + # `). + from pcapfile.protocols.network.ip import IP # isort:skip + + try: + return 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): + _undecoded_last.append((key, count, anchor)) + warn(f'Frame {count}: {Enum_LinkType.ETHERNET!r} decoding failed ({error!r}); ' + 'frame left undecoded', AttributeWarning, stacklevel=stacklevel()) + return None + + +def _upper_layer(version: 'int', data: 'bytes') -> 'Optional[tuple[int, bytes]]': + """Step over an IP header to the upper-layer protocol it carries. Args: - packet: PyPCAPFile packet. + version: IP version of the header ``data`` starts with, ``4`` or ``6``. + data: Raw octets of the IP 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. + The upper-layer protocol number and the octets from its header on, or + :data:`None` if no upper-layer header follows. + + 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 + 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 + Fragment header as if that header were not there. A later fragment of + either has no upper-layer header, since from its offset on it carries a + slice of the payload (:rfc:`791#section-3.2`, :rfc:`8200#section-4.5`). + Nor does a truncated header or header chain. 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 + Each of these rules is the default engine's own, as are the ones + :func:`_left_out` applies to the TCP header, so that the warning + :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 + 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. """ - link = packet.packet - if _is_raw(link) or getattr(link, 'type', None) != Enum_EtherType.Internet_Protocol_version_6: - return False + if version == 4: + if len(data) < IPV4_HEADER_LEN or data[0] >> 4 != 4: + return None + ihl = (data[0] & 0x0F) * 4 + if ihl < IPV4_HEADER_LEN or len(data) < ihl: + return None + 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:] - 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 - + return None + extent = struct.unpack_from('!H', data, 4)[0] + if not extent: + extent = jumbo_payload_length(data[6], data[IPV6_HEADER_LEN:]) or 0 + data = data[:IPV6_HEADER_LEN + extent] nxt, offset = data[6], IPV6_HEADER_LEN while nxt in IPV6_EXTENSION_HEADERS: if len(data) < offset + 8: - return False + return None 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 + return None 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 + if len(data) < offset: + return None + return nxt, data[offset:] + + +def _left_out(packet: 'Packet', ipv4: 'Optional[IP]') -> 'Optional[str]': + """Name the TCP a PyPCAPFile frame carries that this engine cannot read. + + Args: + packet: PyPCAPFile packet, which :func:`_transport` found no TCP segment in. + ipv4: Its IPv4 layer, as :func:`_network` fetches it. + + Returns: + ``'ipv6'`` if the frame is IPv6 carrying TCP, directly or behind + extension headers; ``'tunnel'`` if it is IPv4 or IPv6 carrying TCP in one + or more of the :data:`IP_TUNNELS`; otherwise :data:`None`. Either may be + behind VLAN tags (see :func:`_untag`), and each IP header is stepped over + as :func:`_upper_layer` does. An ESP payload cannot be seen into, so does + not count. + + The TCP header has to be at least :data:`TCP_MIN_HEADER_LEN` octets + long, as its Data Offset gives it, with that many captured. In a tunnel, + the whole of the header its Data Offset gives has to be there too, as far + as the inner IP packet's own length: the default engine's TCP parser + rejects one cut short, e.g. by the snapshot length, and only directly + over IPv6 does it then read the segment out of the raw octets left + behind (:func:`pcapkit.toolkit.pcap.tcp_segment`, :issue:`1518`). + + """ + 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): + 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) + if untagged is None or untagged[0] != Enum_EtherType.Internet_Protocol_version_6: + return None + upper = _upper_layer(6, untagged[1]) + + kind = 'ipv6' + while upper is not None and upper[0] in IP_TUNNELS: + upper, kind = _upper_layer(IP_TUNNELS[upper[0]], upper[1]), 'tunnel' + 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 or (kind == 'tunnel' and header > len(segment)): + return None + return kind def _capture_of(packet: 'Packet') -> 'tuple[int, Any]': @@ -413,38 +596,67 @@ def _capture_of(packet: 'Packet') -> 'tuple[int, Any]': 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. +def _decline_tcp(packet: 'Packet', ipv4: 'Optional[IP]', count: 'int') -> 'None': + """Warn that a frame carrying TCP this engine cannot read is left out. Args: - packet: PyPCAPFile packet, which :func:`_network` found no IPv4 layer in. + packet: PyPCAPFile packet, which :func:`_transport` found no TCP segment in. + ipv4: Its IPv4 layer, as :func:`_network` fetches it. 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. + AttributeWarning: If ``packet`` carries TCP that :func:`_left_out` names, + and is the first frame of its capture carrying that kind of TCP to + reach :func:`tcp_reassembly` or :func:`tcp_traceflow`. Later frames + of the same capture and kind are left out without another warning, + and every other capture or kind warns once of its own, however their + frames interleave. Only the :data:`TCP_WARNED_LIMIT` warnings given + most recently are remembered, so with more than that read at once a + forgotten one is given again: a warning can repeat, but is never + omitted. """ key, anchor = _capture_of(packet) - if key in _ipv6_tcp_warned: + if all((key, kind) in _tcp_warned for kind in TCP_LEFT_OUT): return - if not _ipv6_carries_tcp(packet): + kind = _left_out(packet, ipv4) + if kind is None or (key, kind) in _tcp_warned: 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 " + _tcp_warned[key, kind] = anchor + while len(_tcp_warned) > TCP_WARNED_LIMIT: + _tcp_warned.popitem(last=False) + warn(f'Frame {count}: {TCP_LEFT_OUT[kind]}; later such frames of this capture are left out ' 'without another warning', AttributeWarning, stacklevel=stacklevel()) +def _ipv4_payload(ipv4: 'IP') -> 'bytes': + """Fetch the raw payload of a PyPCAPFile IPv4 packet, as far as its Total Length. + + Args: + ipv4: PyPCAPFile IPv4 packet, whose payload `PyPCAPFile`_ left undecoded. + + Returns: + The octets after the header, up to the Total Length. A Total Length of + ``0``, which TCP segmentation offload leaves in a capture taken on the + sending host, leaves the rest of the frame; one shorter than the header + leaves none (:issue:`1547`). + + Note: + `PyPCAPFile`_ keeps every octet after the header as the payload, so a + frame padded to the Ethernet minimum carries its padding there. The + default engine reads such octets as a trailer, outside the payload + (:issue:`1209`), and so does this (:issue:`1577`). + + .. _PyPCAPFile: https://github.com/kisom/pypcapfile + + """ + payload = _maybe_unhex(bytes(ipv4.payload)) + if not ipv4.len: + return payload + return payload[:max(ipv4.len - ipv4.hl * 4, 0)] + + def _transport(ipv4: 'IP') -> 'Optional[bytes]': """Fetch the verbatim TCP segment carried by a PyPCAPFile IPv4 packet. @@ -452,18 +664,19 @@ def _transport(ipv4: 'IP') -> 'Optional[bytes]': ipv4: PyPCAPFile IPv4 packet. Returns: - The raw TCP segment, or :data:`None` if the packet does not carry a - TCP payload long enough to hold a header. + 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`). """ - if ipv4.p != Enum_TransType.TCP: + if ipv4.p != Enum_TransType.TCP or ipv4.off: return None - - segment = ipv4.payload - if not _is_raw(segment): + if not _is_raw(ipv4.payload): return None - segment = _maybe_unhex(bytes(segment)) + segment = _ipv4_payload(ipv4) if len(segment) < TCP_MIN_HEADER_LEN: return None return segment @@ -619,8 +832,8 @@ def ipv4_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'IP_Packet[IPv4Ad Data for IPv4 reassembly. * 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`) and the - **DF** flag is :data:`False`. + 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`. * 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`. @@ -632,7 +845,7 @@ def ipv4_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'IP_Packet[IPv4Ad :class:`pcapkit.foundation.reassembly.ipv4.IPv4` """ - ipv4 = _network(packet) + ipv4 = _network(packet, count) if ipv4 is None: return None if ipv4.flags & IPV4_FLAG_DF: # dismiss not fragmented packet @@ -686,8 +899,9 @@ 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; TCP over IPv6 cannot - be, as `PyPCAPFile`_ has no IPv6 decoder. + 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, + 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`. @@ -696,10 +910,11 @@ def tcp_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'TCP_Packet | None 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. + AttributeWarning: If ``packet`` carries TCP over IPv6, or tunnelled in IP, + 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; each of the two kinds warns once of its own. See Also: :class:`pcapkit.foundation.reassembly.tcp.TCP` @@ -707,13 +922,9 @@ def tcp_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'TCP_Packet | None .. _PyPCAPFile: https://github.com/kisom/pypcapfile """ - ipv4 = _network(packet) - if ipv4 is None: - _decline_ipv6_tcp(packet, count) - return None - - segment = _transport(ipv4) - if segment is None: + ipv4 = _network(packet, count) + if ipv4 is None or (segment := _transport(ipv4)) is None: + _decline_tcp(packet, ipv4, count) return None # NOTE: imported only once there is something to decode, so that declining a @@ -759,8 +970,9 @@ 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; TCP over IPv6 cannot - be, as `PyPCAPFile`_ has no IPv6 decoder. + 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, + 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`. @@ -769,10 +981,11 @@ def tcp_traceflow(packet: 'Packet', *, data_link: 'Enum_LinkType', 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. + AttributeWarning: If ``packet`` carries TCP over IPv6, or tunnelled in IP, + 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; each of the two kinds warns once of its own. See Also: :class:`pcapkit.foundation.traceflow.tcp.TCP` @@ -780,13 +993,9 @@ def tcp_traceflow(packet: 'Packet', *, data_link: 'Enum_LinkType', .. _PyPCAPFile: https://github.com/kisom/pypcapfile """ - ipv4 = _network(packet) - if ipv4 is None: - _decline_ipv6_tcp(packet, count) - return None - - segment = _transport(ipv4) - if segment is None: + ipv4 = _network(packet, count) + if ipv4 is None or (segment := _transport(ipv4)) is None: + _decline_tcp(packet, ipv4, count) return None # NOTE: imported only once there is something to decode, so that declining a diff --git a/tests/toolkit/test_pypcapfile_tcp_last_unit.py b/tests/toolkit/test_pypcapfile_tcp_last_unit.py index 24e0d1075e..21f36ac398 100644 --- a/tests/toolkit/test_pypcapfile_tcp_last_unit.py +++ b/tests/toolkit/test_pypcapfile_tcp_last_unit.py @@ -39,6 +39,9 @@ class FakeIP: """Stand-in for :class:`pcapfile.protocols.network.ip.IP`.""" def __init__(self, payload: bytes) -> None: + self.hl = 5 # header length, in 4-octet words + self.len = 20 + len(payload) # total length + self.off = 0 # fragment offset: a first fragment self.p = 6 self.src = b'10.1.1.2' self.dst = b'10.1.1.3' diff --git a/tests/toolkit/test_pypcapfile_unit.py b/tests/toolkit/test_pypcapfile_unit.py index 9ea26a00f3..0a345948bd 100644 --- a/tests/toolkit/test_pypcapfile_unit.py +++ b/tests/toolkit/test_pypcapfile_unit.py @@ -150,6 +150,80 @@ def ipv6_frame(packet: bytes) -> FakeEthernet: #: A TCP segment with a full header and a payload. TCP_SEGMENT = wire.tcp(b'payload', seq=1, syn=True) +#: An IPv6 jumbogram carrying :data:`TCP_SEGMENT`: a Payload Length of 0, and a +#: Hop-by-Hop Options header whose Jumbo Payload option gives the real one (RFC 2675). +JUMBOGRAM = (wire.ipv6(b'', nxt=0) + + bytes([6, 0, 0xC2, 4]) + (8 + len(TCP_SEGMENT)).to_bytes(4, 'big') + TCP_SEGMENT) + +#: TPIDs of the 802.1Q customer tag and the 802.1ad service tag. +C_TAG, S_TAG = 0x8100, 0x88A8 + + +def vlan_tags(ethertype: int, *tpids: int) -> bytes: + """The octets after the first TPID of an Ethernet frame tagged once per ``tpids``. + + That is, each tag's control information and the EtherType after it -- the + next tag's TPID, or ``ethertype`` after the last tag. + + """ + return b''.join((100 + number).to_bytes(2, 'big') + kind.to_bytes(2, 'big') + for number, kind in enumerate((*tpids[1:], ethertype))) + + +def tagged(packet: bytes, ethertype: int, *tpids: int) -> bytes: + """A raw Ethernet frame carrying ``packet`` behind one VLAN tag per ``tpids``.""" + return wire.ethernet(vlan_tags(ethertype, *tpids) + packet, tpids[0]) + + +def tagged_frame(packet: bytes, ethertype: int, *tpids: int) -> FakeEthernet: + """A tagged frame as PyPCAPFile leaves it: type the first TPID, payload hex-encoded.""" + frame = FakeEthernet(binascii.hexlify(vlan_tags(ethertype, *tpids) + packet)) + frame.type = tpids[0] + return frame + + +def fake_ip_decoder() -> dict[str, types.ModuleType]: + """Stand-in modules for :mod:`pcapfile.protocols.network.ip`, for :data:`sys.modules`. + + Their ``IP`` records what it was given, and refuses what is not an IPv4 + header of at least 20 octets as the real decoder does. + + """ + class IP: + """Stand-in for :class:`pcapfile.protocols.network.ip.IP`.""" + + calls: list = [] + + def __init__(self, packet: bytes, layers: int = 0) -> None: + self.calls.append((packet, layers)) + if packet[0] >> 4 != 4 or packet[0] & 0x0F <= 4: + raise AssertionError('not an IPv4 packet.') + struct.unpack('!BBHHHBBHII', packet[:20]) + + modules = {name: types.ModuleType(name) for name in ( + 'pcapfile', 'pcapfile.protocols', 'pcapfile.protocols.network', + 'pcapfile.protocols.network.ip')} + modules['pcapfile.protocols.network.ip'].IP = IP + return modules + + +def data_offset(segment: bytes, words: int) -> bytes: + """``segment``, a TCP segment, with its Data Offset set to ``words``.""" + return segment[:12] + bytes([words << 4 | segment[12] & 0x0F]) + segment[13:] + + +def total_length(packet: bytes, value: int) -> bytes: + """``packet``, an IPv4 packet with no options, with its Total Length set to ``value``.""" + header = bytearray(packet[:20]) + header[2:4], header[10:12] = value.to_bytes(2, 'big'), b'\x00\x00' + header[10:12] = wire.checksum(bytes(header)).to_bytes(2, 'big') + return bytes(header) + packet[20:] + + +def tunnel(packet: bytes, protocol: int, **fields) -> FakeEthernet: + """An IPv4 frame tunnelling ``packet``, as PyPCAPFile decodes it: payload hex-encoded.""" + return FakeEthernet(FakeIP(binascii.hexlify(packet), p=protocol, **fields)) + @unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') class PyPCAPFileToolkitTests(unittest.TestCase): @@ -392,11 +466,36 @@ def test_tcp_adapters_decline_non_tcp_and_truncated_segments(self) -> None: ('not tcp', make_packet(FakeEthernet(FakeIP(b'x' * 40, p=17)))), ('too short', make_packet(FakeEthernet(FakeIP(b'short')))), ('already decoded', make_packet(FakeEthernet(FakeIP(object())))), + # data from mid-datagram, which only looks like a TCP header (#1576) + ('later fragment', make_packet(FakeEthernet(FakeIP(make_tcp(b'x' * 20), off=185)))), ): with self.subTest(case=label): self.assertIsNone(tcp_reassembly(packet, count=1)) self.assertIsNone(tcp_traceflow(packet, data_link=LinkType.ETHERNET, count=1)) + def test_transport_takes_the_first_fragment_to_the_total_length(self) -> None: + from pcapkit.toolkit.pypcapfile import _transport + + segment = make_tcp(b'x') + for label, ipv4, want in ( + # #1577: octets past the Total Length, e.g. Ethernet padding, are not payload + ('padded', FakeIP(binascii.hexlify(segment + bytes(5)), len=20 + len(segment)), segment), + ('options, padded', + FakeIP(binascii.hexlify(segment + bytes(3)), hl=6, len=24 + len(segment)), segment), + ('total length 0, as TSO leaves it', + FakeIP(binascii.hexlify(segment + b'more'), len=0), segment + b'more'), + ('total length past the capture', FakeIP(binascii.hexlify(segment), len=200), segment), + ('total length shorter than the header', FakeIP(binascii.hexlify(segment), len=12), None), + ('total length cuts the tcp header short', FakeIP(binascii.hexlify(segment), len=30), None), + # #1576: a later fragment carries no TCP header, however it looks + ('last fragment', FakeIP(binascii.hexlify(segment), off=1), None), + ('later fragment, more to follow', + FakeIP(binascii.hexlify(segment), off=185, flags=0b001), None), + ('first fragment', FakeIP(binascii.hexlify(segment), flags=0b001), segment), + ): + with self.subTest(case=label): + self.assertEqual(_transport(ipv4), want) + @unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') class PyPCAPFileTCPOverIPv6Tests(unittest.TestCase): @@ -449,10 +548,10 @@ def test_the_oldest_capture_is_forgotten_beyond_the_limit(self) -> None: # 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)()): + with mock.patch.object(toolkit, 'TCP_WARNED_LIMIT', 2), \ + mock.patch.object(toolkit, '_tcp_warned', type(toolkit._tcp_warned)()): messages = self.decline(first, second, third, first_again, third_again) - self.assertEqual(len(toolkit._ipv6_tcp_warned), 2) + self.assertEqual(len(toolkit._tcp_warned), 2) self.assertEqual([message.split(':')[0] for message in messages], ['Frame 1', 'Frame 2', 'Frame 3', 'Frame 4']) @@ -543,6 +642,268 @@ def test_capture_of_keys_a_ctypes_header_by_its_address(self) -> None: self.assertEqual(_capture_of(packet), (id(value), value)) +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class PyPCAPFileEncapsulatedTCPTests(unittest.TestCase): + """TCP behind VLAN tags or in IP tunnels is read or warned about. C.f. #1537.""" + + decline = PyPCAPFileTCPOverIPv6Tests.decline + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_vlan_tagged_tcp_over_ipv6_warns(self) -> None: + hop_by_hop = bytes([6, 0]) + b'\x01\x04' + bytes(4) + for label, frame in ( + ('802.1Q', tagged_frame(wire.ipv6(TCP_SEGMENT, nxt=6), 0x86DD, C_TAG)), + ('802.1ad', tagged_frame(wire.ipv6(TCP_SEGMENT, nxt=6), 0x86DD, S_TAG)), + ('QinQ', tagged_frame(wire.ipv6(TCP_SEGMENT, nxt=6), 0x86DD, S_TAG, C_TAG)), + ('two customer tags', tagged_frame(wire.ipv6(TCP_SEGMENT, nxt=6), 0x86DD, C_TAG, C_TAG)), + ('802.1Q, hop-by-hop', tagged_frame(wire.ipv6(hop_by_hop + TCP_SEGMENT, nxt=0), 0x86DD, + C_TAG)), + ): + with self.subTest(case=label): + messages = self.decline(make_packet(frame)) + self.assertEqual(len(messages), 1, messages) + self.assertTrue(messages[0].startswith('Frame 1: TCP over IPv6 is left out'), + messages) + + def test_tunnelled_tcp_warns(self) -> None: + tcp4, tcp6 = wire.ipv4(TCP_SEGMENT, proto=6), wire.ipv6(TCP_SEGMENT, nxt=6) + hop_by_hop = bytes([6, 0]) + b'\x01\x04' + bytes(4) + for label, frame in ( + ('4in4', tunnel(tcp4, 4)), + ('6in4', tunnel(tcp6, 41)), + ('6in6', ipv6_frame(wire.ipv6(tcp6, nxt=41))), + ('4in6', ipv6_frame(wire.ipv6(tcp4, nxt=4))), + ('4in4in4', tunnel(wire.ipv4(tcp4, proto=4), 4)), + ('6in4in6', ipv6_frame(wire.ipv6(wire.ipv4(tcp6, proto=41), nxt=4))), + ('outer first fragment', tunnel(tcp4, 4, flags=0b001)), + ('inner ipv4 options', tunnel(make_ipv4(TCP_SEGMENT, options=b'\x01\x01\x01\x00'), 4)), + ('inner first fragment', tunnel(wire.ipv4(TCP_SEGMENT, proto=6, mf=True), 4)), + ('inner total length 0', tunnel(total_length(tcp4, 0), 4)), + ('outer padded', tunnel(tcp4 + bytes(6), 4, len=20 + len(tcp4))), + ('inner hop-by-hop', tunnel(wire.ipv6(hop_by_hop + TCP_SEGMENT, nxt=0), 41)), + ('tunnel behind hop-by-hop', + ipv6_frame(wire.ipv6(bytes([41, 0]) + b'\x01\x04' + bytes(4) + tcp6, nxt=0))), + ('6in6 behind 802.1Q', tagged_frame(wire.ipv6(tcp6, nxt=41), 0x86DD, C_TAG)), + ('4in6 behind QinQ', tagged_frame(wire.ipv6(tcp4, nxt=4), 0x86DD, S_TAG, C_TAG)), + # the default engine's TCP parser accepts these headers + ('tcp options', tunnel(wire.ipv4(make_tcp(b'x', options=b'\x02\x04\x05\xb4'), proto=6), 4)), + ('an unknown tcp option', tunnel(wire.ipv4(make_tcp(b'x', options=b'\xfe\x04\x00\x00'), + proto=6), 4)), + ('inner ipv6 jumbogram', tunnel(JUMBOGRAM, 41)), + ('outer ipv4 option', tunnel(tcp4, 4, hl=6, opt=b'\x94\x04\x00\x00')), # router alert + ): + with self.subTest(case=label): + messages = self.decline(make_packet(frame)) + self.assertEqual(len(messages), 1, messages) + self.assertTrue(messages[0].startswith('Frame 1: TCP tunnelled in IP'), messages) + self.assertIn("'pypcapfile' decodes no tunnel", messages[0]) + + def test_no_warning_where_the_default_engine_finds_no_tcp(self) -> None: + tcp4, tcp6 = wire.ipv4(TCP_SEGMENT, proto=6), wire.ipv6(TCP_SEGMENT, nxt=6) + arp = b'\x00\x01\x08\x00' + bytes(24) + decoded = tagged_frame(b'', 0x86DD, C_TAG) + decoded.payload = object() # a decoder PyPCAPFile does not have yet + cut_short = FakeEthernet(binascii.hexlify(b'\x00\x64\x86')) # three octets of a tag + cut_short.type = C_TAG + for label, frame in ( + ('not tunnelled', FakeEthernet(FakeIP(b'x' * 40, p=17))), + ('tunnelled udp', tunnel(wire.ipv4(wire.udp(b'x'), proto=17), 4)), + ('gre', tunnel(bytes(4) + tcp4, 47)), + ('ipv4 payload decoded', FakeEthernet(FakeIP(object(), p=4))), + # a later fragment's data is a slice of the payload, not a header + ('outer later fragment', tunnel(tcp4, 4, off=3)), + ('inner later fragment', tunnel(wire.ipv4(TCP_SEGMENT, proto=6, offset=8), 4)), + ('inner ipv6 later fragment', tunnel(wire.ipv6_fragment(TCP_SEGMENT, offset=8, nxt=6), 41)), + ('inner not version 4', tunnel(b'\x65' + tcp4[1:], 4)), + ('inner header under 20 octets', tunnel(b'\x44' + tcp4[1:], 4)), + ('inner header overruns the packet', tunnel(b'\x4f' + tcp4[1:], 4)), + ('inner ipv4 truncated', tunnel(tcp4[:19], 4)), + ('inner ipv6 truncated', tunnel(tcp6[:39], 41)), + ('inner tcp truncated', tunnel(wire.ipv4(TCP_SEGMENT[:19], proto=6), 4)), + # the Total Length ends the packet inside the TCP header, though more was captured + ('inner total length cuts the tcp header short', tunnel(total_length(tcp4, 30), 4)), + ('outer total length cuts the tunnel short', tunnel(tcp4, 4, len=40)), + ('inner payload length cuts the tcp header short', + tunnel(tcp6[:4] + (10).to_bytes(2, 'big') + tcp6[6:], 41)), + # a Payload Length of 0 with no Jumbo Payload option leaves no payload + ('inner payload length 0, no jumbo option', tunnel(tcp6[:4] + b'\x00\x00' + tcp6[6:], 41)), + ('ipv6 payload length 0, no jumbo option', ipv6_frame(tcp6[:4] + b'\x00\x00' + tcp6[6:])), + # in a tunnel, a header the default engine's TCP parser rejects is no TCP + ('tunnelled data offset past the capture', + tunnel(wire.ipv4(data_offset(TCP_SEGMENT, 9), proto=6), 4)), + ('tunnelled data offset past the inner total length', + 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 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), + 0x86DD, C_TAG)), + ('inner esp', ipv6_frame(wire.ipv6(wire.ipv6(bytes(8) + TCP_SEGMENT, nxt=50), nxt=41))), + ('tunnel header overruns the frame', + ipv6_frame(wire.ipv6(bytes([41, 200]) + bytes(6) + tcp6, nxt=0))), + ('802.1Q, then arp', tagged_frame(arp, 0x0806, C_TAG)), + ('802.1Q, payload decoded', decoded), + ('802.1Q cut short', cut_short), + # 0x9100 is a pre-standard QinQ TPID, which the default engine does not dissect + ('0x9100 tag', tagged_frame(tcp6, 0x86DD, 0x9100)), + ): + with self.subTest(case=label): + self.assertEqual(self.decline(make_packet(frame)), []) + + def test_network_decodes_the_ipv4_behind_the_tags(self) -> None: + from unittest import mock + + from pcapkit.toolkit.pypcapfile import _network + + packet = wire.ipv4(TCP_SEGMENT, proto=6) + with mock.patch.dict('sys.modules', fake_ip_decoder()) as modules: + IP = modules['pcapfile.protocols.network.ip'].IP + ipv4 = _network(make_packet(tagged_frame(packet, 0x0800, S_TAG, C_TAG))) + self.assertIsInstance(ipv4, IP) + self.assertEqual(IP.calls, [(packet, 0)]) # every tag stripped, network layer only + + def test_malformed_ipv4_behind_tags_warns_once_per_frame(self) -> None: + # As PyPCAPFile._decode warns of an untagged frame it cannot decode, + # however many adapters then fetch the tagged frame's network layer. + from unittest import mock + + from pcapkit.const.reg.linktype import LinkType + from pcapkit.toolkit.pypcapfile import ipv4_reassembly, tcp_reassembly, tcp_traceflow + from pcapkit.utilities.warnings import AttributeWarning + + bad = tagged_frame(b'\x44' + wire.ipv4(TCP_SEGMENT, proto=6)[1:], 0x0800, C_TAG) # IHL 4 + other = FakeEthernet(FakeIP(b'x' * 40, p=17, flags=0b010)) # decoded, nothing to warn of + packets = [make_packet(frame) for frame in (bad, bad, other, bad)] + for packet in packets[1:]: + packet.header = packets[0].header # one capture + with mock.patch.dict('sys.modules', fake_ip_decoder()), \ + warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + for count, packet in enumerate(packets, start=1): + self.assertIsNone(ipv4_reassembly(packet, count=count)) + self.assertIsNone(tcp_reassembly(packet, count=count)) + self.assertIsNone(tcp_traceflow(packet, data_link=LinkType.ETHERNET, count=count)) + messages = [str(item.message) for item in caught if issubclass(item.category, AttributeWarning)] + self.assertEqual(messages, [ + f"Frame {count}: decoding failed (AssertionError('not an IPv4 " + "packet.')); frame left undecoded" for count in (1, 2, 4) + ]) + + def test_tcp_over_ipv6_the_tcp_parser_rejects_still_warns(self) -> None: + # Over IPv6, unlike in a tunnel, the default engine reads such a header + # out of the raw octets its TCP parser leaves. + for label, frame in ( + ('data offset past the capture', ipv6_frame(wire.ipv6(data_offset(TCP_SEGMENT, 9), nxt=6))), + ('option of length 0', ipv6_frame(wire.ipv6(make_tcp(b'x', options=b'\x02\x00\x05\xb4'), + nxt=6))), + ('jumbogram', ipv6_frame(JUMBOGRAM)), + ): + with self.subTest(case=label): + messages = self.decline(make_packet(frame)) + self.assertEqual(len(messages), 1, messages) + self.assertTrue(messages[0].startswith('Frame 1: TCP over IPv6'), messages) + + def test_malformed_tunnelled_tcp_the_header_walk_cannot_see_into_still_warns(self) -> None: + # The residual _upper_layer documents: the default engine rejects these + # options, and so finds no TCP, but nothing short of dissecting the frame + # would show that -- the warning errs on the side of being given. + tcp4 = wire.ipv4(TCP_SEGMENT, proto=6) + for label, frame in ( + ('tunnelled tcp option of length 0', + 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 + ): + with self.subTest(case=label): + messages = self.decline(make_packet(frame)) + self.assertEqual(len(messages), 1, messages) + self.assertTrue(messages[0].startswith('Frame 1: TCP tunnelled in IP'), messages) + + def test_detection_leaves_the_warnings_filters_and_the_logger_alone(self) -> None: + # The warnings filters and the logger are process-wide state of the + # caller's: telling a frame apart must neither change them nor reset + # what they have already seen. + import logging + + from pcapkit.const.reg.linktype import LinkType + from pcapkit.toolkit.pypcapfile import tcp_reassembly, tcp_traceflow + + logger = logging.getLogger('pcapkit') + level = logger.level + frames = (ipv6_frame(wire.ipv6(wire.ipv6(data_offset(TCP_SEGMENT, 9), nxt=6), nxt=41)), + tunnel(wire.ipv4(make_tcp(b'x', options=b'\x02\x00\x05\xb4'), proto=6), 4), + ipv6_frame(wire.ipv6(TCP_SEGMENT, nxt=6))) + + def user_warning() -> None: + warnings.warn('a warning of the caller', UserWarning) + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('default') + filters = list(warnings.filters) + user_warning() + # the adapters called as the engine calls them, with nothing between + for count, frame in enumerate(frames, start=1): + packet = make_packet(frame) + tcp_reassembly(packet, count=count) + tcp_traceflow(packet, data_link=LinkType.ETHERNET, count=count) + self.assertEqual(warnings.filters, filters) + self.assertEqual(logger.level, level) + user_warning() # the same call site, so the 'default' action shows it only once + self.assertEqual([str(item.message) for item in caught if item.category is UserWarning], + ['a warning of the caller']) + + def test_each_frame_is_walked_once_per_adapter_call(self) -> None: + # Declining costs one header walk per frame and adapter, however the + # frame is malformed: nothing is retried or dissected again. + from unittest import mock + + from pcapkit.const.reg.linktype import LinkType + from pcapkit.protocols.protocol import ProtocolBase + from pcapkit.toolkit import pypcapfile as toolkit + + built = [] # type: list[str] + init = ProtocolBase.__init__ + + def spy_init(this, *args, **kwargs): + built.append(type(this).__name__) + return init(this, *args, **kwargs) + + cut = wire.ipv6(wire.ipv6(make_tcp(b'', options=b'\x02\x04\x05\xb4'), nxt=6), nxt=41)[:-4] + packets = [make_packet(ipv6_frame(cut)) for _ in range(10)] # a TCP header cut short in a tunnel + for packet in packets[1:]: + packet.header = packets[0].header # one capture, which never warns + with mock.patch.object(toolkit, '_upper_layer', wraps=toolkit._upper_layer) as walk, \ + mock.patch.object(ProtocolBase, '__init__', spy_init), \ + warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + for count, packet in enumerate(packets, start=1): + self.assertIsNone(toolkit.tcp_reassembly(packet, count=count)) + self.assertIsNone(toolkit.tcp_traceflow(packet, data_link=LinkType.ETHERNET, count=count)) + self.assertEqual([str(item.message) for item in caught], []) + # the outer IPv6 header and the inner one, per frame and adapter ... + self.assertEqual(walk.call_count, 2 * 2 * len(packets)) + # ... and no protocol of the default engine's dissected along the way + self.assertEqual(built, []) + + def test_each_kind_warns_once_per_capture(self) -> None: + frames = (ipv6_frame(wire.ipv6(TCP_SEGMENT, nxt=6)), + ipv6_frame(wire.ipv6(wire.ipv6(TCP_SEGMENT, nxt=6), nxt=41)), + tagged_frame(wire.ipv6(TCP_SEGMENT, nxt=6), 0x86DD, C_TAG), + tunnel(wire.ipv4(TCP_SEGMENT, proto=6), 4)) + packets = [make_packet(frame) for frame in frames] + for packet in packets[1:]: + packet.header = packets[0].header # all four frames of one capture + + messages = self.decline(*packets) + self.assertEqual(len(messages), 2, messages) + self.assertTrue(messages[0].startswith('Frame 1: TCP over IPv6'), messages) + self.assertTrue(messages[1].startswith('Frame 2: TCP tunnelled in IP'), messages) + + @unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') @unittest.skipUnless(HAS_PYPCAPFILE, 'pypcapfile not installed or not importable') class PyPCAPFileToolkitAgainstRealDecodersTests(unittest.TestCase): @@ -721,6 +1082,251 @@ def test_engine_warns_once_per_capture_on_tcp_over_ipv6(self) -> None: finally: close_extractor(extractor) + def test_network_decodes_behind_vlan_tags_as_untagged(self) -> None: + from pcapfile.protocols.linklayer.ethernet import Ethernet + + from pcapkit.toolkit.pypcapfile import _ipv4_2dict, _maybe_unhex, _network + + packet = make_ipv4(make_tcp(b'body'), options=b'\x94\x04\x00\x00') + untagged = Ethernet(wire.ethernet(packet, 0x0800), 1).payload + want = _ipv4_2dict(untagged), _maybe_unhex(bytes(untagged.payload)) + for label, tpids in (('802.1Q', (C_TAG,)), ('802.1ad', (S_TAG,)), ('QinQ', (S_TAG, C_TAG)), + ('two customer tags', (C_TAG, C_TAG))): + with self.subTest(case=label): + ipv4 = _network(make_packet(Ethernet(tagged(packet, 0x0800, *tpids), 1))) + self.assertEqual(type(ipv4).__name__, 'IP') + self.assertEqual((_ipv4_2dict(ipv4), _maybe_unhex(bytes(ipv4.payload))), want) + + for label, frame, undecoded in ( + ('802.1Q, then ipv6', tagged(wire.ipv6(TCP_SEGMENT, nxt=6), 0x86DD, C_TAG), False), + ('802.1Q, then not ipv4', tagged(b'\x65' + packet[1:], 0x0800, C_TAG), True), + ('802.1Q, then a truncated ipv4 header', tagged(packet[:19], 0x0800, C_TAG), True), + ('802.1Q cut short', wire.ethernet(b'\x00\x64\x08', C_TAG), False), + ('0x9100 tag', tagged(packet, 0x0800, 0x9100), False), + ): + with self.subTest(case=label), warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + self.assertIsNone(_network(make_packet(Ethernet(frame, 1)), 7)) + self.assertEqual([str(item.message).split(' decoding failed')[0] for item in caught], + ['Frame 7: '] if undecoded else []) + + def test_engine_warns_alike_of_tagged_and_untagged_ipv4_it_cannot_decode(self) -> None: + from pcapkit import extract + + bad = b'\x44' + wire.ipv4(TCP_SEGMENT, proto=6)[1:] # IHL 4 + frames = (wire.ethernet(bad, 0x0800), tagged(bad, 0x0800, S_TAG, C_TAG), + wire.ethernet(wire.ipv4(TCP_SEGMENT, proto=6), 0x0800)) + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, 'undecodable.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, + ipv4=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 'decoding failed' in str(item.message)] + # one each, the untagged from the engine and the tagged from the toolkit, + # in the same words after the frame number + self.assertEqual([message.split(': ', 1)[0] for message in messages], + ['Frame 1', 'Frame 2']) + self.assertEqual(messages[0].split(': ', 1)[1], messages[1].split(': ', 1)[1]) + self.assertEqual([tuple(flow.index) for flow in extractor.trace.tcp], [(3,)]) + finally: + close_extractor(extractor) + + #: What :meth:`read_both` compares between the two engines. + ASPECTS = ('ipv4', 'tcp', 'trace', 'ipv4 datagrams', 'tcp datagrams', 'flows') + + def read(self, engine: str, path: str, tracedir: str) -> dict: + """Every field ``engine`` hands IPv4 and TCP reassembly and TCP flow tracing for ``path``. + + Inputs are keyed by frame number, and a trace input leaves out ``frame``, + the engine's own frame object -- as in ``test_engine_agreement_runtime``. + + """ + from unittest import mock + + from pcapkit import extract + from pcapkit.foundation.reassembly.reassembly import ReassemblyBase + from pcapkit.foundation.traceflow.traceflow import TraceFlowBase + + seen = {'ipv4': {}, 'tcp': {}, 'trace': {}} # type: dict + reassemble, trace = ReassemblyBase.__call__, TraceFlowBase.__call__ + + def fields(packet, skip=()): + return {key: bytes(value) if isinstance(value, bytearray) else value + for key, value in packet.items() if key not in skip} + + def spy_reassembly(this, packet): + seen[type(this).__name__.lower()][packet.num] = fields(packet) + return reassemble(this, packet) + + def spy_trace(this, packet): + seen['trace'][packet.index] = fields(packet, ('frame',)) + return trace(this, packet) + + with warnings.catch_warnings(record=True) as caught, \ + mock.patch.object(ReassemblyBase, '__call__', spy_reassembly), \ + mock.patch.object(TraceFlowBase, '__call__', spy_trace): + warnings.simplefilter('always') + extractor = extract(fin=path, nofile=True, engine=engine, reassembly=True, + ipv4=True, ipv6=False, tcp=True, trace=True, + trace_fout=tracedir, trace_format='json') + try: + self.assertEqual(extractor._exnam, engine) + self.assertEqual([str(item.message) for item in caught + if 'left out' in str(item.message)], []) + for kind in ('ipv4', 'tcp'): + seen[f'{kind} datagrams'] = [fields(datagram, ('packet',)) + for datagram in getattr(extractor.reassembly, kind)] + seen['flows'] = [(flow.label, tuple(flow.index)) for flow in extractor.trace.tcp] + finally: + close_extractor(extractor) + return seen + + def read_both(self, frames) -> tuple[dict, dict]: + """:meth:`read` a capture of ``frames`` with ``pypcapfile``, then with ``default``.""" + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, 'capture.pcap') + with open(path, 'wb') as file: + file.write(wire.pcap([(1, number, frame) for number, frame in enumerate(frames)])) + return (self.read('pypcapfile', path, os.path.join(tmp, 'pypcapfile')), + self.read('default', path, os.path.join(tmp, 'default'))) + + def test_vlan_tagged_ipv4_agrees_with_the_default_engine(self) -> None: + datagram = wire.udp(b'u' * 24) + frames = ( + tagged(wire.ipv4(wire.tcp(b'hello', seq=100, syn=True), proto=6), 0x0800, C_TAG), + tagged(wire.ipv4(wire.tcp(b'ok', seq=500, ack=106, sport=9, dport=40000), + proto=6, reverse=True), 0x0800, C_TAG), + tagged(wire.ipv4(wire.tcp(b'world', seq=106, ack=501), proto=6), 0x0800, S_TAG, C_TAG), + tagged(wire.ipv4(wire.tcp(b'!', seq=111, ack=501), proto=6), 0x0800, C_TAG, C_TAG), + tagged(wire.ipv4(wire.tcp(b'bye', seq=112, ack=501, fin=True), proto=6), 0x0800, S_TAG), + wire.ethernet(wire.ipv4(wire.tcp(b'late', seq=115, ack=501), proto=6), 0x0800), + # an IPv4 datagram in two tagged fragments, for IPv4 reassembly + tagged(wire.ipv4(datagram[:16], proto=17, mf=True, ident=0x99), 0x0800, C_TAG), + tagged(wire.ipv4(datagram[16:], proto=17, offset=16, ident=0x99), 0x0800, S_TAG, C_TAG), + # IPv4 and TCP options, on a flow of their own + tagged(make_ipv4(make_tcp(b'opts', options=b'\x02\x04\x05\xb4'), + options=b'\x94\x04\x00\x00'), 0x0800, C_TAG), + ) + mine, theirs = self.read_both(frames) + + self.assertEqual(sorted(mine['tcp']), [1, 2, 3, 4, 5, 6, 9]) + self.assertEqual(sorted(mine['ipv4']), [1, 2, 3, 4, 5, 6, 7, 8, 9]) + self.assertIn(datagram, [bytes(item['payload']) for item in mine['ipv4 datagrams']]) + self.assertEqual(len(mine['flows']), 2) + for aspect in self.ASPECTS: + with self.subTest(aspect=aspect): + self.assertEqual(mine[aspect], theirs[aspect]) + + def test_later_fragment_is_not_read_as_tcp(self) -> None: + # C.f. #1576: a SYN, then later fragments whose data merely looks like a + # TCP header -- untagged and tagged -- and a first fragment, which does + # carry one. + data = b'A' * 20 + b'B' * 20 + frames = ( + wire.ethernet(wire.ipv4(wire.tcp(b'', seq=0, syn=True), proto=6, df=True), 0x0800), + wire.ethernet(wire.ipv4(data, proto=6, offset=1480, ident=9), 0x0800), + tagged(wire.ipv4(data, proto=6, offset=8, mf=True, ident=10), 0x0800, C_TAG), + wire.ethernet(wire.ipv4(wire.tcp(b'first', seq=1, ack=1), proto=6, mf=True, ident=11), + 0x0800), + ) + mine, theirs = self.read_both(frames) + + self.assertEqual(sorted(mine['tcp']), [1, 4]) + for aspect in self.ASPECTS: + with self.subTest(aspect=aspect): + self.assertEqual(mine[aspect], theirs[aspect]) + + def test_tcp_payload_ends_at_the_total_length(self) -> None: + # C.f. #1577: Ethernet padding past the Total Length is not TCP payload. + frames = ( + wire.ethernet(wire.ipv4(wire.tcp(b'x', seq=100, ack=1), proto=6), 0x0800) + bytes(5), + tagged(wire.ipv4(wire.tcp(b'y', seq=101, ack=1), proto=6), 0x0800, C_TAG) + bytes(1), + # 0, as TCP segmentation offload leaves it: the rest of the frame + wire.ethernet(total_length(wire.ipv4(wire.tcp(b'tso', seq=102, ack=1), proto=6), 0), + 0x0800), + # shorter than the header: no payload at all + wire.ethernet(total_length(wire.ipv4(wire.tcp(b'bogus', seq=105, ack=1), proto=6), 12), + 0x0800), + # past the end of the capture: what was captured + wire.ethernet(total_length(wire.ipv4(wire.tcp(b'cut', seq=105, ack=1), proto=6), 200), + 0x0800), + ) + mine, theirs = self.read_both(frames) + + self.assertEqual(sorted(mine['tcp']), [1, 2, 3, 5]) + self.assertEqual([mine['tcp'][number]['payload'] for number in (1, 2, 3, 5)], + [b'x', b'y', b'tso', b'cut']) + for aspect in self.ASPECTS: + with self.subTest(aspect=aspect): + self.assertEqual(mine[aspect], theirs[aspect]) + + def test_engine_does_not_warn_of_tunnelled_tcp_headers_the_default_engine_rejects(self) -> None: + from pcapkit import extract + + tcp4 = wire.ipv4(TCP_SEGMENT, proto=6) + syn = make_tcp(b'', flags=0x02, options=b'\x02\x04\x05\xb4') + frames = ( + # a tunnelled SYN whose options the snapshot length cut off + wire.ethernet(wire.ipv6(wire.ipv6(syn, nxt=6), nxt=41), 0x86DD)[:-2], + # a tunnelled TCP header with a Data Offset under 5 words + wire.ethernet(wire.ipv4(wire.ipv4(data_offset(TCP_SEGMENT, 4), proto=6), proto=4), 0x0800), + wire.ethernet(wire.ipv4(tcp4, proto=4), 0x0800), + ) + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, 'tunnels.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') + self.assertEqual([str(item.message).split(': ', 1)[0] for item in caught + if 'left out' in str(item.message)], ['Frame 3']) + finally: + close_extractor(extractor) + + def test_engine_warns_once_per_capture_on_tunnelled_tcp(self) -> None: + from pcapkit import extract + + tcp4, tcp6 = wire.ipv4(TCP_SEGMENT, proto=6), wire.ipv6(TCP_SEGMENT, nxt=6) + frames = ( + wire.ethernet(tcp4, 0x0800), + wire.ethernet(wire.ipv4(tcp4, proto=4), 0x0800), + wire.ethernet(wire.ipv4(tcp6, proto=41), 0x0800), + wire.ethernet(wire.ipv6(tcp6, nxt=41), 0x86DD), + tagged(tcp6, 0x86DD, C_TAG), + tagged(wire.ipv4(tcp4, proto=4), 0x0800, C_TAG), + ) + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, 'tunnels.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: + messages = [str(item.message) for item in caught + if 'left out' in str(item.message)] + self.assertEqual(extractor._exnam, 'pypcapfile') + self.assertEqual(len(messages), 2, messages) + self.assertTrue(messages[0].startswith('Frame 2: TCP tunnelled in IP'), messages) + self.assertTrue(messages[1].startswith('Frame 5: TCP over IPv6'), messages) + self.assertEqual([tuple(flow.index) for flow in extractor.trace.tcp], [(1,)]) + finally: + close_extractor(extractor) + if __name__ == '__main__': unittest.main()