refactor(toolkit): revert pyshark's agreement-only code (#1708)

docs/source/pcapkit/foundation/engines/index.rst

@@ -76,10 +76,7 @@ offers, and the ones that cannot say so rather than quietly doing less:
 | Engine                                                          | Gap, and how it is surfaced                                   |
 +=================================================================+===============================================================+
 | :class:`~pcapkit.foundation.engines.pyshark.PyShark`            | no reassembly -- disabled with an                             |
-|                                                                 | :class:`~pcapkit.utilities.warnings.AttributeWarning`, and    |
-|                                                                 | :program:`tshark`'s own IP reassembly is turned off, so a     |
-|                                                                 | fragmented datagram is read frame by frame, as the built-in   |
-|                                                                 | engine reads it                                               |
+|                                                                 | :class:`~pcapkit.utilities.warnings.AttributeWarning`         |
 +-----------------------------------------------------------------+---------------------------------------------------------------+
 | :class:`~pcapkit.foundation.engines.pypcap.PyPCAP`              | no protocol dissection at all, hence no reassembly and no     |
 |                                                                 | flow tracing -- both disabled with an                         |

tests/foundation/engines/test_runtime_engines.py

@@ -448,8 +448,7 @@ def close(self) -> None:
         engine = PyShark(extractor)
         with mock.patch.object(engine._expkg, 'FileCapture', return_value=capture) as file_capture:
             engine.run()
-        file_capture.assert_called_once_with('capture.pcap', keep_packets=False,
-                                             override_prefs=PyShark.OVERRIDE_PREFS)
+        file_capture.assert_called_once_with('capture.pcap', keep_packets=False)
         extractor.record_header.assert_called_once_with()
         self.assertFalse(extractor._flag_r)
         self.assertIsNone(extractor._reasm.ipv4)

tests/toolkit/test_pyshark_unit.py

@@ -680,7 +680,6 @@ def test_the_engine_passes_its_frame_count(self) -> None:
         engine = PyShark.__new__(PyShark)  # skip __init__, which imports pyshark
         engine._extractor = extractor
         engine._extmp = types.SimpleNamespace(next=lambda: next(stream))
-        engine._exraw_failed = True  # no capture file to read frames from: traced as tshark dissects them (#1639)
 
         counts = []
         while True:
@@ -819,66 +818,81 @@ def test_a_whole_header_is_traced(self) -> None:
 
 
 @unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed')
-class PySharkReassemblyPrefsTests(unittest.TestCase):
-    """#1612: the engine runs tshark with IPv4 and IPv6 reassembly off."""
+class PySharkAddressTests(unittest.TestCase):
+    """#1671: a TCP layer whose IP layer lacks an address is not traced.
+
+    tshark reports no ``ip.dst`` behind an LSRR or SSRR option whose route data
+    is no whole number of addresses. Reading it raised a bare
+    :exc:`AttributeError` and aborted the whole extraction.
+
+    """
 
     def setUp(self) -> None:
         reimport_once_per_class(self)
 
-    def test_run_turns_tshark_ip_reassembly_off(self) -> None:
-        from pcapkit.foundation.engines.pyshark import PyShark
+    def test_an_ip_layer_without_an_address_is_skipped(self) -> None:
+        from pcapkit.toolkit import pyshark as toolkit
 
-        file_capture = mock.Mock(return_value='capture')
-        extractor = types.SimpleNamespace(
-            _exlyr='none', _exptl='null', _exctx=None, _flag_r=False, _ipv4=False, _ipv6=False,
-            _tcp=True, _flag_v=False, _ifnm='in.pcap', record_header=mock.Mock())
-        engine = PyShark.__new__(PyShark)  # skip __init__, which imports pyshark
-        engine._extractor = extractor
-        engine._expkg = types.SimpleNamespace(FileCapture=file_capture)
-        engine.run()
-        file_capture.assert_called_once_with(
-            'in.pcap', keep_packets=False,
-            override_prefs={'ip.defragment': 'FALSE', 'ipv6.defragment': 'FALSE'})
-        self.assertEqual(engine._extmp, 'capture')
+        for ipv6 in (False, True):
+            for missing in ('src', 'dst'):
+                with self.subTest(ipv6=ipv6, missing=missing):
+                    packet = FakePySharkPacket(ipv6=ipv6, encap_type='1')
+                    fields = {name: getattr(packet.layers[1], name) for name in ('src', 'dst') if name != missing}
+                    packet.layers[1] = FakeLayer(packet.layers[1].layer_name, **fields)
+                    self.assertIsNone(toolkit.tcp_traceflow(packet, count=3))
+
+    def test_an_ip_layer_with_both_addresses_is_traced(self) -> None:
+        import ipaddress
+
+        from pcapkit.toolkit import pyshark as toolkit
+
+        for ipv6, addresses in ((False, ('192.0.2.1', '198.51.100.1')), (True, ('2001:db8::1', '2001:db8::2'))):
+            with self.subTest(ipv6=ipv6):
+                data = toolkit.tcp_traceflow(FakePySharkPacket(ipv6=ipv6, encap_type='1'), count=3)
+                self.assertEqual((data.index, data.src, data.dst),
+                                 (3, *map(ipaddress.ip_address, addresses)))
 
 
-def tcp_frame(version: int, segment: bytes, *, reverse: bool = False) -> bytes:
+def tcp_frame(version: int, segment: bytes) -> bytes:
     """An Ethernet frame of an IPv4 or IPv6 packet carrying the TCP ``segment``."""
     if version == 4:
-        return wire.ethernet(wire.ipv4(segment, proto=6, reverse=reverse), 0x0800, reverse=reverse)
-    return wire.ethernet(wire.ipv6(segment, nxt=6, reverse=reverse), 0x86DD, reverse=reverse)
-
-
-def fragments(version: int, segment: bytes, ident: int, size: int = 24) -> 'list[bytes]':
-    """The TCP ``segment`` as one IPv4 or IPv6 datagram, cut into ``size``-octet fragments."""
-    frames = []
-    for offset in range(0, len(segment), size):
-        piece, more = segment[offset:offset + size], offset + size < len(segment)
-        if version == 4:
-            frames.append(wire.ethernet(wire.ipv4(piece, proto=6, ident=ident, offset=offset, mf=more),
-                                        0x0800))
-        else:
-            frames.append(wire.ethernet(wire.ipv6_fragment(piece, nxt=6, ident=ident, offset=offset,
-                                                           mf=more), 0x86DD))
-    return frames
+        return wire.ethernet(wire.ipv4(segment, proto=6), 0x0800)
+    return wire.ethernet(wire.ipv6(segment, nxt=6), 0x86DD)
+
+
+def route_option(kind: int, length: int, data: bytes) -> bytes:
+    """An IPv4 source route option of ``kind`` claiming ``length`` octets, pointer 4, then ``data``, padded."""
+    raw = bytes([kind, length, 4]) + data
+    return raw + bytes(-len(raw) % 4)
 
 
-def read_trace(test: unittest.TestCase, engine: str, frames: 'list[bytes]') -> 'dict[str, Any]':
-    """Read ``frames`` with ``engine``, with TCP flow tracing on.
+def with_options(packet: bytes, options: bytes) -> bytes:
+    """The IPv4 ``packet`` with ``options`` after its fixed header, its length fields and checksum updated."""
+    header = bytearray(packet[:20] + options)
+    header[0] = 0x40 | len(header) // 4
+    header[2:4] = (len(packet) + len(options)).to_bytes(2, 'big')
+    header[10:12] = b'\x00\x00'
+    header[10:12] = wire.checksum(bytes(header)).to_bytes(2, 'big')
+    return bytes(header) + packet[20:]
+
+
+def read_trace(test: unittest.TestCase, frames: 'list[bytes]') -> 'dict[str, Any]':
+    """Read ``frames`` with the PyShark engine, with TCP flow tracing on.
 
     Returns:
         The input flow tracing receives for each frame number -- every field but
         ``frame``, ``header`` and ``payload``, which PyShark cannot supply -- and
         the traced flows' labels and frames.
 
-    Skips ``test`` unless ``engine`` ran, which for PyShark needs ``pyshark``,
-    the ``tshark`` binary and Python 3.13 or older.
+    Skips ``test`` unless the engine ran, which needs ``pyshark``, the
+    ``tshark`` binary and Python 3.13 or older.
 
     """
     from pcapkit import extract
     from pcapkit.foundation.traceflow.traceflow import TraceFlowBase
 
-    if engine != 'default' and importlib.util.find_spec(engine) is None:
+    engine = 'pyshark'
+    if importlib.util.find_spec(engine) is None:
         test.skipTest(f'{engine} not installed')
     trace = TraceFlowBase.__call__  # type: Callable[..., Any]
     seen = {}  # type: dict[int, dict[str, Any]]
@@ -907,7 +921,7 @@ def spy(this: 'Any', packet: 'Any') -> 'Any':
 
 @unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed')
 class PySharkTsharkTraceTests(unittest.TestCase):
-    """The PyShark engine traces what the default engine does, read by a real tshark.
+    """What the PyShark engine traces, read by a real tshark.
 
     Every capture is built here. Each test skips unless the PyShark engine runs.
 
@@ -916,14 +930,6 @@ class PySharkTsharkTraceTests(unittest.TestCase):
     def setUp(self) -> None:
         reimport_once_per_class(self)
 
-    def agree(self, frames: 'list[bytes]') -> 'dict[str, Any]':
-        """Check PyShark traces ``frames`` as the default engine does, frame by frame."""
-        base = read_trace(self, 'default', frames)
-        seen = read_trace(self, 'pyshark', frames)
-        self.assertEqual(seen['trace'], base['trace'])
-        self.assertEqual(seen['flows'], base['flows'])
-        return base
-
     def test_a_tcp_header_tshark_cannot_dissect_whole_is_skipped(self) -> None:
         # #1611: one good flow, then a segment with each Data Offset, then the
         # same segment cut at each octet of its header and at its end
@@ -937,47 +943,28 @@ def test_a_tcp_header_tshark_cannot_dissect_whole_is_skipped(self) -> None:
                            for words in range(16)]
                 whole = tcp_frame(version, segment)
                 cuts = [whole[:len(whole) - len(segment) + octets] for octets in range(21)]
-                base = self.agree(good + offsets + cuts)
+                seen = read_trace(self, good + offsets + cuts)
                 # frames 4-19 are Data Offsets 0-15, and 20-40 cuts at 0-20 octets
-                self.assertEqual(sorted(base['trace']), [1, 2, 3, *range(9, 20), 40])
-
-    def test_fragmented_tcp_is_traced_at_its_first_fragment(self) -> None:
-        # #1612: tshark traced it at the fragment that completes the datagram
-        for version in (4, 6):
-            with self.subTest(ip=version):
-                frames = []
-                for ident, sport in ((0x100, 40000), (0x101, 40001)):
-                    frames.append(tcp_frame(version, wire.tcp(b'', seq=100, syn=True, sport=sport)))
-                    frames += fragments(version, wire.tcp(wire.payload_pattern(60), seq=101, ack=1,
-                                                          sport=sport), ident)
-                base = self.agree(frames)
-                self.assertEqual([index for _, index in base['flows']], [(1, 2), (6, 7)])
-
-    def test_a_stream_around_fragmented_segments_is_traced_as_by_the_default_engine(self) -> None:
-        # #1612: with its IP reassembly off, tshark sizes a fragmented segment
-        # from its first fragment, so its sequence analysis flags a lost
-        # segment on the later frames of the stream. tcp_traceflow reads none
-        # of those analysis fields.
-        for version in (4, 6):
-            with self.subTest(ip=version):
-                frames = [
-                    tcp_frame(version, wire.tcp(b'', seq=100, syn=True)),
-                    tcp_frame(version, wire.tcp(b'', seq=500, ack=101, syn=True, sport=9, dport=40000),
-                              reverse=True),
-                    tcp_frame(version, wire.tcp(b'', seq=101, ack=501)),
-                    *fragments(version, wire.tcp(wire.payload_pattern(100), seq=101, ack=501), 0x100,
-                               size=32),
-                    tcp_frame(version, wire.tcp(b'', seq=501, ack=201, sport=9, dport=40000), reverse=True),
-                    tcp_frame(version, wire.tcp(b'abc', seq=201, ack=501)),
-                    tcp_frame(version, wire.tcp(b'', seq=501, ack=204, sport=9, dport=40000), reverse=True),
-                    *fragments(version, wire.tcp(wire.payload_pattern(100, seed=2), seq=204, ack=501), 0x101,
-                               size=32),
-                    tcp_frame(version, wire.tcp(b'xyz', seq=304, ack=501)),
-                    tcp_frame(version, wire.tcp(b'', seq=307, ack=501, fin=True)),
-                ]
-                base = self.agree(frames)
-                # every frame but the later fragments, 5-7 and 12-14
-                self.assertEqual(sorted(base['trace']), [1, 2, 3, 4, 8, 9, 10, 11, 15, 16])
+                self.assertEqual(sorted(seen['trace']), [1, 2, 3, *range(9, 20), 40])
+                self.assertEqual([index for _, index in seen['flows']], [(1, 2, 3), (*range(9, 20), 40)])
+
+    def test_a_route_option_tshark_reports_no_destination_behind_is_skipped(self) -> None:
+        # #1671: route data that is no whole number of addresses, in the IPv4
+        # header carrying the segment, directly or inside 4in4; the extraction
+        # reads on past them
+        hop = bytes([10, 0, 0, 1])
+        segment = wire.tcp(b'data', seq=7, ack=1, sport=50000)
+        malformed = [route_option(kind, length, data) for kind, length, data in (
+            (0x83, 9, hop + b'\x01\x02'), (0x89, 9, hop + b'\x01\x02'), (0x83, 5, b'\x01\x02'),
+            (0x83, 13, hop + bytes(6)))]
+        frames = [tcp_frame(4, wire.tcp(b'', seq=100, syn=True)),
+                  *(wire.ethernet(with_options(wire.ipv4(segment, proto=6), option), 0x0800) for option in malformed),
+                  wire.ethernet(wire.ipv4(with_options(wire.ipv4(segment, proto=6), malformed[0]), proto=4), 0x0800),
+                  tcp_frame(4, wire.tcp(b'hello', seq=101, ack=1)),
+                  tcp_frame(4, wire.tcp(b'', seq=106, ack=1, fin=True))]
+        seen = read_trace(self, frames)
+        self.assertEqual(sorted(seen['trace']), [1, 7, 8])
+        self.assertEqual([index for _, index in seen['flows']], [(1, 7, 8)])
 
 
 if __name__ == '__main__':