fix(reassembly): report an unfragmented datagram whose BUFID is buffered, and leave IPv6 atomic fragments' buffers alone (#2270)

docs/source/pcapkit/foundation/reassembly/ip/index.rst

@@ -103,6 +103,9 @@ Departures from the procedure above:
   :rfc:`8200#section-4.5` for IPv6.
 * A buffer flushed by an unfragmented datagram reusing its ``BUFID`` is
   reported as an incomplete datagram rather than discarded.
+* An IPv6 atomic fragment, one with fragment offset zero and **M** clear,
+  flushes nothing: it is reassembled in isolation and the fragments buffered
+  under its ``BUFID`` are kept, per :rfc:`6946#section-4` (:issue:`2270`).
 * ``TDL`` starts at ``-1`` rather than ``0``, so that a datagram carrying no
   data can complete (:issue:`1359`).
 * Only the final fragment's partial last ``RCVBT`` block is set. A non-final

pcapkit/foundation/reassembly/ip.py

@@ -67,12 +67,21 @@ class IP(ReassemblyBase[Packet[_AT], Datagram[_AT], BufferID, Buffer[_AT]], Gene
        <pcapkit.foundation.reassembly.data.ip.Datagram.conflict>`
        (:issue:`477`). An incomplete buffer expires
        after ``timeout`` seconds of capture time, 60 by default for both IPv4
-       and IPv6.
+       and IPv6. A datagram with fragment offset zero and **MF** clear is
+       reported even when its BUFID is already buffered, rather than dropped
+       (:issue:`2270`).
 
     """
     if TYPE_CHECKING:
         protocol: 'Type[IP_Protocol]'
 
+    #: bool: Whether a datagram with fragment offset zero and **MF** clear is
+    #: reassembled apart from the buffer its BUFID names, rather than flushing
+    #: that buffer first. IPv4 flushes it (:rfc:`791#section-3.2`); IPv6
+    #: overrides this, since :rfc:`6946#section-4` keeps an atomic fragment's
+    #: buffer intact (:issue:`2270`).
+    __isolate_atomic__: 'bool' = False
+
     ##########################################################################
     # Methods.
     ##########################################################################
@@ -147,13 +156,22 @@ def reassembly(self, info: 'Packet[_AT]') -> 'None':
         # belong to.
         self._dtgram.extend(self.expire(TS))
 
-        # when non-fragmented (possibly discarded) packet received
+        # when non-fragmented (possibly discarded) packet received: it is
+        # reassembled on its own, in a buffer of its own (#2270)
+        isolated = False
         if not FO and not MF:
-            if BUFID in self._buffer:
+            if self.__isolate_atomic__:
+                # an IPv6 atomic fragment "MUST" be processed in isolation, and
+                # fragments queued under its BUFID "MUST NOT" be discarded
+                # (:rfc:`6946#section-4`), so the buffer is left alone
+                isolated = True
+            elif BUFID in self._buffer:
+                # :rfc:`791#section-3.2` flushes the buffer and then submits
+                # this datagram, so the flushed one is reported incomplete and
+                # this one goes on to a fresh buffer below
                 self._dtgram.extend(
                     self.submit(self._buffer.pop(BUFID), bufid=BUFID)
                 )
-                return
 
         # the header of the fragment at offset zero is the reassembled datagram's
         header = b'' if FO else self._rectify_header(info.header, BUFID[3])
@@ -162,8 +180,9 @@ def reassembly(self, info: 'Packet[_AT]') -> 'None':
         # block of the data buffer, its partial last block included -- 8191
         # would cover octets 0 to 65527 only, so a fragment reaching octet 65528,
         # e.g. at the largest Fragment Offset, would index past it (#1565)
-        if BUFID not in self._buffer:
-            self._buffer[BUFID] = Buffer(
+        buf: 'Buffer[_AT]'
+        if isolated or BUFID not in self._buffer:
+            buf = Buffer(
                 TDL=-1,                              # Total Data Length
                 RCVBT=bytearray(8192),              # Fragment Received Bit Table
                 index=[],                           # index record
@@ -172,12 +191,13 @@ def reassembly(self, info: 'Packet[_AT]') -> 'None':
                 timestamp=TS,                       # first-arriving fragment's clock reading
                 conflict=[],                        # conflicting octet ranges
             )
+            if not isolated:
+                self._buffer[BUFID] = buf
         else:
+            buf = self._buffer[BUFID]
             # put header into header buffer
-            if not FO:  # pylint: disable=else-if-used
-                self._buffer[BUFID].__update__(header=header)
-
-        buf = self._buffer[BUFID]
+            if not FO:
+                buf.__update__(header=header)
 
         # append packet index
         buf.index.append(info.num)
@@ -270,8 +290,12 @@ def reassembly(self, info: 'Packet[_AT]') -> 'None':
         if not MF:
             buf.__update__(TDL=length + FO)
 
+        # an isolated datagram has no buffer to wait in, so it is submitted
+        # whether or not it is complete
+        if isolated:
+            self._dtgram.extend(self.submit(buf, bufid=BUFID))
         # when datagram is reassembled in whole
-        if self._is_complete(buf):
+        elif self._is_complete(buf):
             self._dtgram.extend(
                 self.submit(self._buffer.pop(BUFID), bufid=BUFID, checked=True)
             )

pcapkit/foundation/reassembly/ipv6.py

@@ -92,7 +92,9 @@ class IPv6(IP):
        The reassembled ``header`` leaves out the Fragment header and carries
        its Next Header, per :rfc:`8200#section-4.5`, whichever engine fed it;
        1.4.x kept the Fragment header on the ``default`` and ``pcapng``
-       engines (:issue:`415`).
+       engines (:issue:`415`). An atomic fragment is reassembled in
+       isolation, without flushing a buffer that shares its BUFID
+       (:issue:`2270`).
 
     """
 
@@ -124,6 +126,13 @@ class IPv6(IP):
     #: which reports the condition in its log record instead.
     __timeout__ = 60.0
 
+    #: bool: An atomic fragment, one with fragment offset zero and **M**
+    #: clear, is reassembled on its own and leaves the buffer its BUFID names
+    #: untouched: :rfc:`6946#section-4` says it "MUST" be processed in
+    #: isolation and that the fragments queued under that BUFID "MUST NOT" be
+    #: discarded (:issue:`2270`).
+    __isolate_atomic__ = True
+
     ##########################################################################
     # Methods.
     ##########################################################################

tests/foundation/reassembly/test_ip.py

@@ -88,13 +88,19 @@ class TestIP(IP):
         reasm(self._packet(num=1, fo=0, mf=True, payload=b'abcdefgh'))
         reasm(self._packet(num=2, fo=0, mf=False, payload=b'ignored', header=b'fresh'))
 
-        datagram, = reasm.datagram
+        # the flushed datagram is reported incomplete, and the unfragmented one
+        # after it on its own (#2270)
+        datagram, single = reasm.datagram
         self.assertFalse(datagram.completed)
         self.assertEqual(datagram.index, (1,))
         self.assertEqual(datagram.header, b'ip-header')
         self.assertEqual(datagram.payload, (b'abcdefgh',))
         self.assertIsNone(datagram.packet)
-        self.assertEqual(len(callback_calls), 1)
+        self.assertTrue(single.completed)
+        self.assertEqual(single.index, (2,))
+        self.assertEqual(single.header, b'fresh')
+        self.assertEqual(single.payload, b'ignored')
+        self.assertEqual(len(callback_calls), 2)
 
         pending = TestIP()
         pending(self._packet(num=3, fo=8, mf=True, payload=b'ijkl', header=b'ignored'))

tests/foundation/reassembly/test_ip_unfragmented_bufid_collision_2270_unit.py

@@ -0,0 +1,81 @@
+# -*- coding: utf-8 -*-
+"""An unfragmented datagram whose BUFID is already buffered is reported.
+
+GitHub issue #2270. A datagram with fragment offset zero and **MF** clear whose
+BUFID matched a buffered one submitted that buffer and returned, so the datagram
+itself was never reported. IPv6 inherited this, so a colliding atomic fragment
+also flushed the datagram being reassembled, and its final fragment then opened
+a buffer of its own that never completed.
+
+:rfc:`791#section-3.2` flushes the buffer and then submits the datagram, so
+IPv4 reports both, the flushed one incomplete. :rfc:`6946#section-4` says an
+atomic fragment is processed in isolation and the queued fragments are not
+discarded, so IPv6 reports the atomic fragment alone and still completes the
+datagram it interrupted.
+
+Every case builds its fragments in memory and reads no capture. Classes are
+imported inside each test, after :func:`~tests._support.reimport_once_per_class`.
+
+"""
+
+from ipaddress import ip_address
+import unittest
+
+from tests._support import reimport_once_per_class
+
+
+class TestIPReassemblyUnfragmentedBufidCollision(unittest.TestCase):
+    """Pin that an unfragmented datagram sharing a buffered BUFID is reported."""
+
+    def setUp(self) -> None:
+        reimport_once_per_class(self)
+
+    def _packet(self, *, v6: bool, num: int, fo: int, mf: bool, payload: bytes):
+        """A fragment of the one BUFID every case shares."""
+        from pcapkit.const.reg.transtype import TransType
+        from pcapkit.foundation.reassembly.data.ip import Packet
+
+        if v6:
+            src, dst, ihl = ip_address('2001:db8::1'), ip_address('2001:db8::2'), 40
+        else:
+            src, dst, ihl = ip_address('192.0.2.1'), ip_address('192.0.2.2'), 20
+        return Packet((src, dst, 9, TransType.UDP), num, fo, ihl, mf, ihl + len(payload),
+                      b'H' * ihl, bytearray(payload), 1000.0 + num)
+
+    def _reassemble(self, cls, packets, strict: bool):
+        reasm = cls(strict=strict)
+        for packet in packets:
+            reasm(packet)
+        return reasm, [(datagram.index, datagram.completed, datagram.payload) for datagram in reasm.datagram]
+
+    def test_ipv4_flushes_the_buffer_and_reports_the_datagram(self) -> None:
+        from pcapkit.foundation.reassembly.data.data import Completion
+        from pcapkit.foundation.reassembly.ipv4 import IPv4
+
+        packets = [self._packet(v6=False, num=1, fo=0, mf=True, payload=b'A' * 16),
+                   self._packet(v6=False, num=2, fo=0, mf=False, payload=b'B' * 16)]
+        for strict in (True, False):
+            with self.subTest(strict=strict):
+                reasm, datagrams = self._reassemble(IPv4, packets, strict)
+                flushed = (b'A' * 16,) if strict else b'A' * 16
+                self.assertEqual(datagrams, [((1,), Completion.PARTIAL, flushed),
+                                             ((2,), Completion.COMPLETE, b'B' * 16)])
+                self.assertEqual(reasm._buffer, {})
+
+    def test_ipv6_atomic_fragment_leaves_the_buffer_alone(self) -> None:
+        from pcapkit.foundation.reassembly.data.data import Completion
+        from pcapkit.foundation.reassembly.ipv6 import IPv6
+
+        packets = [self._packet(v6=True, num=1, fo=0, mf=True, payload=b'A' * 16),
+                   self._packet(v6=True, num=2, fo=0, mf=False, payload=b'B' * 16),
+                   self._packet(v6=True, num=3, fo=16, mf=False, payload=b'C' * 16)]
+        for strict in (True, False):
+            with self.subTest(strict=strict):
+                reasm, datagrams = self._reassemble(IPv6, packets, strict)
+                self.assertEqual(datagrams, [((2,), Completion.COMPLETE, b'B' * 16),
+                                             ((1, 3), Completion.COMPLETE, b'A' * 16 + b'C' * 16)])
+                self.assertEqual(reasm._buffer, {})
+
+
+if __name__ == '__main__':
+    unittest.main()