fix(toolkit): dpkt reads SYN, FIN and RST from the flags octet whatever the NS bit (#1618)

pcapkit/toolkit/dpkt.py

@@ -742,6 +742,32 @@ def ipv6_reassembly(packet: 'Packet', timestamp: 'float', *,
     return None
 
 
+def _tcp_flags(tcp: 'TCP') -> 'tuple[bool, bool, bool]':
+    """Read the RST, SYN and FIN flags of a DPKT TCP segment.
+
+    Args:
+        tcp: DPKT TCP packet.
+
+    Returns:
+        Whether RST, SYN and FIN are set, in that order.
+
+    Note:
+        Each flag is masked out of :attr:`dpkt.tcp.TCP.flags` by its ``TH_*``
+        value, so it is read from the flags octet alone, as the default engine
+        reads it. `DPKT`_ 1.9.8 makes that field nine bits wide, its top bit
+        the NS bit (:rfc:`3540`): read by its position among the field's binary
+        digits, every flag was one bit low with NS set, a SYN reading as a FIN
+        (:issue:`1618`).
+
+    .. _DPKT: https://dpkt.readthedocs.io
+
+    """
+    from dpkt.tcp import TH_FIN, TH_RST, TH_SYN  # isort:skip
+
+    flags = tcp.flags
+    return bool(flags & TH_RST), bool(flags & TH_SYN), bool(flags & TH_FIN)
+
+
 def tcp_reassembly(packet: 'Packet', timestamp: 'float', *,
                    count: 'int' = -1) -> 'TCP_Packet | None':
     """Make data for TCP reassembly.
@@ -778,7 +804,7 @@ def tcp_reassembly(packet: 'Packet', timestamp: 'float', *,
     """
     ip, tcp = _tcp_carrier(packet)
     if ip is not None and tcp is not None:
-        flags = bin(tcp.flags)[2:].zfill(8)
+        rst, syn, fin = _tcp_flags(tcp)
         wire, _ = _layer2bytes(packet, tcp)
         raw_len = len(tcp.data)                                 # payload length, header excludes
 
@@ -792,9 +818,9 @@ def tcp_reassembly(packet: 'Packet', timestamp: 'float', *,
             num=count,                                          # original packet range number
             ack=tcp.ack,                                        # acknowledgement
             dsn=tcp.seq,                                        # data sequence number
-            rst=bool(int(flags[5])),                            # reset connection flag
-            syn=bool(int(flags[6])),                            # synchronise flag
-            fin=bool(int(flags[7])),                            # finish flag
+            rst=rst,                                            # reset connection flag
+            syn=syn,                                            # synchronise flag
+            fin=fin,                                            # finish flag
             header=wire[:tcp.off * 4],                          # raw bytes type header
             payload=bytearray(bytes(tcp.data)),                 # raw bytearray type payload
             first=tcp.seq,                                      # first sequence number of payload
@@ -844,16 +870,16 @@ def tcp_traceflow(packet: 'Packet', timestamp: 'float', *,
     """
     ip, tcp = _tcp_carrier(packet)
     if ip is not None and tcp is not None:
-        flags = bin(tcp.flags)[2:].zfill(8)
+        rst, syn, fin = _tcp_flags(tcp)
         wire, _ = _layer2bytes(packet, tcp)
 
         data = TF_TCP_Packet(  # type: ignore[type-var]
             protocol=data_link,                                         # data link type from global header
             index=count,                                                # frame number
             frame=packet2frame(packet, timestamp, data_link=data_link),  # extracted packet
-            syn=bool(int(flags[6])),                                    # TCP synchronise (SYN) flag
-            fin=bool(int(flags[7])),                                    # TCP finish (FIN) flag
-            rst=bool(int(flags[5])),                                    # TCP reset (RST) flag
+            syn=syn,                                                    # TCP synchronise (SYN) flag
+            fin=fin,                                                    # TCP finish (FIN) flag
+            rst=rst,                                                    # TCP reset (RST) flag
             src=ipaddress.ip_address(ip.src),                           # source IP
             dst=ipaddress.ip_address(ip.dst),                           # destination IP
             srcport=tcp.sport,                                          # TCP source port

tests/toolkit/test_dpkt_tcp_ns_flags_1618_unit.py

@@ -0,0 +1,106 @@
+# -*- coding: utf-8 -*-
+"""The dpkt engine reads RST, SYN and FIN as the default engine does, whatever the NS bit. C.f. #1618.
+
+`dpkt`_ 1.9.8 makes :attr:`dpkt.tcp.TCP.flags` nine bits wide, its top bit the
+NS bit (:rfc:`3540`). :func:`~pcapkit.toolkit.dpkt.tcp_reassembly` and
+:func:`~pcapkit.toolkit.dpkt.tcp_traceflow` read each flag by its position among
+that field's binary digits, padded to eight, so with NS set every flag was read
+one bit low: a SYN as a FIN, an RST as a SYN, a PSH as an RST. They now mask
+each flag out of the flags octet, as the default engine reads it.
+
+Every capture is built here, so the module is in the unit tier.
+
+.. _dpkt: https://dpkt.readthedocs.io
+
+"""
+from __future__ import annotations
+
+import importlib.util
+import struct
+import unittest
+
+from tests._support import reimport_once_per_class
+from tests.foundation import _roundtrip as wire
+from tests.toolkit.test_dpkt_scapy_ipv4_header_checks_1607_unit import read
+from tests.toolkit.test_tunnelled_tcp_flow_key_1581_unit import has
+
+RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper')
+HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS)
+
+#: The low nibble of the data offset octet: three reserved bits, then NS.
+NS = 0x01
+NIBBLES = range(16)
+
+
+def segment(number: int, nibble: int, octet: int) -> bytes:
+    """A TCP header of 20 octets on a port pair of its own, its offset octet's low nibble and flags octet as given."""
+    return struct.pack('!HHIIBBHHH', 1000 + number, 80, 100, 0, 0x50 | nibble, octet, 8192, 0, 0)
+
+
+def expected(octet: int) -> 'dict[str, bool]':
+    """RST, SYN and FIN as the flags octet ``octet`` sets them (:rfc:`9293#section-3.1`)."""
+    return {'rst': bool(octet & 0x04), 'syn': bool(octet & 0x02), 'fin': bool(octet & 0x01)}
+
+
+@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed')
+class DPKTTCPNSFlagsTests(unittest.TestCase):
+    """C.f. #1618."""
+
+    def setUp(self) -> None:
+        reimport_once_per_class(self)
+        if not has('dpkt'):
+            self.skipTest('dpkt not installed')
+
+    def test_adapters_read_each_flag_from_the_flags_octet(self) -> None:
+        # Every flags octet behind every NS and reserved bit, through both
+        # adapters; the header handed on keeps the captured NS bit.
+        import dpkt
+
+        from pcapkit.const.reg.linktype import LinkType
+        from pcapkit.toolkit import dpkt as toolkit
+
+        wrong = []
+        for nibble in NIBBLES:
+            for octet in range(256):
+                header = segment(0, nibble, octet)
+                frame = wire.ethernet(wire.ipv4(header, proto=6, df=True), 0x0800)
+                packet = dpkt.ethernet.Ethernet(frame)
+                toolkit.attach_buffer(packet, frame)
+                want = expected(octet)
+                for name, data in (
+                    ('tcp_reassembly', toolkit.tcp_reassembly(packet, 1.0)),
+                    ('tcp_traceflow', toolkit.tcp_traceflow(packet, 1.0, data_link=LinkType.ETHERNET)),
+                ):
+                    got = {flag: data[flag] for flag in want}
+                    if got != want or data['header'] != header:
+                        wrong.append((name, hex(nibble), hex(octet), got, want))
+        self.assertEqual(wrong, [])
+
+    def test_a_syn_with_ns_set_is_read_as_a_syn(self) -> None:
+        # The case #1618 reports: one SYN, NS set, read as a FIN.
+        seen = read(self, 'dpkt', [wire.ethernet(wire.ipv4(segment(1, NS, 0x02), proto=6, df=True), 0x0800)])
+        for aspect in ('tcp', 'trace'):
+            with self.subTest(aspect=aspect):
+                self.assertEqual({flag: seen[aspect][1][flag] for flag in ('rst', 'syn', 'fin')},
+                                 {'rst': False, 'syn': True, 'fin': False})
+
+    def test_extraction_agrees_with_the_default_engine(self) -> None:
+        # Every input to TCP reassembly and flow tracing, every field of it, and
+        # what comes out, for every flags octet with NS and the reserved bits
+        # clear and set.
+        cases = [(nibble, octet) for nibble in (0x0, NS, 0xE, 0xF) for octet in range(256)]
+        frames = [wire.ethernet(wire.ipv4(segment(number, nibble, octet), proto=6, df=True), 0x0800)
+                  for number, (nibble, octet) in enumerate(cases)]
+        base = read(self, 'default', frames)
+        self.assertEqual(sorted(base['tcp']), list(range(1, len(cases) + 1)))
+        self.assertEqual([{flag: base['tcp'][number][flag] for flag in ('rst', 'syn', 'fin')}
+                          for number in range(1, len(cases) + 1)],
+                         [expected(octet) for _, octet in cases])
+        seen = read(self, 'dpkt', frames)
+        for aspect in ('tcp', 'trace', 'ids', 'flows'):
+            with self.subTest(aspect=aspect):
+                self.assertEqual(seen[aspect], base[aspect])
+
+
+if __name__ == '__main__':
+    unittest.main()