diff --git a/docs/source/pcapkit/foundation/engines/index.rst b/docs/source/pcapkit/foundation/engines/index.rst index efbf1110dd..c4a7bdcf89 100644 --- a/docs/source/pcapkit/foundation/engines/index.rst +++ b/docs/source/pcapkit/foundation/engines/index.rst @@ -92,8 +92,8 @@ 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, and TCP tunnelled in IP (e.g. 6in4, 4in4), are | -| | left out of TCP reassembly and flow tracing, each with an | +| | TCP over IPv6, tunnelled in IP (e.g. 6in4) or not, is left | +| | out of TCP reassembly and flow tracing, 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 | diff --git a/docs/source/pcapkit/toolkit/3rdparty.rst b/docs/source/pcapkit/toolkit/3rdparty.rst index bbc3bece1f..eea167ab8b 100644 --- a/docs/source/pcapkit/toolkit/3rdparty.rst +++ b/docs/source/pcapkit/toolkit/3rdparty.rst @@ -185,11 +185,13 @@ cannot be used for it. 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 + behind 802.1Q and 802.1ad tags as `PyPCAPFile`_ reads an untagged one, and + TCP tunnelled in IP (e.g. 4in4, 4in6) from its innermost IPv4 packet, keyed by + that packet's addresses as the default engine keys it, while TCP over IPv6, + tunnelled (e.g. 6in4, 6in6) or not, is 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. + :class:`~pcapkit.utilities.warnings.AttributeWarning` once per capture. .. autofunction:: pcapkit.toolkit.pypcapfile.ipv4_reassembly diff --git a/pcapkit/toolkit/dpkt.py b/pcapkit/toolkit/dpkt.py index b4d2d9a29d..9a771af80f 100644 --- a/pcapkit/toolkit/dpkt.py +++ b/pcapkit/toolkit/dpkt.py @@ -72,6 +72,17 @@ #: .. _DPKT: https://dpkt.readthedocs.io BUFFER_ATTR = '__pcapkit_buffer__' +#: IP protocol numbers of the IP-in-IP tunnels the default engine dissects +#: (see :attr:`Internet.__proto__ +#: `), over IPv4 and +#: IPv6 alike: IPv4 (protocol 4, :rfc:`2003`) and IPv6 (protocol 41, +#: :rfc:`2473` and :rfc:`4213`). `DPKT`_ also decodes an IPv4 packet's +#: protocol 0 as IPv4, its ``IP_PROTO_IP``, where the default engine reads an +#: IPv6 Hop-by-Hop Options header, so :func:`_tcp_carrier` does not enter it. +#: +#: .. _DPKT: https://dpkt.readthedocs.io +IP_TUNNELS = frozenset((int(Enum_TransType.IPv4), int(Enum_TransType.IPv6))) + class RecordTimestamp: """A record's capture timestamp, with what a PCAP frame record needs of it. @@ -420,6 +431,44 @@ def _network(packet: 'Packet') -> 'Optional[IP | IP6]': return getattr(packet, 'ip6', None) +def _tcp_carrier(packet: 'Packet') -> 'tuple[Optional[IP | IP6], Optional[TCP]]': + """Fetch the IP layer of a DPKT packet that carries its TCP segment, and the segment. + + Args: + packet: DPKT packet, i.e. the outermost layer. + + Returns: + The innermost IPv4 or IPv6 packet of the network layer (see + :func:`_network`) and of the IP-in-IP tunnels it carries, and its + :class:`dpkt.tcp.TCP` payload, if it has one; else :data:`None` for the + latter, and for both with no network layer. A tunnel is a packet whose + protocol, or for IPv6 the next header after its extension headers, is + one of the :data:`IP_TUNNELS`. + + Note: + `DPKT`_ decodes an IPv4 or IPv6 packet inside another (:rfc:`2003`, + :rfc:`2473`, :rfc:`4213`) as its ``data``. The TCP endpoints are the + innermost packet's addresses, as for the default engine + (:func:`pcapkit.toolkit.pcap.tcp_segment`); the outer ones are the + tunnel's (:issue:`1581`). + + .. _DPKT: https://dpkt.readthedocs.io + + """ + import dpkt # isort:skip + + ip = _network(packet) + if ip is None: + return None, None + while getattr(ip, 'p', None) in IP_TUNNELS and isinstance(ip.data, (dpkt.ip.IP, dpkt.ip6.IP6)): + ip = ip.data + + tcp = getattr(ip, 'tcp', None) # type: Optional[TCP] + if tcp is None and type(getattr(ip, 'data', None)).__name__ == 'TCP': + tcp = cast('TCP', ip.data) + return ip, tcp + + def ipv6_hdr_len(ipv6: 'IP6') -> 'int': """Calculate length of headers before IPv6 Fragment header. @@ -670,7 +719,8 @@ def tcp_reassembly(packet: 'Packet', timestamp: 'float', *, Data for TCP reassembly. * If the ``packet`` can be used for TCP reassembly. A packet can be reassembled - if it contains TCP layer (:class:`dpkt.tcp.TCP`). + if it contains TCP layer (:class:`dpkt.tcp.TCP`), keyed by the IP packet that + carries it -- in an IP-in-IP tunnel, the innermost (see :func:`_tcp_carrier`). * 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`. @@ -687,14 +737,8 @@ def tcp_reassembly(packet: 'Packet', timestamp: 'float', *, :class:`pcapkit.foundation.reassembly.tcp.TCP` """ - ip = _network(packet) - if ip is None: - return None - - tcp = getattr(ip, 'tcp', None) # type: Optional[TCP] - if tcp is None and type(getattr(ip, 'data', None)).__name__ == 'TCP': - tcp = cast('TCP', ip.data) - if tcp is not None: + ip, tcp = _tcp_carrier(packet) + if ip is not None and tcp is not None: flags = bin(tcp.flags)[2:].zfill(8) wire, _ = _layer2bytes(packet, tcp) raw_len = len(tcp.data) # payload length, header excludes @@ -740,7 +784,8 @@ def tcp_traceflow(packet: 'Packet', timestamp: 'float', *, Data for TCP reassembly. * If the ``packet`` can be used for TCP flow tracing. A packet can be reassembled - if it contains TCP layer (:class:`dpkt.tcp.TCP`). + if it contains TCP layer (:class:`dpkt.tcp.TCP`), keyed by the IP packet that + carries it -- in an IP-in-IP tunnel, the innermost (see :func:`_tcp_carrier`). * If the ``packet`` can be reassembled, then the :obj:`dict` mapping of data for TCP flow tracing (:term:`trace.tcp.packet`) will be returned; otherwise, returns :data:`None`. @@ -757,14 +802,8 @@ def tcp_traceflow(packet: 'Packet', timestamp: 'float', *, :class:`pcapkit.foundation.traceflow.tcp.TCP` """ - ip = _network(packet) - if ip is None: - return None - - tcp = getattr(ip, 'tcp', None) # type: Optional[TCP] - if tcp is None and type(getattr(ip, 'data', None)).__name__ == 'TCP': - tcp = cast('TCP', ip.data) - if tcp is not None: + ip, tcp = _tcp_carrier(packet) + if ip is not None and tcp is not None: flags = bin(tcp.flags)[2:].zfill(8) wire, _ = _layer2bytes(packet, tcp) diff --git a/pcapkit/toolkit/pcap.py b/pcapkit/toolkit/pcap.py index 427276cbd8..377fbf03be 100644 --- a/pcapkit/toolkit/pcap.py +++ b/pcapkit/toolkit/pcap.py @@ -19,7 +19,11 @@ 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.ip import IP as Protocol_IP +from pcapkit.protocols.misc.null import NoPayload from pcapkit.protocols.misc.raw import Raw +from pcapkit.protocols.protocol import ProtocolBase +from pcapkit.protocols.transport.tcp import TCP if TYPE_CHECKING: from ipaddress import IPv4Address, IPv6Address @@ -32,8 +36,6 @@ from pcapkit.protocols.internet.ipv6 import IPv6 from pcapkit.protocols.internet.ipv6_frag import IPv6_Frag from pcapkit.protocols.misc.pcap import Frame - from pcapkit.protocols.protocol import ProtocolBase - from pcapkit.protocols.transport.tcp import TCP __all__ = ['ipv4_reassembly', 'ipv6_reassembly', 'tcp_reassembly', 'tcp_traceflow', 'tcp_segment', 'TCPSegment'] @@ -46,7 +48,7 @@ class TCPSegment(NamedTuple): """ - #: Info of the IP layer carrying the segment. + #: Info of the IP layer carrying the segment, the innermost one in a tunnel. ip: 'Data_IPv4 | Data_IPv6' #: TCP source port. srcport: 'int' @@ -68,6 +70,38 @@ class TCPSegment(NamedTuple): payload: 'bytes' +def _carrier(frame: 'ProtocolBase') -> 'tuple[Optional[IPv4 | IPv6], Optional[TCP]]': + """Fetch the TCP layer of a frame and the IP layer that carries it. + + Args: + frame: PCAP frame or PCAP-NG block. + + Returns: + The IPv4 or IPv6 layer nearest above the frame's TCP layer, and that TCP + layer; with no TCP layer, the last IPv4 or IPv6 layer of the frame and + :data:`None`. With no IP layer, neither. + + Note: + The nearest IP layer is the innermost one of an IP-in-IP tunnel + (:rfc:`2003`, :rfc:`2473`, :rfc:`4213`), whose addresses are the TCP + endpoints', as in Wireshark's TCP conversations; the outermost, which + ``frame['IP']`` finds, is the tunnel's (:issue:`1581`). It is found by + walking down from the outermost, so the headers between them do not + intervene: IPv6 dissects its own extension headers, and an AH or other + header inside IPv4 is a layer of its own but not an IP layer. + + """ + if 'IP' not in frame: + return None, None + ip = cast('IPv4 | IPv6', frame['IP']) + layer = ip.payload + while isinstance(layer, ProtocolBase) and not isinstance(layer, (NoPayload, TCP)): + if isinstance(layer, Protocol_IP): + ip = cast('IPv4 | IPv6', layer) + layer = layer.payload + return ip, (cast('TCP', frame['TCP']) if 'TCP' in frame else None) + + def tcp_segment(frame: 'ProtocolBase') -> 'Optional[TCPSegment]': """The TCP segment ``frame`` carries, or :data:`None` if it carries none. @@ -78,11 +112,13 @@ def tcp_segment(frame: 'ProtocolBase') -> 'Optional[TCPSegment]': frame: PCAP frame or PCAP-NG block. Returns: - The fields of the TCP layer if :mod:`pcapkit` parsed one. Otherwise, the - fields of the fixed 20-octet TCP header (:rfc:`9293#section-3.1`) if the - IP payload is a :class:`~pcapkit.protocols.misc.raw.Raw` the TCP parser - rejected, of protocol TCP, from a datagram that is not a later fragment, - with at least those 20 octets captured; else :data:`None`. + The fields of the TCP layer if :mod:`pcapkit` parsed one, with those of + the IP layer that carries it -- in a tunnel, the innermost + (:issue:`1581`). Otherwise, the fields of the fixed 20-octet TCP header + (:rfc:`9293#section-3.1`) if the payload of the innermost IP layer is a + :class:`~pcapkit.protocols.misc.raw.Raw` the TCP parser rejected, of + protocol TCP, from a datagram that is not a later fragment, with at + least those 20 octets captured; else :data:`None`. Note: A capture snapped inside the TCP header leaves a Data Offset that runs @@ -92,14 +128,17 @@ def tcp_segment(frame: 'ProtocolBase') -> 'Optional[TCPSegment]': are all in the fixed header, so the segment still belongs to its flow, as it does for :mod:`dpkt`, :mod:`scapy` and Wireshark (:issue:`1518`). The header and payload are then the captured octets either side of the - Data Offset. + Data Offset. Inside a tunnel, this holds of the IP layer carrying the + segment, as it does without one. """ - if 'TCP' in frame: - tcp = cast('TCP', frame['TCP']) + ip, tcp = _carrier(frame) + if ip is None: + return None + if tcp is not None: tcp_info = tcp.info return TCPSegment( - ip=cast('IPv4 | IPv6', frame['IP']).info, + ip=ip.info, srcport=tcp_info.srcport.port, dstport=tcp_info.dstport.port, seq=tcp_info.seq, @@ -111,9 +150,6 @@ def tcp_segment(frame: 'ProtocolBase') -> 'Optional[TCPSegment]': payload=tcp.packet.payload, ) - if 'IP' not in frame: - return None - ip = cast('IPv4 | IPv6', frame['IP']) ip_info = ip.info raw = ip.payload # NOTE: a Raw with no error is one the caller asked for, by stopping the diff --git a/pcapkit/toolkit/pypcapfile.py b/pcapkit/toolkit/pypcapfile.py index e2f090c572..c73034f60d 100644 --- a/pcapkit/toolkit/pypcapfile.py +++ b/pcapkit/toolkit/pypcapfile.py @@ -20,8 +20,9 @@ from "this frame carries no IPv6 fragment". For the same reason :func:`tcp_reassembly` and :func:`tcp_traceflow` see TCP - over IPv4 only, and decline a frame carrying TCP over IPv6. That loss is not - silent: it is reported with one + over IPv4 only, and decline a frame carrying TCP over IPv6 -- TCP whose + innermost IP layer is IPv6, tunnelled in IP (e.g. 6in4, 6in6) or not. That + loss is not silent: it is reported with one :exc:`~pcapkit.utilities.warnings.AttributeWarning` per capture, on its first such frame, rather than one per frame. @@ -30,16 +31,16 @@ :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`). 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`). + Nor does it decode a tunnel, so :func:`_innermost` steps through the IPv4 + and IPv6 packets carried in IP (e.g. 4in4, 4in6) to the innermost, and + decodes it with the same decoder if it is IPv4: the TCP it carries is read + and keyed by its addresses, as the default engine keys it, not by the + tunnel's (:issue:`1581`). The warning follows the default engine's own rules + for where it finds TCP, read off the raw headers (see :func:`_upper_layer`). + IPv4 carried in IPv6 (4in6) also 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 @@ -165,12 +166,12 @@ )) #: The warning :func:`_decline_tcp` gives for each kind of TCP that -#: :func:`_left_out` finds this engine cannot read. +#: :func:`_left_out` finds this engine cannot read. TCP tunnelled in IP is no +#: kind of its own: the innermost IP layer carrying it is read if it is IPv4, +#: and is TCP over IPv6 otherwise (see :func:`_innermost`, :issue:`1581`). 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 @@ -561,10 +562,10 @@ def _ipv4_in_ipv6(packet: 'Packet') -> 'Optional[IP]': 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. + gives the first IPv4 layer of a frame however deep; TCP is keyed by the + innermost instead (see :func:`_innermost`). 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 @@ -579,12 +580,29 @@ def _ipv4_in_ipv6(packet: 'Packet') -> 'Optional[IP]': upper = _upper_layer(6, upper[1]) if upper is None or upper[0] != Enum_TransType.IPv4: return None + return _decode_ipv4(upper[1]) + +def _decode_ipv4(data: 'bytes') -> 'Optional[IP]': + """Decode an IPv4 packet that `PyPCAPFile`_ left undecoded inside another IP packet. + + Args: + data: Raw octets from the IPv4 header on, as far as the IP packet + carrying it goes (see :func:`_upper_layer`). + + Returns: + The packet, decoded as :func:`_network` decodes one behind VLAN tags, or + :data:`None` if it will not decode or the default engine rejects its + header (see :func:`_ipv4_accepted`), which then dissects nothing past it. + + .. _PyPCAPFile: https://github.com/kisom/pypcapfile + + """ # 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) + ipv4 = IP(data, 0) except (struct.error, AssertionError, ValueError, IndexError, KeyError): return None return ipv4 if _ipv4_accepted(ipv4) else None @@ -690,17 +708,16 @@ def _upper_layer(version: 'int', data: 'bytes') -> 'Optional[tuple[int, bytes]]' Note: 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 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. + :func:`_left_out` applies to the TCP header, so that :func:`_innermost` + stops at the IP layer the default engine finds TCP in, and the warning + :func:`_decline_tcp` gives fires on the frames it finds TCP over IPv6 in. + This reads only as far as the upper-layer protocol number and decodes + nothing, so it does not see an IPv6 extension header option the default + engine rejects, which leaves the header raw. Every IPv4 header on the + way is checked by :func:`_innermost` (see :func:`_ipv4_accepted`), and a + TCP header the default engine's TCP parser rejects directly over its IP + layer, tunnelled or not, it still reads out of the raw octets left + behind (:func:`pcapkit.toolkit.pcap.tcp_segment`). """ if version == 4: @@ -738,57 +755,89 @@ def _upper_layer(version: 'int', data: 'bytes') -> 'Optional[tuple[int, bytes]]' 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. +def _innermost(packet: 'Packet', count: 'int' = -1) -> 'tuple[Optional[IP], Optional[tuple[int, bytes]]]': + """Fetch the innermost IP layer of a PyPCAPFile frame, the one carrying its TCP. Args: - packet: PyPCAPFile packet, which :func:`_transport` found no TCP segment in. - ipv4: Its IPv4 layer, as :func:`_network` fetches it. + packet: PyPCAPFile packet. + count: Packet index, for the warning :func:`_network` gives. 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 innermost IPv4 packet and :data:`None`, if the innermost IP layer is + IPv4; :data:`None` and the upper-layer protocol and octets of the + innermost IPv6 packet, as :func:`_upper_layer` gives them, if it is + IPv6; otherwise neither. + + The outermost IP layer is the IPv4 packet :func:`_network` fetches, or + an IPv6 packet, behind VLAN tags (see :func:`_untag`) or not. Each IPv4 + or IPv6 packet one of them carries in the :data:`IP_TUNNELS`, directly + or behind the headers :func:`_upper_layer` steps over, is the next, as + deep as they go. An IPv4 packet in a tunnel is decoded by + :func:`_decode_ipv4`, and one it does not decode ends the walk with + neither, as the default engine then dissects no TCP past it. - 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`). + Note: + The default engine keys a TCP segment by the IP layer nearest above it + (:func:`pcapkit.toolkit.pcap.tcp_segment`), which in a tunnel is the + innermost, so the TCP endpoints are its addresses and not the tunnel's + (:issue:`1581`). """ - if ipv4 is not None: # read by PyPCAPFile, so only a tunnel can hide TCP + ipv4 = _network(packet, count) + if ipv4 is not None: # a later fragment has no upper-layer header, see ``_upper_layer`` 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 + return ipv4, 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 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 + return None, 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' + version, data = IP_TUNNELS[upper[0]], upper[1] + upper = _upper_layer(version, data) + if version == 6: + ipv4 = None + elif (ipv4 := _decode_ipv4(data)) is None: + return None, None + if ipv4 is not None: + return ipv4, None + return None, upper + + +def _left_out(upper: 'Optional[tuple[int, bytes]]') -> 'Optional[str]': + """Name the TCP a PyPCAPFile frame carries that this engine cannot read. + + Args: + upper: The upper-layer protocol and octets of the frame's innermost IP + layer, if that is IPv6, as :func:`_innermost` gives them. + + Returns: + ``'ipv6'`` if they are a TCP header, i.e. the frame carries TCP over + IPv6, directly or behind extension headers, tunnelled in IP or not; + otherwise :data:`None`. 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. The rest of + the header need not be: the default engine's TCP parser rejects one cut + short, e.g. by the snapshot length, and then reads the segment out of + the raw octets left behind (:func:`pcapkit.toolkit.pcap.tcp_segment`, + :issue:`1518`). + + """ 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)): + if (segment[12] >> 4) * 4 < TCP_MIN_HEADER_LEN: # the Data Offset, in octets return None - return kind + return 'ipv6' def _capture_of(packet: 'Packet') -> 'tuple[int, Any]': @@ -819,12 +868,13 @@ def _capture_of(packet: 'Packet') -> 'tuple[int, Any]': return id(anchor), anchor -def _decline_tcp(packet: 'Packet', ipv4: 'Optional[IP]', count: 'int') -> 'None': +def _decline_tcp(packet: 'Packet', upper: 'Optional[tuple[int, bytes]]', count: 'int') -> 'None': """Warn that a frame carrying TCP this engine cannot read is left out. Args: packet: PyPCAPFile packet, which :func:`_transport` found no TCP segment in. - ipv4: Its IPv4 layer, as :func:`_network` fetches it. + upper: The upper-layer protocol and octets of its innermost IP layer, if + that is IPv6, as :func:`_innermost` gives them. count: Packet index. Warns: @@ -842,7 +892,7 @@ def _decline_tcp(packet: 'Packet', ipv4: 'Optional[IP]', count: 'int') -> 'None' key, anchor = _capture_of(packet) if all((key, kind) in _tcp_warned for kind in TCP_LEFT_OUT): return - kind = _left_out(packet, ipv4) + kind = _left_out(upper) if kind is None or (key, kind) in _tcp_warned: return @@ -1149,9 +1199,9 @@ def tcp_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'TCP_Packet | None * If the ``packet`` can be used for TCP reassembly. A packet can be reassembled 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. + :func:`_transport`). In an IP-in-IP tunnel the IPv4 layer is the innermost + IP layer, whose addresses key the segment (see :func:`_innermost`); TCP over + IPv6, tunnelled or not, cannot be, as `PyPCAPFile`_ has no IPv6 decoder. * If the ``packet`` can be reassembled, then the :obj:`dict` mapping of data for TCP reassembly (:term:`reasm.tcp.packet`) will be returned; otherwise, returns :data:`None`. @@ -1160,11 +1210,11 @@ def tcp_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'TCP_Packet | None IPv4 address. Warns: - 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 + AttributeWarning: If ``packet`` carries TCP over IPv6, tunnelled in IP or + not, 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. + warning. See Also: :class:`pcapkit.foundation.reassembly.tcp.TCP` @@ -1172,9 +1222,9 @@ def tcp_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'TCP_Packet | None .. _PyPCAPFile: https://github.com/kisom/pypcapfile """ - ipv4 = _network(packet, count) + ipv4, upper = _innermost(packet, count) if ipv4 is None or (segment := _transport(ipv4)) is None: - _decline_tcp(packet, ipv4, count) + _decline_tcp(packet, upper, count) return None # NOTE: imported only once there is something to decode, so that declining a @@ -1222,9 +1272,9 @@ def tcp_traceflow(packet: 'Packet', *, data_link: 'Enum_LinkType', * 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, 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. + :func:`_transport`). In an IP-in-IP tunnel the IPv4 layer is the innermost + IP layer, whose addresses key the segment (see :func:`_innermost`); TCP over + IPv6, tunnelled or not, cannot be, as `PyPCAPFile`_ has no IPv6 decoder. * If the ``packet`` can be traced, then the :obj:`dict` mapping of data for TCP flow tracing (:term:`trace.tcp.packet`) will be returned; otherwise, returns :data:`None`. @@ -1233,11 +1283,11 @@ def tcp_traceflow(packet: 'Packet', *, data_link: 'Enum_LinkType', IPv4 address. Warns: - 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 + AttributeWarning: If ``packet`` carries TCP over IPv6, tunnelled in IP or + not, 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. + warning. See Also: :class:`pcapkit.foundation.traceflow.tcp.TCP` @@ -1245,9 +1295,9 @@ def tcp_traceflow(packet: 'Packet', *, data_link: 'Enum_LinkType', .. _PyPCAPFile: https://github.com/kisom/pypcapfile """ - ipv4 = _network(packet, count) + ipv4, upper = _innermost(packet, count) if ipv4 is None or (segment := _transport(ipv4)) is None: - _decline_tcp(packet, ipv4, count) + _decline_tcp(packet, upper, count) return None # NOTE: imported only once there is something to decode, so that declining a diff --git a/pcapkit/toolkit/pyshark.py b/pcapkit/toolkit/pyshark.py index b4173afa81..4244a4dde5 100644 --- a/pcapkit/toolkit/pyshark.py +++ b/pcapkit/toolkit/pyshark.py @@ -18,7 +18,7 @@ """ import decimal import ipaddress -from typing import TYPE_CHECKING, cast +from typing import TYPE_CHECKING from pcapkit.const.reg.linktype import LinkType as Enum_LinkType from pcapkit.foundation.traceflow.data.tcp import Packet as TF_TCP_Packet @@ -455,6 +455,34 @@ def packet2dict(packet: 'Packet') -> 'dict[str, Any]': return dict_ +def _carrier(packet: 'Packet') -> 'tuple[Any, Any]': + """Fetch the TCP layer of a PyShark packet and the IP layer that carries it. + + Args: + packet: PyShark packet. + + Returns: + The ``ip`` or ``ipv6`` layer nearest above the packet's first ``tcp`` + layer, and that TCP layer; :data:`None` for either the packet lacks. + + Note: + The nearest is the innermost IP layer of an IP-in-IP tunnel + (:rfc:`2003`, :rfc:`2473`, :rfc:`4213`), whose addresses are the TCP + endpoints', as for the default engine + (:func:`pcapkit.toolkit.pcap.tcp_segment`). ``packet.ip`` is the first + ``ip`` layer instead, which is the tunnel's (:issue:`1581`). + + """ + ip = None # type: Any + for layer in packet.layers: + name = layer.layer_name.lower() + if name == 'tcp': + return ip, layer + if name in ('ip', 'ipv6'): + ip = layer + return ip, None + + def tcp_traceflow(packet: 'Packet', *, count: 'int' = -1) -> 'TF_TCP_Packet | None': """Trace packet flow for TCP. @@ -468,7 +496,8 @@ def tcp_traceflow(packet: 'Packet', *, count: 'int' = -1) -> 'TF_TCP_Packet | No Data for TCP flow tracing. * If the ``packet`` can be used for TCP flow tracing. A packet can be reassembled - if it contains TCP layer. + if it contains TCP layer, keyed by the IP layer that carries it -- in an + IP-in-IP tunnel, the innermost (see :func:`_carrier`). * If the ``packet`` can be reassembled, then the :obj:`dict` mapping of data for TCP flow tracing (:term:`trace.tcp.packet`) will be returned; otherwise, returns :data:`None`. @@ -482,15 +511,8 @@ def tcp_traceflow(packet: 'Packet', *, count: 'int' = -1) -> 'TF_TCP_Packet | No :class:`pcapkit.foundation.traceflow.tcp.TCP` """ - if 'IP' in packet: - ip = cast('Packet', packet.ip) - elif 'IPv6' in packet: - ip = cast('Packet', packet.ipv6) - else: - return None - - if 'TCP' in packet: - tcp = cast('Packet', packet.tcp) + ip, tcp = _carrier(packet) + if ip is not None and tcp is not None: # NOTE: the link type is keyed on ``frame.encap_type`` -- Wireshark's # internal ``WTAP_ENCAP_*`` number -- through diff --git a/pcapkit/toolkit/scapy.py b/pcapkit/toolkit/scapy.py index 746b4e65db..08e99953f2 100644 --- a/pcapkit/toolkit/scapy.py +++ b/pcapkit/toolkit/scapy.py @@ -301,6 +301,37 @@ def _tcp_segment(tcp: 'TCP') -> 'tuple[bytes, bytes] | None': return segment[:hdr_len], segment[hdr_len:] +def _carrier(layer: 'Packet') -> 'IP | IPv6 | None': + """Fetch the IP layer that carries ``layer``. + + Args: + layer: A layer of a dissected Scapy packet. + + Returns: + The nearest :class:`~scapy.layers.inet.IP` or + :class:`~scapy.layers.inet6.IPv6` layer above ``layer``, or + :data:`None` if there is none. + + Note: + The nearest is the innermost IP layer of an IP-in-IP tunnel + (:rfc:`2003`, :rfc:`2473`, :rfc:`4213`), whose addresses are the TCP + endpoints', as for the default engine + (:func:`pcapkit.toolkit.pcap.tcp_segment`). ``packet['IP']`` finds the + outermost IPv4 layer instead, which is the tunnel's, or with IPv4 + carried in IPv6 the inner one (:issue:`1581`). The extension header + layers `Scapy`_ dissects between an IP layer and its payload are not + IP layers, so are stepped over. + + """ + from scapy.layers.inet import IP + from scapy.layers.inet6 import IPv6 + + carrier = layer.underlayer + while carrier is not None and not isinstance(carrier, (IP, IPv6)): + carrier = carrier.underlayer + return carrier + + def ipv4_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'IP_Packet[IPv4Address] | None': """Make data for IPv4 reassembly. @@ -441,6 +472,8 @@ def tcp_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'TCP_Packet | None * If the ``packet`` can be used for TCP reassembly. A packet can be reassembled if it contains TCP layer (:class:`scapy.layers.inet.TCP`) whose fixed header lies inside its datagram, and whose Data Offset counts at least that header. + It is keyed by the IP layer that carries it -- in an IP-in-IP tunnel, the + innermost (see :func:`_carrier`). * 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`. @@ -448,16 +481,9 @@ def tcp_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'TCP_Packet | None :class:`pcapkit.foundation.reassembly.tcp.TCP` """ - if 'IP' in packet: - ip = cast('IP', packet['IP']) - elif 'IPv6' in packet: - ip = cast('IPv6', packet['IPv6']) - else: - return None - if 'TCP' in packet: tcp = cast('TCP', packet['TCP']) - if (segment := _tcp_segment(tcp)) is None: + if (ip := _carrier(tcp)) is None or (segment := _tcp_segment(tcp)) is None: return None header, payload = segment @@ -507,9 +533,8 @@ def tcp_traceflow(packet: 'Packet', *, count: 'int' = -1) -> 'TF_TCP_Packet | No """ if 'TCP' in packet: - ip = cast('IP', packet['IP']) if 'IP' in packet else cast('IPv6', packet['IPv6']) tcp = cast('TCP', packet['TCP']) - if (segment := _tcp_segment(tcp)) is None: + if (ip := _carrier(tcp)) is None or (segment := _tcp_segment(tcp)) is None: return None header, payload = segment diff --git a/tests/toolkit/test_pypcapfile_header_checks_unit.py b/tests/toolkit/test_pypcapfile_header_checks_unit.py index c9cfc4b4c4..839e38c1bb 100644 --- a/tests/toolkit/test_pypcapfile_header_checks_unit.py +++ b/tests/toolkit/test_pypcapfile_header_checks_unit.py @@ -240,18 +240,24 @@ def test_extension_headers_are_parsed_once_per_frame(self) -> None: 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: + def test_tunnelled_tcp_behind_ah_is_found(self) -> None: + # C.f. #1581: a tunnel behind AH or an IPv6 extension header in IPv4 is + # walked into, and the TCP read from the innermost IPv4 layer. + from pcapkit.toolkit.pypcapfile import _innermost, _transport + 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)), + inner_ah = wire.ipv4(base.ah(6) + base.TCP_SEGMENT, proto=51) + for label, frame, inner in ( + ('4in4 behind ah', base.tunnel(base.ah(4) + tcp4, 51), tcp4), + ('4in4, inner ah', base.tunnel(inner_ah, 4), inner_ah), + ('4in4 behind destination options', base.tunnel(base.options_header(4) + tcp4, 60), tcp4), ): - 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) + with self.subTest(case=label), mock.patch.dict('sys.modules', base.fake_ip_decoder()) as modules: + IP = modules['pcapfile.protocols.network.ip'].IP + ipv4, _ = _innermost(base.make_packet(frame)) + self.assertIsInstance(ipv4, IP) + self.assertEqual(IP.calls, [(inner, 0)]) + self.assertEqual(_transport(ipv4), base.TCP_SEGMENT) if __name__ == '__main__': diff --git a/tests/toolkit/test_pypcapfile_unit.py b/tests/toolkit/test_pypcapfile_unit.py index a485b59071..fc9949f42c 100644 --- a/tests/toolkit/test_pypcapfile_unit.py +++ b/tests/toolkit/test_pypcapfile_unit.py @@ -26,6 +26,7 @@ from tests._support import close_extractor, reimport_once_per_class from tests.foundation import _roundtrip as wire +from tests.toolkit import test_tunnelled_tcp_flow_key_1581_unit as tunnels RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) @@ -185,8 +186,9 @@ def tagged_frame(packet: bytes, ethertype: int, *tpids: int) -> FakeEthernet: 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. + Their ``IP`` records what it was given, refuses what is not an IPv4 header + of at least 20 octets as the real decoder does, and decodes the fields as + it does, with the options and the payload hex-encoded. """ class IP: @@ -198,10 +200,15 @@ 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]) - # 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] + fields = struct.unpack('!BBHHHBBHII', packet[:20]) + self.v, self.hl = fields[0] >> 4, fields[0] & 0x0F + self.tos, self.len, self.id = fields[1:4] + self.flags, self.off = fields[4] >> 13, fields[4] & 0x1FFF + self.ttl, self.p, self.sum = fields[5:8] + self.src, self.dst = (str(ipaddress.IPv4Address(field)).encode() for field in fields[8:]) + # the options and payload, sliced without a bounds check, as the real one does + self.opt = binascii.hexlify(packet[20:self.hl * 4]) + self.payload = binascii.hexlify(packet[self.hl * 4:]) modules = {name: types.ModuleType(name) for name in ( 'pcapfile', 'pcapfile.protocols', 'pcapfile.protocols.network', @@ -716,40 +723,82 @@ def test_vlan_tagged_tcp_over_ipv6_warns(self) -> None: self.assertTrue(messages[0].startswith('Frame 1: TCP over IPv6 is left out'), messages) - def test_tunnelled_tcp_warns(self) -> None: + def test_tunnelled_tcp_over_ipv6_warns(self) -> None: + # C.f. #1581: TCP whose innermost IP layer is IPv6 is TCP over IPv6, + # however deep the tunnel, and warns as such. + from unittest import mock + 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))), + ('6in4in4', tunnel(wire.ipv4(tcp6, proto=41), 4)), + ('6in6 in a 4in4', tunnel(wire.ipv4(wire.ipv6(tcp6, nxt=41), proto=41), 4)), + ('outer first fragment', tunnel(tcp6, 41, flags=0b001)), ('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)), + ('6in4 behind ah', tunnel(ah(41) + tcp6, 51)), ('inner ipv6 jumbogram', tunnel(JUMBOGRAM, 41)), - ('outer ipv4 option', tunnel(tcp4, 4, hl=6, opt=b'\x94\x04\x00\x00')), # router alert + ('outer ipv4 option', tunnel(tcp6, 41, hl=6, opt=b'\x94\x04\x00\x00')), # router alert ): - with self.subTest(case=label): + with self.subTest(case=label), mock.patch.dict('sys.modules', fake_ip_decoder()): 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]) + self.assertTrue(messages[0].startswith('Frame 1: TCP over IPv6 is left out'), messages) + + def test_tunnelled_tcp_over_ipv4_is_found_in_the_innermost_ipv4(self) -> None: + # C.f. #1581: TCP whose innermost IP layer is IPv4 is read from that + # layer, which keys it, rather than left out with a warning. + from unittest import mock + + from pcapkit.toolkit.pypcapfile import _innermost, _transport + + tcp4 = wire.ipv4(TCP_SEGMENT, proto=6) + options = make_tcp(b'x', options=b'\x02\x04\x05\xb4') + ipv4_options = make_ipv4(TCP_SEGMENT, options=b'\x01\x01\x01\x00') + for label, frame, inner, segment in ( + ('4in4', tunnel(tcp4, 4), tcp4, TCP_SEGMENT), + ('4in6', ipv6_frame(wire.ipv6(tcp4, nxt=4)), tcp4, TCP_SEGMENT), + ('4in4in4', tunnel(wire.ipv4(tcp4, proto=4), 4), tcp4, TCP_SEGMENT), + ('4in6in4', tunnel(wire.ipv6(tcp4, nxt=4), 41), tcp4, TCP_SEGMENT), + ('4in6 behind QinQ', tagged_frame(wire.ipv6(tcp4, nxt=4), 0x86DD, S_TAG, C_TAG), tcp4, + TCP_SEGMENT), + ('4in4 behind ah', tunnel(ah(4) + tcp4, 51), tcp4, TCP_SEGMENT), + ('4in4 behind destination options', tunnel(options_header(4) + tcp4, 60), tcp4, TCP_SEGMENT), + ('outer first fragment', tunnel(tcp4, 4, flags=0b001), tcp4, TCP_SEGMENT), + ('outer padded', tunnel(tcp4 + bytes(6), 4, len=20 + len(tcp4)), tcp4, TCP_SEGMENT), + ('outer ipv4 option', tunnel(tcp4, 4, hl=6, opt=b'\x94\x04\x00\x00'), tcp4, TCP_SEGMENT), + ('inner ipv4 options', tunnel(ipv4_options, 4), ipv4_options, TCP_SEGMENT), + ('inner first fragment', tunnel(wire.ipv4(TCP_SEGMENT, proto=6, mf=True), 4), + wire.ipv4(TCP_SEGMENT, proto=6, mf=True), TCP_SEGMENT), + ('inner total length 0', tunnel(total_length(tcp4, 0), 4), total_length(tcp4, 0), + TCP_SEGMENT), + ('inner ah', tunnel(wire.ipv4(ah(6) + TCP_SEGMENT, proto=51), 4), + wire.ipv4(ah(6) + TCP_SEGMENT, proto=51), TCP_SEGMENT), + ('tcp options', tunnel(wire.ipv4(options, proto=6), 4), wire.ipv4(options, proto=6), options), + # the default engine reads these out of the raw octets (#1518), in a tunnel too + ('tcp option of length 0', tunnel(wire.ipv4(make_tcp(b'x', options=b'\x02\x00\x05\xb4'), + proto=6), 4), + wire.ipv4(make_tcp(b'x', options=b'\x02\x00\x05\xb4'), proto=6), + make_tcp(b'x', options=b'\x02\x00\x05\xb4')), + ('data offset past the capture', tunnel(wire.ipv4(data_offset(TCP_SEGMENT, 15), proto=6), 4), + wire.ipv4(data_offset(TCP_SEGMENT, 15), proto=6), data_offset(TCP_SEGMENT, 15)), + ): + with self.subTest(case=label), mock.patch.dict('sys.modules', fake_ip_decoder()) as modules: + IP = modules['pcapfile.protocols.network.ip'].IP + ipv4, upper = _innermost(make_packet(frame)) + self.assertIsInstance(ipv4, IP) + self.assertIsNone(upper) + self.assertEqual(IP.calls[-1], (inner, 0)) + self.assertEqual(_transport(ipv4), segment) def test_no_warning_where_the_default_engine_finds_no_tcp(self) -> None: + from unittest import mock + 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) @@ -779,16 +828,15 @@ def test_no_warning_where_the_default_engine_finds_no_tcp(self) -> None: # 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)), + ('tunnelled ipv6 data offset 4', tunnel(wire.ipv6(data_offset(TCP_SEGMENT, 4), nxt=6), 41)), # 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')), + # or the inner one (#1581), as the default engine dissects nothing past it + ('inner ipv4 option', tunnel(make_ipv4(TCP_SEGMENT, options=b'\x94\x02\x00\x00'), 4)), + ('6in4 in an inner ipv4 it rejects', + tunnel(make_ipv4(tcp6, protocol=41, options=b'\x94\x02\x00\x00'), 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), @@ -802,7 +850,7 @@ def test_no_warning_where_the_default_engine_finds_no_tcp(self) -> None: # 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): + with self.subTest(case=label), mock.patch.dict('sys.modules', fake_ip_decoder()): self.assertEqual(self.decline(make_packet(frame)), []) def test_network_decodes_the_ipv4_behind_the_tags(self) -> None: @@ -845,35 +893,21 @@ def test_malformed_ipv4_behind_tags_warns_once_per_frame(self) -> None: ]) 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. + # Over IPv6 the default engine reads such a header out of the raw octets + # its TCP parser leaves, in a tunnel too, where that IPv6 layer keys it (#1581). 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', + ('6in4, data offset past the capture', tunnel(wire.ipv6(data_offset(TCP_SEGMENT, 9), nxt=6), 41)), + ('6in6, option past its area', ipv6_frame(wire.ipv6(wire.ipv6(make_tcp(b'x', options=b'\x08\x0a\x00\x00'), nxt=6), nxt=41))), - ('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)) self.assertEqual(len(messages), 1, messages) - self.assertTrue(messages[0].startswith('Frame 1: TCP tunnelled in IP'), messages) + self.assertTrue(messages[0].startswith('Frame 1: TCP over IPv6'), 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 @@ -886,8 +920,10 @@ def test_detection_leaves_the_warnings_filters_and_the_logger_alone(self) -> Non logger = logging.getLogger('pcapkit') level = logger.level + # options the default engine's TCP and IPv4 parsers each reject, so log 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), + FakeEthernet(FakeIP(ah(6) + make_tcp(b'x', options=b'\x02\x00\x05\xb4'), p=51)), + FakeEthernet(FakeIP(TCP_SEGMENT, hl=6, opt=b'\x94\x02\x00\x00')), ipv6_frame(wire.ipv6(TCP_SEGMENT, nxt=6))) def user_warning() -> None: @@ -927,7 +963,7 @@ def spy_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 + packet.header = packets[0].header # one capture, which warns once 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: @@ -935,25 +971,29 @@ def spy_init(this, *args, **kwargs): 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], []) + self.assertEqual([str(item.message).split(': ', 1)[0] for item in caught], ['Frame 1']) # 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: + def test_tcp_over_ipv6_warns_once_per_capture_tunnelled_or_not(self) -> None: + # C.f. #1581: TCP over IPv6 in a tunnel is no kind of its own. + from unittest import mock + 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)) + tunnel(wire.ipv6(TCP_SEGMENT, nxt=6), 41), + tunnel(wire.ipv4(wire.ipv6(TCP_SEGMENT, nxt=6), proto=41), 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 + packet.header = packets[0].header # all five frames of one capture - messages = self.decline(*packets) - self.assertEqual(len(messages), 2, messages) + with mock.patch.dict('sys.modules', fake_ip_decoder()): + messages = self.decline(*packets) + self.assertEqual(len(messages), 1, 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') @@ -1318,45 +1358,24 @@ def test_tcp_payload_ends_at_the_total_length(self) -> None: 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: + def test_engine_warns_of_tunnelled_tcp_over_ipv6_only(self) -> None: + # C.f. #1581: tunnelled TCP is read where its innermost IP layer is + # IPv4, keyed by that layer, and is TCP over IPv6 where it is IPv6. from pcapkit import extract - tcp4 = wire.ipv4(TCP_SEGMENT, proto=6) syn = make_tcp(b'', flags=0x02, options=b'\x02\x04\x05\xb4') frames = ( + tunnels.frame(tunnels.ip(4, TCP_SEGMENT, '10.9.9.1', '10.9.9.2', 6)), + tunnels.tunnelled('4in4', TCP_SEGMENT), + # a tunnelled TCP header with a Data Offset under 5 words is no TCP + tunnels.tunnelled('4in4', data_offset(TCP_SEGMENT, 4)), + tunnels.tunnelled('6in4', TCP_SEGMENT), + # nor warned of again, as for TCP over IPv6 with no tunnel + wire.ethernet(wire.ipv6(TCP_SEGMENT, nxt=6), 0x86DD), # 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), + tunnels.tunnelled('6in6', syn)[:-2], + tagged(tunnels.ip(4, tunnels.ip(4, TCP_SEGMENT, '10.0.0.3', '10.0.0.4', 6), + *tunnels.OUTER[4], 4), 0x0800, C_TAG), ) with tempfile.TemporaryDirectory() as tmp: path = os.path.join(tmp, 'tunnels.pcap') @@ -1372,13 +1391,29 @@ def test_engine_warns_once_per_capture_on_tunnelled_tcp(self) -> None: 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,)]) + self.assertEqual(len(messages), 1, messages) + self.assertTrue(messages[0].startswith('Frame 4: TCP over IPv6'), messages) + self.assertEqual([(flow.label.rsplit('-', 1)[0], tuple(flow.index)) + for flow in extractor.trace.tcp], + [('10.9.9.1_40000-10.9.9.2_9', (1,)), ('10.0.0.1_40000-10.0.0.2_9', (2,)), + ('10.0.0.3_40000-10.0.0.4_9', (7,))]) finally: close_extractor(extractor) + def test_two_connections_through_one_tunnel_are_two_flows(self) -> None: + # C.f. #1581: as for the other engines (see + # tests.toolkit.test_tunnelled_tcp_flow_key_1581_unit), over IPv4 only. + for kind in ('4in4', '4in6'): + with self.subTest(kind=kind): + mine, theirs = self.read_both(tunnels.connections(kind)) + self.assertEqual([index for _, index in mine['flows']], [(1, 2, 3), (4, 5, 6)]) + self.assertEqual([label.rsplit('-', 1)[0] for label, _ in mine['flows']], + ['10.0.0.1_40000-10.0.0.2_9', '10.0.0.3_40000-10.0.0.4_9']) + self.assertEqual(sorted({str(item['id'].src[0]) for item in mine['tcp datagrams']}), + ['10.0.0.1', '10.0.0.3']) + for aspect in self.ASPECTS: + self.assertEqual(mine[aspect], theirs[aspect], aspect) + @unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') @unittest.skipUnless(HAS_PYPCAPFILE, 'pypcapfile not installed or not importable') @@ -1504,31 +1539,44 @@ def test_ipv4_options_agree(self) -> None: ] 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 + def test_tunnelled_tcp_over_ipv4_agrees(self) -> None: + # C.f. #1581: read from, and keyed by, the innermost IPv4 layer. + ip, inner, outer = tunnels.ip, tunnels.INNER[4], tunnels.OUTER - 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) + def in4(packet: bytes, protocol: int = 4, **fields) -> bytes: + return tunnels.frame(ip(4, packet, *outer[4], protocol, **fields)) + + def tcp4(segment: bytes, connection: int = 0, protocol: int = 6, **fields) -> bytes: + return ip(4, segment, *inner[connection], protocol, **fields) + + segment = wire.tcp(b'hello', seq=1, ack=1) + bad_option = make_tcp(b'x', options=b'\x02\x00\x05\xb4') # an MSS of length 0 + frames = [ + tunnels.tunnelled('4in4', segment), + tunnels.tunnelled('4in6', segment, 1), + in4(ip(4, tcp4(segment), '172.16.0.1', '172.16.0.2', 4)), + in4(ip(6, tcp4(segment, 1), 'fd01::1', 'fd01::2', 4), 41), + in4(ah(4) + tcp4(segment), 51), + in4(options_header(4) + tcp4(segment, 1), 60), + in4(tcp4(ah(6) + segment), 4), + in4(tcp4(options_header(6) + segment, 1, 0), 4), + in4(tcp4(segment), 4, mf=True), + in4(tcp4(segment, 1, mf=True), 4), + in4(tcp4(segment)) + bytes(6), + tagged(ip(4, tcp4(segment, 1), *outer[4], 4), 0x0800, C_TAG), + # a TCP header the TCP parser rejects, read out of the raw octets (#1518) + in4(tcp4(data_offset(segment, 9))), + in4(tcp4(bad_option, 1)), + # 15 to 19: no TCP -- an inner header the IPv4 parser rejects, a later + # inner fragment, a Data Offset under 5, and behind AH, no fall-back + # to the raw octets + in4(make_ipv4(segment, options=b'\x94\x02\x00\x00')), + in4(tcp4(segment, offset=8)), + in4(tcp4(data_offset(segment, 4))), + in4(tcp4(ah(6) + bad_option, 1, 51)), + in4(tcp4(ah(6) + data_offset(segment, 9), 0, 51)), + ] + self.agree(frames, tcp=list(range(1, 15)), trace=list(range(1, 15))) if __name__ == '__main__': diff --git a/tests/toolkit/test_pyshark_unit.py b/tests/toolkit/test_pyshark_unit.py index 21c8df081e..8ec7ea7076 100644 --- a/tests/toolkit/test_pyshark_unit.py +++ b/tests/toolkit/test_pyshark_unit.py @@ -74,6 +74,10 @@ def __init__(self, *, ipv6: bool = False, tcp: bool = True, self.tcp = self.layers[2] else: self.layers = self.layers[:2] + if not ip: + # PyShark tests a packet for a layer by its ``layers``, so a packet + # with no IP layer has none there either + del self.layers[1] if not ether: # #775 tier 1 regression coverage: drop the Ethernet layer so # ``layers[0]`` -- what ``tcp_traceflow`` feeds the lookup -- is the diff --git a/tests/toolkit/test_tunnelled_tcp_flow_key_1581_unit.py b/tests/toolkit/test_tunnelled_tcp_flow_key_1581_unit.py new file mode 100644 index 0000000000..ab4f4a991d --- /dev/null +++ b/tests/toolkit/test_tunnelled_tcp_flow_key_1581_unit.py @@ -0,0 +1,309 @@ +# -*- coding: utf-8 -*- +"""TCP in an IP-in-IP tunnel is keyed by the innermost IP header. C.f. #1581. + +The TCP reassembly and flow tracing adapters keyed a segment by the frame's +*first* IP layer, which in a tunnel is the tunnel's own: every connection through +one tunnel that shares a port pair merged into one flow, under the tunnel's +endpoints. The default engine, scapy and pyshark did so (scapy found the inner +IPv4 layer of 4in6 only), and dpkt found no TCP in a tunnel at all. Each now +keys it by the IP layer nearest above the TCP segment, as Wireshark's TCP +conversations do -- and dpkt enters only the tunnels the default engine +dissects, protocols 4 and 41, not the protocol 0 it also decodes as IPv4. + +Every capture is built here, so the module is in the unit tier. pyshark runs +only where the interpreter supports it, with tshark on the path. The +``pypcapfile`` engine's counterparts need `pypcapfile`, so live in +:mod:`tests.toolkit.test_pypcapfile_unit`, which the PyPCAPFile CI cell runs. + +""" +from __future__ import annotations + +import importlib.util +import ipaddress +import io +import os +import tempfile +import unittest +import warnings +from typing import TYPE_CHECKING +from unittest import mock + +from tests._support import close_extractor, reimport_once_per_class +from tests.foundation import _roundtrip as wire + +if TYPE_CHECKING: + from typing import Any + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) + +#: The tunnel's endpoints, by IP version. +OUTER = {4: ('192.0.2.1', '198.51.100.2'), 6: ('2001:db8:ffff::1', '2001:db8:ffff::2')} +#: The endpoints of the two connections through it, by IP version. +INNER = {4: (('10.0.0.1', '10.0.0.2'), ('10.0.0.3', '10.0.0.4')), + 6: (('fd00::1', 'fd00::2'), ('fd00::3', 'fd00::4'))} +#: The IP protocol number that carries each IP version in a tunnel. +TUNNEL = {4: 4, 6: 41} +#: Each tunnel, named inner version first: ``6in4`` is IPv6 carried in IPv4. +KINDS = {'4in4': (4, 4), '6in4': (6, 4), '4in6': (4, 6), '6in6': (6, 6)} +#: Engines that report dissected fields only: no TCP reassembly, and no TCP +#: octets in what they hand flow tracing (as ``NO_TCP_OCTETS`` in +#: :mod:`tests.foundation.engines.test_engine_agreement_runtime`). +FIELDS_ONLY = frozenset(('pyshark',)) + + +def ip(version: int, payload: bytes, src: str, dst: str, proto: int, **fields: 'Any') -> bytes: + """An IPv4 or IPv6 packet from ``src`` to ``dst`` carrying ``payload`` as ``proto``. + + ``fields`` are :func:`tests.foundation._roundtrip.ipv4`'s, for IPv4 only. + + """ + if version == 4: + packet = bytearray(wire.ipv4(payload, proto=proto, **fields)) + packet[12:20] = ipaddress.ip_address(src).packed + ipaddress.ip_address(dst).packed + packet[10:12] = b'\x00\x00' + packet[10:12] = wire.checksum(bytes(packet[:20])).to_bytes(2, 'big') + return bytes(packet) + packet = bytearray(wire.ipv6(payload, nxt=proto)) + packet[8:40] = ipaddress.ip_address(src).packed + ipaddress.ip_address(dst).packed + return bytes(packet) + + +def frame(packet: bytes) -> bytes: + """An Ethernet frame carrying the IPv4 or IPv6 ``packet``.""" + return wire.ethernet(packet, 0x0800 if packet[0] >> 4 == 4 else 0x86DD) + + +def tunnelled(kind: str, segment: bytes, connection: int = 0, *, inner_proto: int = 6) -> bytes: + """An Ethernet frame of a ``kind`` tunnel carrying ``segment`` on ``connection``.""" + inner, outer = KINDS[kind] + packet = ip(inner, segment, *INNER[inner][connection], inner_proto) + return frame(ip(outer, packet, *OUTER[outer], TUNNEL[inner])) + + +def connections(kind: str) -> 'list[bytes]': + """Two connections through one ``kind`` tunnel, on the same port pair. + + Each is a SYN, a data segment and a FIN, from port 40000 to 9. + + """ + return [tunnelled(kind, segment, connection) + for connection in (0, 1) + for segment in (wire.tcp(b'', seq=100, syn=True), wire.tcp(b'hello', seq=101, ack=1), + wire.tcp(b'', seq=106, ack=1, fin=True))] + + +def has(engine: str) -> bool: + """Whether ``engine``'s third-party package is installed.""" + return engine == 'default' or importlib.util.find_spec(engine) is not None + + +def read(test: unittest.TestCase, engine: str, frames: 'list[bytes]', *, + pcapng: bool = False) -> 'dict[str, Any]': + """Read ``frames`` with ``engine``, with TCP reassembly and flow tracing. + + Returns: + Every input the engine hands TCP reassembly and flow tracing, keyed by + frame number -- every field, but a trace input's ``frame``, and its + ``header`` and ``payload`` too for an engine in :data:`FIELDS_ONLY` -- + and the reassembled datagrams' ``id`` and the traced flows' labels and + frames. An engine in :data:`FIELDS_ONLY` reassembles nothing, so has + no ``id`` there. + + """ + from pcapkit import extract + from pcapkit.foundation.reassembly.reassembly import ReassemblyBase + from pcapkit.foundation.traceflow.traceflow import TraceFlowBase + + seen = {'tcp': {}, 'trace': {}} # type: dict[str, Any] + reassemble, trace = ReassemblyBase.__call__, TraceFlowBase.__call__ + fields_only = engine in FIELDS_ONLY + skip = ('frame', 'header', 'payload') if fields_only else ('frame',) + + def fields(packet: 'Any', skip: 'tuple[str, ...]' = ()) -> 'dict[str, Any]': + 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: 'Any', packet: 'Any') -> 'Any': + seen['tcp'][packet.num] = fields(packet) + return reassemble(this, packet) + + def spy_trace(this: 'Any', packet: 'Any') -> 'Any': + seen['trace'][packet.index] = fields(packet, skip) + return trace(this, packet) + + records =[(1, number, octets) for number, octets in enumerate(frames)] + with tempfile.TemporaryDirectory() as tmp, warnings.catch_warnings(), \ + mock.patch.object(ReassemblyBase, '__call__', spy_reassembly), \ + mock.patch.object(TraceFlowBase, '__call__', spy_trace): + path = os.path.join(tmp, 'tunnels.pcapng' if pcapng else 'tunnels.pcap') + with open(path, 'wb') as file: + file.write(wire.pcapng(records) if pcapng else wire.pcap(records)) + warnings.simplefilter('ignore') + extractor = extract(fin=path, nofile=True, engine=engine, reassembly=not fields_only, + tcp=True, trace=True, trace_fout=os.path.join(tmp, 'trace'), + trace_format='json') + try: + if extractor._exnam != engine: # pylint: disable=protected-access + test.skipTest(f'{engine} did not run') + seen['ids'] = None if fields_only else [datagram.id for datagram in extractor.reassembly.tcp] + seen['flows'] = [(flow.label, tuple(flow.index)) for flow in extractor.trace.tcp] + finally: + close_extractor(extractor) + return seen + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class TunnelledTCPFlowKeyTests(unittest.TestCase): + """Every engine keys tunnelled TCP by the innermost IP header, as the default engine does.""" + + ENGINES = ('default', 'dpkt', 'scapy', 'pyshark') + + def setUp(self) -> None: + reimport_once_per_class(self) + + def agree(self, frames: 'list[bytes]', engines: 'tuple[str, ...]' = ENGINES[1:]) -> 'dict[str, Any]': + """Check each of ``engines`` hands on what the default engine does for ``frames``. + + That is every field of every TCP reassembly and flow tracing input, the + reassembled datagrams' ``id`` and the traced flows -- for an engine in + :data:`FIELDS_ONLY`, the flow tracing inputs less their octets, and the + flows. Returns the default engine's. + + """ + base = read(self, 'default', frames) + for engine in engines: + with self.subTest(engine=engine): + if not has(engine): + self.skipTest(f'{engine} not installed') + seen = read(self, engine, frames) + want = base + if engine in FIELDS_ONLY: + want = {'flows': base['flows'], 'trace': { + number: {key: value for key, value in record.items() if key not in ('header', 'payload')} + for number, record in base['trace'].items()}} + for aspect in want: + self.assertEqual(seen[aspect], want[aspect], aspect) + return base + + def test_two_connections_through_one_tunnel_are_two_flows(self) -> None: + for engine in self.ENGINES: + for kind, (inner, _) in KINDS.items(): + for pcapng in ((False, True) if engine == 'default' else (False,)): + with self.subTest(engine=engine, kind=kind, pcapng=pcapng): + if not has(engine): + self.skipTest(f'{engine} not installed') + seen = read(self, engine, connections(kind), pcapng=pcapng) + pairs = sorted(tuple(map(ipaddress.ip_address, pair)) for pair in INNER[inner]) + # one flow per connection, under its own endpoints, not the tunnel's + self.assertEqual( + [label.rsplit('-', 1)[0] for label, _ in seen['flows']], + [f'{src}_40000-{dst}_9'.replace(':', '.') for src, dst in INNER[inner]]) + self.assertEqual([index for _, index in seen['flows']], [(1, 2, 3), (4, 5, 6)]) + self.assertEqual(sorted({(record['src'], record['dst']) + for record in seen['trace'].values()}), pairs) + if engine in FIELDS_ONLY: + continue + # and one reassembly buffer per connection + self.assertEqual(sorted({(d.src[0], d.dst[0]) for d in seen['ids']}), pairs) + self.assertEqual(sorted({(record['bufid'][0], record['bufid'][2]) + for record in seen['tcp'].values()}), pairs) + + def test_protocol_0_is_no_tunnel(self) -> None: + # dpkt decodes an IPv4 protocol 0 as IPv4, its IP_PROTO_IP; the default + # engine reads an IPv6 Hop-by-Hop Options header there, and so finds no + # TCP in the IPv4 packet behind it. + frames = [frame(ip(4, ip(4, segment, *INNER[4][connection], 6), *OUTER[4], 0)) + for connection, segment in enumerate((wire.tcp(b'', seq=100, syn=True), + wire.tcp(b'hello', seq=101, ack=1)))] + base = self.agree(frames) + self.assertEqual((base['trace'], base['flows'], base['ids']), ({}, [], [])) + + def test_tunnelled_segments_agree_with_the_default_engine(self) -> None: + # The other engines hand reassembly and flow tracing every field the + # default engine does, however the tunnel is built. + segment = wire.tcp(b'payload', seq=7, ack=1) + # a Data Offset of 6 words in a 20-octet segment, which the TCP parser + # rejects and the default engine reads out of the raw octets (#1518) + cut = segment[:12] + b'\x60' + segment[13:20] + hop_by_hop = bytes([4, 0, 1, 4, 0, 0, 0, 0]) + frames = [ + *(tunnelled(kind, segment) for kind in KINDS), + # tunnels in tunnels + frame(ip(4, ip(4, ip(4, segment, *INNER[4][0], 6), '172.16.0.1', '172.16.0.2', 4), + *OUTER[4], 4)), + frame(ip(4, ip(6, ip(4, segment, *INNER[4][0], 6), 'fd01::1', 'fd01::2', 4), + *OUTER[4], 41)), + frame(ip(6, ip(4, ip(6, segment, *INNER[6][0], 6), '172.16.0.1', '172.16.0.2', 41), + *OUTER[6], 4)), + # a TCP header cut short, in a tunnel + *(tunnelled(kind, cut) for kind in KINDS), + # behind an IPv6 extension header, and padded + frame(ip(6, hop_by_hop + ip(4, segment, *INNER[4][1], 6), *OUTER[6], 0)), + tunnelled('4in4', wire.tcp(b'', seq=1, ack=1)) + bytes(6), + ] + base = self.agree(frames) + self.assertEqual(sorted(base['trace']), list(range(1, len(frames) + 1))) + + def test_tunnelled_first_fragments_agree_with_the_default_engine(self) -> None: + # Not for pyshark: tshark reassembles IP fragments, so dissects no TCP in + # a first fragment, tunnelled or not. + segment = wire.tcp(b'payload', seq=7, ack=1) + frames = [ + # the tunnel's first fragment, then the inner datagram's + frame(ip(4, ip(4, segment, *INNER[4][1], 6), *OUTER[4], 4, mf=True)), + frame(ip(4, ip(4, segment, *INNER[4][1], 6, mf=True), *OUTER[4], 4)), + ] + base = self.agree(frames, ('dpkt', 'scapy')) + self.assertEqual(sorted(base['trace']), [1, 2]) + + def test_the_default_engine_reads_the_inner_header(self) -> None: + from pcapkit import extract + from pcapkit.toolkit.pcap import tcp_segment + + segment = wire.tcp(b'payload', seq=7, ack=1) + cut = segment[:12] + b'\x60' + segment[13:20] + for label, octets, want in ( + ('4in4', tunnelled('4in4', segment), INNER[4][0]), + ('6in6', tunnelled('6in6', segment, 1), INNER[6][1]), + # the TCP parser rejects the header, so the segment is read out of + # the raw octets the inner IP layer is left with + ('4in4, header cut short', tunnelled('4in4', cut), INNER[4][0]), + ('6in4, header cut short', tunnelled('6in4', cut, 1), INNER[6][1]), + ): + with self.subTest(case=label): + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + extractor = extract(fin=io.BytesIO(wire.pcap([(1, 0, octets)])), nofile=True) + self.addCleanup(close_extractor, extractor) + record = tcp_segment(extractor.frame[0]) + assert record is not None + self.assertEqual((str(record.ip.src), str(record.ip.dst)), want) + self.assertEqual((record.srcport, record.dstport, record.seq), (40000, 9, 7)) + + def test_no_segment_where_the_inner_header_carries_none(self) -> None: + from pcapkit import extract + from pcapkit.toolkit.pcap import tcp_segment + + segment = wire.tcp(b'payload', seq=7, ack=1) + arp = bytes.fromhex('0001080006040001') + bytes(6) + bytes(4) + bytes(6) + bytes(4) + for label, octets in ( + ('no ip layer at all', wire.ethernet(arp, 0x0806)), + ('tunnelled udp', tunnelled('4in4', wire.udp(b'x'), inner_proto=17)), + # an inner header the IPv4 parser rejects leaves the outer one's payload raw + ('inner ihl past the packet', frame(ip(4, b'\x4f' + ip(4, segment, *INNER[4][0], 6)[1:], + *OUTER[4], 4))), + ('inner later fragment', frame(ip(4, ip(4, segment, *INNER[4][0], 6, offset=8), + *OUTER[4], 4))), + ('inner data offset 4', tunnelled('4in4', segment[:12] + b'\x40' + segment[13:])), + ): + with self.subTest(case=label): + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + extractor = extract(fin=io.BytesIO(wire.pcap([(1, 0, octets)])), nofile=True) + self.addCleanup(close_extractor, extractor) + self.assertIsNone(tcp_segment(extractor.frame[0])) + + +if __name__ == '__main__': + unittest.main()