diff --git a/pcapkit/toolkit/pcap.py b/pcapkit/toolkit/pcap.py index 242d91c443..e1c999593d 100644 --- a/pcapkit/toolkit/pcap.py +++ b/pcapkit/toolkit/pcap.py @@ -10,26 +10,136 @@ cannot be used for it. """ -from typing import TYPE_CHECKING, cast +from typing import TYPE_CHECKING, NamedTuple, cast from pcapkit.const.ipv6.extension_header import ExtensionHeader as Enum_ExtensionHeader +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.misc.raw import Raw if TYPE_CHECKING: from ipaddress import IPv4Address, IPv6Address + from typing import Optional from pcapkit.const.reg.linktype import LinkType + from pcapkit.protocols.data.internet.ipv4 import IPv4 as Data_IPv4 + from pcapkit.protocols.data.internet.ipv6 import IPv6 as Data_IPv6 from pcapkit.protocols.internet.ipv4 import IPv4 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'] +class _TCPSegment(NamedTuple): + """The fields of a TCP segment that the reassembly and flow tracing records take.""" + + #: Info of the IP layer carrying the segment. + ip: 'Data_IPv4 | Data_IPv6' + #: TCP source port. + srcport: 'int' + #: TCP destination port. + dstport: 'int' + #: TCP sequence number. + seq: 'int' + #: TCP acknowledgement number. + ack: 'int' + #: TCP synchronise (SYN) flag. + syn: 'bool' + #: TCP finish (FIN) flag. + fin: 'bool' + #: TCP reset (RST) flag. + rst: 'bool' + #: Raw TCP header, as captured. + header: 'bytes' + #: Raw TCP payload, as captured. + payload: 'bytes' + + +def _tcp_segment(frame: 'ProtocolBase') -> 'Optional[_TCPSegment]': + """The TCP segment ``frame`` carries, or :data:`None` if it carries none. + + Args: + 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`. + + Note: + A capture snapped inside the TCP header leaves a Data Offset that runs + past the captured octets, and the TCP parser rejects such a header by + design (:issue:`1404`), so the segment arrives as + :class:`~pcapkit.protocols.misc.raw.Raw`. The ports, numbers and flags + 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. + + """ + if 'TCP' in frame: + tcp = cast('TCP', frame['TCP']) + tcp_info = tcp.info + return _TCPSegment( + ip=cast('IPv4 | IPv6', frame['IP']).info, + srcport=tcp_info.srcport.port, + dstport=tcp_info.dstport.port, + seq=tcp_info.seq, + ack=tcp_info.ack, + syn=tcp_info.flags.syn, + fin=tcp_info.flags.fin, + rst=tcp_info.flags.rst, + header=tcp.packet.header, + 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 + # dissection at a ``layer`` or ``protocol``, not a segment TCP rejected. + if (not isinstance(raw, Raw) or raw.info.error is None + or ip_info.protocol != Enum_TransType.TCP): + return None + + # A later fragment's payload starts mid-datagram, with no TCP header in it. + if ip_info.version == 4: + offset = ip_info.offset + else: + frag = cast('IPv6', ip).extension_headers.get(Enum_ExtensionHeader.IPv6_Frag) # type: ignore[call-overload] + offset = 0 if frag is None else cast('IPv6_Frag', frag).info.offset + if offset: + return None + + octets = bytes(raw.packet.payload) + hdr_len = (octets[12] >> 4) * 4 if len(octets) >= 20 else 0 + if hdr_len < 20: + return None + flags = octets[13] + return _TCPSegment( + ip=ip_info, + srcport=int.from_bytes(octets[0:2], 'big'), + dstport=int.from_bytes(octets[2:4], 'big'), + seq=int.from_bytes(octets[4:8], 'big'), + ack=int.from_bytes(octets[8:12], 'big'), + syn=bool(flags & 0x02), + fin=bool(flags & 0x01), + rst=bool(flags & 0x04), + header=octets[:hdr_len], + payload=octets[hdr_len:], + ) + + def ipv4_reassembly(frame: 'Frame') -> 'IP_Packet[IPv4Address] | None': """Make data for IPv4 reassembly. @@ -150,7 +260,9 @@ def tcp_reassembly(frame: 'Frame') -> 'TCP_Packet | None': A tuple of data for TCP reassembly. * If the ``frame`` can be used for TCP reassembly. A frame can be reassembled - if it contains TCP layer (:class:`~pcapkit.protocols.transport.tcp.TCP`). + if it contains TCP layer (:class:`~pcapkit.protocols.transport.tcp.TCP`), + or a TCP segment whose header the TCP parser rejected, as it does one the + snapshot length cut short (:issue:`1518`). * If the ``frame`` can be reassembled, then the :obj:`dict` mapping of data for TCP reassembly (:term:`reasm.tcp.packet`) will be returned; otherwise, returns :data:`None`. @@ -158,30 +270,25 @@ def tcp_reassembly(frame: 'Frame') -> 'TCP_Packet | None': :class:`pcapkit.foundation.reassembly.tcp.TCP` """ - if 'TCP' in frame: - ip = cast('IPv4 | IPv6', frame['IP']) - ip_info = ip.info - tcp = cast('TCP', frame['TCP']) - tcp_info = tcp.info - - raw_len = len(tcp.packet.payload) + if (segment := _tcp_segment(frame)) is not None: + raw_len = len(segment.payload) data = TCP_Packet( bufid=( - ip_info.src, # source IP address - tcp_info.srcport.port, # source port - ip_info.dst, # destination IP address - tcp_info.dstport.port, # destination port + segment.ip.src, # source IP address + segment.srcport, # source port + segment.ip.dst, # destination IP address + segment.dstport, # destination port ), num=frame.info.number, # original packet range number - ack=tcp_info.ack, # acknowledgement - dsn=tcp_info.seq, # data sequence number - syn=tcp_info.flags.syn, # synchronise flag - fin=tcp_info.flags.fin, # finish flag - rst=tcp_info.flags.rst, # reset connection flag - header=tcp.packet.header, # raw bytes type header - payload=bytearray(tcp.packet.payload), # raw bytearray type payload - first=tcp_info.seq, # first sequence number of payload - last=tcp_info.seq + raw_len - 1, # last sequence number of payload + ack=segment.ack, # acknowledgement + dsn=segment.seq, # data sequence number + syn=segment.syn, # synchronise flag + fin=segment.fin, # finish flag + rst=segment.rst, # reset connection flag + header=segment.header, # raw bytes type header + payload=bytearray(segment.payload), # raw bytearray type payload + first=segment.seq, # first sequence number of payload + last=segment.seq + raw_len - 1, # last sequence number of payload len=raw_len, # payload length, header excludes timestamp=float(frame.info.time_epoch), # capture timestamp ) @@ -200,7 +307,9 @@ def tcp_traceflow(frame: 'Frame', *, data_link: 'LinkType') -> 'TF_TCP_Packet | Data for TCP reassembly. * If the ``packet`` can be used for TCP flow tracing. A frame can be reassembled - if it contains TCP layer (:class:`~pcapkit.protocols.transport.tcp.TCP`). + if it contains TCP layer (:class:`~pcapkit.protocols.transport.tcp.TCP`), + or a TCP segment whose header the TCP parser rejected, as it does one the + snapshot length cut short (:issue:`1518`). * If the ``frame`` 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`. @@ -208,28 +317,23 @@ def tcp_traceflow(frame: 'Frame', *, data_link: 'LinkType') -> 'TF_TCP_Packet | :class:`pcapkit.foundation.traceflow.tcp.TCP` """ - if 'TCP' in frame: - ip = cast('IPv4 | IPv6', frame['IP']) - ip_info = ip.info - tcp = cast('TCP', frame['TCP']) - tcp_info = tcp.info - + if (segment := _tcp_segment(frame)) is not None: data = TF_TCP_Packet( # type: ignore[type-var] protocol=data_link, # data link type from global header index=frame.info.number, # frame number frame=frame.info, # extracted frame info - syn=tcp_info.flags.syn, # TCP synchronise (SYN) flag - fin=tcp_info.flags.fin, # TCP finish (FIN) flag - rst=tcp_info.flags.rst, # TCP reset (RST) flag - src=ip_info.src, # source IP - dst=ip_info.dst, # destination IP - srcport=tcp_info.srcport.port, # TCP source port - dstport=tcp_info.dstport.port, # TCP destination port + syn=segment.syn, # TCP synchronise (SYN) flag + fin=segment.fin, # TCP finish (FIN) flag + rst=segment.rst, # TCP reset (RST) flag + src=segment.ip.src, # source IP + dst=segment.ip.dst, # destination IP + srcport=segment.srcport, # TCP source port + dstport=segment.dstport, # TCP destination port timestamp=float(frame.info.time_epoch), # frame timestamp - seq=tcp_info.seq, # TCP sequence number - ack=tcp_info.ack, # TCP acknowledgement number - header=tcp.packet.header, # raw bytes type header - payload=bytearray(tcp.packet.payload), # raw bytearray type payload + seq=segment.seq, # TCP sequence number + ack=segment.ack, # TCP acknowledgement number + header=segment.header, # raw bytes type header + payload=bytearray(segment.payload), # raw bytearray type payload ) return data return None diff --git a/pcapkit/toolkit/pcapng.py b/pcapkit/toolkit/pcapng.py index f8f7e8441d..117b3000ee 100644 --- a/pcapkit/toolkit/pcapng.py +++ b/pcapkit/toolkit/pcapng.py @@ -18,6 +18,7 @@ from pcapkit.foundation.traceflow.data.tcp import Packet as TF_TCP_Packet from pcapkit.protocols.data.misc.pcap.frame import Frame as Data_Frame from pcapkit.protocols.data.misc.pcap.frame import FrameInfo as Data_FrameInfo +from pcapkit.toolkit.pcap import _tcp_segment if TYPE_CHECKING: from ipaddress import IPv4Address, IPv6Address @@ -26,7 +27,6 @@ from pcapkit.protocols.internet.ipv6 import IPv6 from pcapkit.protocols.internet.ipv6_frag import IPv6_Frag from pcapkit.protocols.misc.pcapng import PCAPNG, Packet - from pcapkit.protocols.transport.tcp import TCP __all__ = ['ipv4_reassembly', 'ipv6_reassembly', 'tcp_reassembly', 'tcp_traceflow', 'block2frame'] @@ -154,7 +154,9 @@ def tcp_reassembly(frame: 'PCAPNG') -> 'TCP_Packet | None': A tuple of data for TCP reassembly. * If the ``frame`` can be used for TCP reassembly. A frame can be reassembled - if it contains TCP layer (:class:`~pcapkit.protocols.transport.tcp.TCP`). + if it contains TCP layer (:class:`~pcapkit.protocols.transport.tcp.TCP`), + or a TCP segment whose header the TCP parser rejected, as it does one the + snapshot length cut short (:issue:`1518`). * If the ``frame`` can be reassembled, then the :obj:`dict` mapping of data for TCP reassembly (:term:`reasm.tcp.packet`) will be returned; otherwise, returns :data:`None`. @@ -162,32 +164,27 @@ def tcp_reassembly(frame: 'PCAPNG') -> 'TCP_Packet | None': :class:`pcapkit.foundation.reassembly.tcp.TCP` """ - if 'TCP' in frame: - ip = cast('IPv4 | IPv6', frame['IP']) - ip_info = ip.info - tcp = cast('TCP', frame['TCP']) - tcp_info = tcp.info - - raw_len = len(tcp.packet.payload) + if (segment := _tcp_segment(frame)) is not None: + raw_len = len(segment.payload) frame_info = cast('Packet', frame.info) data = TCP_Packet( bufid=( - ip_info.src, # source IP address - tcp_info.srcport.port, # source port - ip_info.dst, # destination IP address - tcp_info.dstport.port, # destination port + segment.ip.src, # source IP address + segment.srcport, # source port + segment.ip.dst, # destination IP address + segment.dstport, # destination port ), num=frame_info.number, # original packet range number - ack=tcp_info.ack, # acknowledgement - dsn=tcp_info.seq, # data sequence number - syn=tcp_info.flags.syn, # synchronise flag - fin=tcp_info.flags.fin, # finish flag - rst=tcp_info.flags.rst, # reset connection flag - header=tcp.packet.header, # raw bytes type header - payload=bytearray(tcp.packet.payload), # raw bytearray type payload - first=tcp_info.seq, # first sequence number of payload - last=tcp_info.seq + raw_len - 1, # last sequence number of payload + ack=segment.ack, # acknowledgement + dsn=segment.seq, # data sequence number + syn=segment.syn, # synchronise flag + fin=segment.fin, # finish flag + rst=segment.rst, # reset connection flag + header=segment.header, # raw bytes type header + payload=bytearray(segment.payload), # raw bytearray type payload + first=segment.seq, # first sequence number of payload + last=segment.seq + raw_len - 1, # last sequence number of payload len=raw_len, # payload length, header excludes timestamp=float(frame_info.timestamp_epoch), # capture timestamp ) @@ -206,7 +203,9 @@ def tcp_traceflow(frame: 'PCAPNG', *, nanosecond: 'bool' = False) -> 'TF_TCP_Pac Data for TCP reassembly. * If the ``packet`` can be used for TCP flow tracing. A frame can be reassembled - if it contains TCP layer (:class:`~pcapkit.protocols.transport.tcp.TCP`). + if it contains TCP layer (:class:`~pcapkit.protocols.transport.tcp.TCP`), + or a TCP segment whose header the TCP parser rejected, as it does one the + snapshot length cut short (:issue:`1518`). * If the ``frame`` 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`. @@ -214,30 +213,25 @@ def tcp_traceflow(frame: 'PCAPNG', *, nanosecond: 'bool' = False) -> 'TF_TCP_Pac :class:`pcapkit.foundation.traceflow.tcp.TCP` """ - if 'TCP' in frame: - ip = cast('IPv4 | IPv6', frame['IP']) - ip_info = ip.info - tcp = cast('TCP', frame['TCP']) - tcp_info = tcp.info - + if (segment := _tcp_segment(frame)) is not None: frame_info = cast('Packet', frame.info) data = TF_TCP_Packet( # type: ignore[type-var] protocol=frame.linktype, # data link type from global header index=frame_info.number, # frame number frame=block2frame(frame_info, nanosecond=nanosecond), # extracted frame info - syn=tcp_info.flags.syn, # TCP synchronise (SYN) flag - fin=tcp_info.flags.fin, # TCP finish (FIN) flag - rst=tcp_info.flags.rst, # TCP reset (RST) flag - src=ip_info.src, # source IP - dst=ip_info.dst, # destination IP - srcport=tcp_info.srcport.port, # TCP source port - dstport=tcp_info.dstport.port, # TCP destination port + syn=segment.syn, # TCP synchronise (SYN) flag + fin=segment.fin, # TCP finish (FIN) flag + rst=segment.rst, # TCP reset (RST) flag + src=segment.ip.src, # source IP + dst=segment.ip.dst, # destination IP + srcport=segment.srcport, # TCP source port + dstport=segment.dstport, # TCP destination port timestamp=float(frame_info.timestamp_epoch), # frame timestamp - seq=tcp_info.seq, # TCP sequence number - ack=tcp_info.ack, # TCP acknowledgement number - header=tcp.packet.header, # raw bytes type header - payload=bytearray(tcp.packet.payload), # raw bytearray type payload + seq=segment.seq, # TCP sequence number + ack=segment.ack, # TCP acknowledgement number + header=segment.header, # raw bytes type header + payload=bytearray(segment.payload), # raw bytearray type payload ) return data return None diff --git a/tests/foundation/traceflow/test_traceflow_snapped_capture_runtime.py b/tests/foundation/traceflow/test_traceflow_snapped_capture_runtime.py new file mode 100644 index 0000000000..c82c2e9c99 --- /dev/null +++ b/tests/foundation/traceflow/test_traceflow_snapped_capture_runtime.py @@ -0,0 +1,184 @@ +# -*- coding: utf-8 -*- +"""A capture snapped inside the TCP headers traces as the full one does (#1518). + +Every frame of :file:`tcp.pcap` snapped by 4 octets, with its original length +kept, cuts four of its seven TCP headers short: frame 1 inside its options and +frames 2, 4 and 7 inside a Timestamps option. The TCP parser rejects those +headers by design (#1404), and the default engine's adapters used to skip the +frames, tracing ``(3, 5) (6,)``. :mod:`dpkt`, :mod:`scapy` and Wireshark's +``tcp.stream`` all trace ``(1, 2, 6, 7) (3, 4, 5)``, the flows of the full +capture, and so does the default engine now that its adapters read the fixed +20-octet header of a rejected segment. + +The snapped capture is built in memory, as PCAP and as PCAP-NG. The module +reads the generated :file:`tcp.pcap`, so it belongs to the fixture-dependent +tier; :mod:`tests.toolkit.test_pcap_tcp_segment_unit` covers the adapters' +edge cases on frames built by hand. + +""" +from __future__ import annotations + +import importlib.util +import io +import os +import struct +import tempfile +import unittest +import warnings +from typing import TYPE_CHECKING + +from tests._support import close_extractor, reimport_once_per_class, sample_path + +if TYPE_CHECKING: + from typing import Any + +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None + for name in ('tbtrim', 'aenum', 'chardet', 'dictdumper')) +ENGINES = tuple(name for name in ('dpkt', 'scapy') if importlib.util.find_spec(name) is not None) +CONTAINERS = ('pcap', 'pcapng') + +#: ``(index, forward, reverse)`` of each bidirectional flow, as every engine and +#: Wireshark trace both the full and the snapped capture. +FLOWS = [((1, 2, 6, 7), (1, 7), (2, 6)), ((3, 4, 5), (3,), (4, 5))] +#: The same, one flow per direction. +UNIDIRECTIONAL = [((1, 7), (1, 7), ()), ((2, 6), (2, 6), ()), ((3,), (3,), ()), ((4, 5), (4, 5), ())] +#: Frame numbers of each reassembled TCP datagram, as :mod:`dpkt` gives them. +DATAGRAMS = [(2, 6), (3,), (4, 5)] + + +def snapped_records(cut: 'int' = 4) -> 'tuple[bytes, list[tuple[int, int, int, bytes]]]': + """:file:`tcp.pcap`'s global header, and its ``(ts_sec, ts_usec, orig_len, + octets)`` records with the last ``cut`` octets of each left uncaptured.""" + with open(sample_path('tcp.pcap'), 'rb') as file: + data = file.read() + header, rest, records = data[:24], data[24:], [] + while rest: + ts_sec, ts_usec, incl_len, orig_len = struct.unpack(' 'bytes': + body += b'\x00' * (-len(body) % 4) + return struct.pack(' 'bytes': + """The snapped capture, as a little-endian microsecond PCAP or PCAP-NG.""" + header, records = snapped_records() + if container == 'pcap': + return header + b''.join(struct.pack('> 32, ticks & 0xFFFFFFFF, + len(octets), orig_len) + octets)) + return b''.join(blocks) + + +def pcap_records(data: 'bytes') -> 'list[tuple[int, int, int, bytes]]': + """The ``(ts_sec, ts_usec, orig_len, octets)`` records of a little-endian PCAP.""" + rest, records = data[24:], [] + while rest: + ts_sec, ts_usec, incl_len, orig_len = struct.unpack(' None: + reimport_once_per_class(self) + tmp = tempfile.TemporaryDirectory() # pylint: disable=consider-using-with + self.addCleanup(tmp.cleanup) + self.tmp = tmp.name + + def _extract(self, container: 'str', engine: 'str' = 'default', **kwargs: 'Any') -> 'Any': + """Extract the snapped capture: from memory for the default engine and + :mod:`dpkt`, from a file for :mod:`scapy`, which opens savefiles by name.""" + from pcapkit import extract + + data = snapped_capture(container) + if engine == 'scapy': + fin = os.path.join(self.tmp, f'snapped.{container}') # type: Any + with open(fin, 'wb') as file: + file.write(data) + else: + fin = io.BytesIO(data) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + extractor = extract(fin=fin, nofile=True, engine=engine, tcp=True, **kwargs) + self.addCleanup(close_extractor, extractor) + return extractor + + def _flows(self, container: 'str', engine: 'str' = 'default', **kwargs: 'Any') -> 'list[tuple]': + extractor = self._extract(container, engine, trace=True, **kwargs) + return [(tuple(flow.index), tuple(flow.forward), tuple(flow.reverse), flow.label) + for flow in extractor.trace.tcp] + + def test_the_default_engine_traces_every_frame(self) -> None: + for container in CONTAINERS: + for bidirectional, expected in ((True, FLOWS), (False, UNIDIRECTIONAL)): + with self.subTest(container=container, bidirectional=bidirectional): + flows = self._flows(container, trace_bidirectional=bidirectional) + self.assertEqual([flow[:3] for flow in flows], expected) + + @unittest.skipUnless(ENGINES, 'neither dpkt nor scapy installed') + def test_the_other_engines_agree(self) -> None: + for container in CONTAINERS: + for bidirectional in (True, False): + base = self._flows(container, trace_bidirectional=bidirectional) + for engine in ENGINES: + with self.subTest(container=container, bidirectional=bidirectional, engine=engine): + self.assertEqual(self._flows(container, engine, trace_bidirectional=bidirectional), base) + + def _trace_files(self, container: 'str', engine: 'str' = 'default') -> 'dict[str, bytes]': + fout = tempfile.mkdtemp(dir=self.tmp) + extractor = self._extract(container, engine, trace=True, trace_fout=fout, trace_format='pcap') + files = {} + for flow in extractor.trace.tcp: + with open(flow.fpout, 'rb') as file: + files[flow.label] = file.read() + return files + + def test_each_traced_pcap_holds_the_captured_records(self) -> None: + # Byte for byte: each flow's file holds its frames' records as captured, + # snapped octets and original length alike (#1543). + _, records = snapped_records() + for container in CONTAINERS: + with self.subTest(container=container): + files = self._trace_files(container) + self.assertEqual(len(files), len(FLOWS)) + for (index, _, _), data in zip(FLOWS, files.values()): + self.assertEqual(pcap_records(data), [records[number - 1] for number in index]) + + @unittest.skipUnless('dpkt' in ENGINES, 'dpkt not installed') + def test_each_traced_pcap_matches_dpkt(self) -> None: + for container in CONTAINERS: + with self.subTest(container=container): + self.assertEqual(self._trace_files(container), self._trace_files(container, 'dpkt')) + + def _datagrams(self, container: 'str', engine: 'str' = 'default') -> 'list[tuple]': + extractor = self._extract(container, engine, reassembly=True, reasm_store=True) + return [(tuple(datagram.index), bytes(datagram.payload)) for datagram in extractor.reassembly.tcp] + + def test_reassembly_takes_every_frame(self) -> None: + for container in CONTAINERS: + with self.subTest(container=container): + self.assertEqual([index for index, _ in self._datagrams(container)], DATAGRAMS) + + @unittest.skipUnless('dpkt' in ENGINES, 'dpkt not installed') + def test_reassembly_matches_dpkt(self) -> None: + for container in CONTAINERS: + with self.subTest(container=container): + self.assertEqual(self._datagrams(container), self._datagrams(container, 'dpkt')) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/toolkit/test_pcap_tcp_segment_unit.py b/tests/toolkit/test_pcap_tcp_segment_unit.py new file mode 100644 index 0000000000..37f949fb8c --- /dev/null +++ b/tests/toolkit/test_pcap_tcp_segment_unit.py @@ -0,0 +1,209 @@ +# -*- coding: utf-8 -*- +"""The default engine's TCP adapters read a TCP header the parser rejected (#1518). + +A capture snapped inside the TCP header leaves a Data Offset running past the +captured octets, and the TCP parser rejects such a header by design (#1404), so +the segment arrives as ``Raw``. The flow tracing and reassembly adapters of +:mod:`pcapkit.toolkit.pcap` and :mod:`pcapkit.toolkit.pcapng` used to skip it; +they now take the ports, numbers and flags from the fixed 20-octet header, as +:mod:`dpkt`, :mod:`scapy` and Wireshark do. + +Every frame is built here, written as PCAP and as PCAP-NG, and read back through +the default engine, so the adapters see the frames a capture produces. The +module reads no sample capture, so it belongs to the unit tier; +:mod:`tests.foundation.traceflow.test_traceflow_snapped_capture_runtime` covers +the snapped ``tcp.pcap`` of the issue. + +""" +from __future__ import annotations + +import importlib.util +import io +import struct +import unittest +import warnings +from typing import TYPE_CHECKING + +from tests._support import close_extractor, reimport_once_per_class + +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) + +ETH_IPV4 = b'\xbb' * 6 + b'\xaa' * 6 + b'\x08\x00' +ETH_IPV6 = b'\xbb' * 6 + b'\xaa' * 6 + b'\x86\xdd' + +#: MSS, NOP, Window Scale, NOP, NOP and SACK Permitted: twelve octets, so a +#: 32-octet header. +OPTIONS = bytes.fromhex('020405b4' '01030306' '01010402') + + +def tcp(*, words: 'int' = 8, options: 'bytes' = OPTIONS, payload: 'bytes' = b'', + flags: 'int' = 0x12) -> 'bytes': + """A TCP segment from port 1234 to 80, ``seq`` 1000 and ``ack`` 2000.""" + return struct.pack('!HHIIBBHHH', 1234, 80, 1000, 2000, words << 4, flags, 65535, 0, 0) + options + payload + + +def ipv4(segment: 'bytes', *, offset: 'int' = 0, mf: 'bool' = False, protocol: 'int' = 6) -> 'bytes': + """An Ethernet frame of IPv4 carrying ``segment`` as ``protocol``, TCP by + default, at fragment ``offset`` octets.""" + fragment = (0x2000 if mf else 0) | offset // 8 + return ETH_IPV4 + struct.pack('!BBHHHBBH4s4s', 0x45, 0, 20 + len(segment), 1, fragment, 64, protocol, 0, + bytes([192, 0, 2, 1]), bytes([198, 51, 100, 1])) + segment + + +def ipv6(segment: 'bytes', *, fragment: 'tuple[int, bool] | None' = None) -> 'bytes': + """An Ethernet frame of IPv6 carrying ``segment`` as TCP, behind a Fragment + header of ``(offset, M flag)`` if given.""" + next_header = 6 + if fragment is not None: + segment = struct.pack('!BBHI', 6, 0, fragment[0] // 8 << 3 | fragment[1], 99) + segment + next_header = 44 + return ETH_IPV6 + struct.pack('!IHBB16s16s', 6 << 28, len(segment), next_header, 64, + bytes.fromhex('20010db8' + '00' * 11 + '01'), + bytes.fromhex('20010db8' + '00' * 11 + '02')) + segment + + +def whole(frame: 'bytes') -> 'tuple[bytes, int]': + """``frame`` captured in full, and its original length.""" + return frame, len(frame) + + +def snap(frame: 'bytes', cut: 'int' = 4) -> 'tuple[bytes, int]': + """``frame`` with its last ``cut`` octets left uncaptured, and its original length.""" + return frame[:len(frame) - cut], len(frame) + + +def pcap(octets: 'bytes', orig_len: 'int') -> 'bytes': + """A little-endian microsecond Ethernet PCAP holding the one record.""" + return (struct.pack(' 'bytes': + body += b'\x00' * (-len(body) % 4) + return struct.pack(' 'bytes': + """A little-endian PCAP-NG holding the one record, on an Ethernet interface.""" + ticks = 1500000000 * 10 ** 6 + 774 + return (_block(0x0A0D0D0A, struct.pack('> 32, ticks & 0xFFFFFFFF, len(octets), orig_len) + octets)) + + +#: SYN, as :func:`tcp` sets it by default with ACK, as ``(SYN, FIN, RST)``. +SYN = (True, False, False) + +#: The record each adapter reports for a header TCP rejected, as +#: ``(header length, payload, (SYN, FIN, RST))``; the ports and numbers are +#: those :func:`tcp` writes. +DECODED = { + 'header cut by the snapshot': (snap(ipv4(tcp())), (28, b'', SYN)), + 'header cut, IPv6': (snap(ipv6(tcp())), (28, b'', SYN)), + 'header cut, first fragment': (snap(ipv4(tcp(), mf=True)), (28, b'', SYN)), + 'header cut, IPv6 first fragment': (snap(ipv6(tcp(), fragment=(0, True))), (28, b'', SYN)), + # MSS with a length of 10, running past a 24-octet header: the header is + # whole, so the payload after it is the segment's own + 'malformed option': (whole(ipv4(tcp(words=6, options=bytes.fromhex('020a0000'), payload=b'hello'))), + (24, b'hello', SYN)), + # RST with ACK, then FIN with RST and ACK: each flag is read from its own bit + 'header cut, RST': (snap(ipv4(tcp(flags=0x14))), (28, b'', (False, False, True))), + 'header cut, FIN and RST': (snap(ipv4(tcp(flags=0x15))), (28, b'', (False, True, True))), +} + +#: Frames the adapters still report nothing for, though their IP payload is a +#: ``Raw`` its parser rejected. +SKIPPED = { + # 19 octets is short of even the fixed header + 'fewer than 20 octets': snap(ipv4(tcp()), 13), + # a Data Offset of 4 declares a header shorter than the fixed one + 'data offset below 5': whole(ipv4(tcp(words=4, options=b'', payload=b'abcd'))), + # a later fragment starts mid-datagram, so its first octets are no header + 'later fragment': whole(ipv4(tcp(words=15, options=b''), offset=8)), + 'IPv6 later fragment': whole(ipv6(tcp(words=15, options=b''), fragment=(8, False))), + # IPv4 in IPv4, the inner header rejected: 30 octets whose 13th reads as a + # Data Offset of 5, so only the protocol says it is no TCP segment + 'IPv4 in IPv4': whole(ipv4(bytes(12) + b'\x50' + bytes(17), protocol=4)), +} + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class TCPSegmentAdapterTests(unittest.TestCase): + """The PCAP and PCAP-NG adapters take the frames TCP rejected, and only those.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _records(self, capture: 'bytes', **kwargs: 'Any') -> 'tuple[Any, Any, Any]': + """``(traced flows, flow tracing record, reassembly record)`` of the one frame.""" + from pcapkit import extract + from pcapkit.const.reg.linktype import LinkType + from pcapkit.toolkit import pcap as toolkit_pcap + from pcapkit.toolkit import pcapng as toolkit_pcapng + + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + extractor = extract(fin=io.BytesIO(capture), nofile=True, store=True, tcp=True, + trace=True, **kwargs) + self.addCleanup(close_extractor, extractor) + flows = [tuple(flow.index) for flow in extractor.trace.tcp] + frame = extractor.frame[-1] + if capture[:4] == b'\x0a\x0d\x0d\x0a': + return flows, toolkit_pcapng.tcp_traceflow(frame), toolkit_pcapng.tcp_reassembly(frame) + return (flows, toolkit_pcap.tcp_traceflow(frame, data_link=LinkType.ETHERNET), + toolkit_pcap.tcp_reassembly(frame)) + + def test_a_rejected_header_is_read_from_the_fixed_header(self) -> None: + for name, ((octets, orig_len), (hdr_len, payload, flags)) in DECODED.items(): + for container, write in (('pcap', pcap), ('pcapng', pcapng)): + with self.subTest(case=name, container=container): + flows, trace, reasm = self._records(write(octets, orig_len)) + self.assertEqual(flows, [(1,)]) + self.assertIsNotNone(trace) + self.assertIsNotNone(reasm) + assert trace is not None and reasm is not None + self.assertEqual((trace.srcport, trace.dstport, trace.seq, trace.ack), + (1234, 80, 1000, 2000)) + self.assertEqual((trace.syn, trace.fin, trace.rst), flags) + # the TCP octets are the frame's last, the header then the payload + start = len(octets) - len(payload) - hdr_len + self.assertEqual(trace.header, octets[start:start + hdr_len]) + self.assertEqual(bytes(trace.payload), payload) + self.assertEqual((reasm.bufid[1], reasm.bufid[3], reasm.dsn, reasm.ack), + (1234, 80, 1000, 2000)) + self.assertEqual((reasm.syn, reasm.fin, reasm.rst), flags) + self.assertEqual((reasm.header, bytes(reasm.payload)), (trace.header, payload)) + self.assertEqual((reasm.first, reasm.last, reasm.len), + (1000, 1000 + len(payload) - 1, len(payload))) + + def test_no_tcp_header_is_read_where_there_is_none(self) -> None: + for name, (octets, orig_len) in SKIPPED.items(): + for container, write in (('pcap', pcap), ('pcapng', pcapng)): + with self.subTest(case=name, container=container): + self.assertEqual(self._records(write(octets, orig_len)), ([], None, None)) + + def test_a_dissection_stopped_at_the_internet_layer_traces_nothing(self) -> None: + # The payload is Raw because the caller asked for no more, not because TCP + # rejected it, so the adapters leave it alone, as they did before #1518. + for container, write in (('pcap', pcap), ('pcapng', pcapng)): + for kwargs in ({'layer': 'Internet'}, {'protocol': 'IPv4'}): + with self.subTest(container=container, **kwargs): + self.assertEqual(self._records(write(*snap(ipv4(tcp()))), **kwargs), ([], None, None)) + + def test_a_parsed_segment_is_reported_as_before(self) -> None: + octets = ipv4(tcp(payload=b'data')) + for container, write in (('pcap', pcap), ('pcapng', pcapng)): + with self.subTest(container=container): + flows, trace, reasm = self._records(write(octets, len(octets))) + self.assertEqual(flows, [(1,)]) + assert trace is not None and reasm is not None + self.assertEqual((trace.header, bytes(trace.payload)), (octets[34:66], b'data')) + self.assertEqual((reasm.header, bytes(reasm.payload), reasm.len), (octets[34:66], b'data', 4)) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/toolkit/test_pcap_unit.py b/tests/toolkit/test_pcap_unit.py index e99126092d..ffeec828b8 100644 --- a/tests/toolkit/test_pcap_unit.py +++ b/tests/toolkit/test_pcap_unit.py @@ -56,6 +56,9 @@ def _make_ipv4(self, *, df: bool = False): len=28, ), packet=types.SimpleNamespace(header=b'I' * 20, payload=b'ipv4data'), + # the next layer, which the TCP adapters inspect when there is no TCP + # layer (#1518); a frame without TCP here carries no ``Raw`` either + payload=None, ) def _make_ipv6(self, *, with_fragment: bool = True):