diff --git a/docs/source/pcapkit/foundation/reassembly/tcp.rst b/docs/source/pcapkit/foundation/reassembly/tcp.rst index 1e49453429..65366281f3 100644 --- a/docs/source/pcapkit/foundation/reassembly/tcp.rst +++ b/docs/source/pcapkit/foundation/reassembly/tcp.rst @@ -207,6 +207,9 @@ Terminology | |--> 'header' : (bytes) initial TCP header | |--> 'payload' : (bytes) reassembled payload | |--> 'packet' : (Protocol) parsed reassembled payload + | |--> 'conflict' : (tuple) sequence ranges on which two segments disagreed + | | |--> (tuple) (first, last), absolute and inclusive + | | |--> ... |--> (Info) data | |--> 'completed' : (Completion) PARTIAL or TIMEOUT --> incomplete | |--> 'id' : (Info) original packet identifier @@ -225,8 +228,20 @@ Terminology | | |--> (bytes) payload fragment | | |--> ... | |--> 'packet' : (None) not implemented + | |--> 'conflict' : (tuple) sequence ranges on which two segments disagreed + | | |--> (tuple) (first, last), absolute and inclusive + | | |--> ... |--> (Info) data ... + ``completed`` and ``conflict`` are independent signals: a datagram can + be :attr:`~pcapkit.foundation.reassembly.data.data.Completion.COMPLETE` + and still carry a non-empty ``conflict`` -- the resolution of a + conflicting overlap is first-write-wins (:rfc:`9293#section-3.10`), so + it never leaves a hole, and a contested range that was later filled in + around does not stop the datagram from completing. ``conflict`` is + what lets a caller tell a clean stream from a contested one, now that + ``completed`` alone no longer can. + reasm.tcp.buffer Data structure for internal buffering when performing reassembly algorithms (:attr:`TCP._buffer `) @@ -256,12 +271,43 @@ Terminology | | |--> 'len' : (int) length of payload buffer | | |--> 'raw' : (bytearray) reassembled payload, | | holes set to b'\x00' + | | |--> 'gap' : (list) sequence ranges still + | | | zero-fill placeholder in 'raw' + | | | |--> (tuple) (first, last), + | | | absolute and + | | | inclusive + | | | |--> ... + | | |--> 'conflict' : (list) sequence ranges on which + | | | an arriving segment disagreed + | | | with bytes already in 'raw' + | | | |--> (tuple) (first, last), + | | | absolute and + | | | inclusive + | | | |--> ... | |--> (int) ACK ... | |--> ... | |--> 'timestamp' : (float) capture timestamp of the | first segment buffered |--> (tuple) BUFID ... + ``gap`` is deliberately **not** derived from ``hdl`` above. ``hdl`` is + shared by every ACK in this dict, while each ACK's own ``raw`` is + private to it, so a different ACK's segment closing a hole in ``hdl`` + says nothing about whether *this* ACK has received anything at the + same sequence numbers -- consulting ``hdl`` for that question + previously discarded a fragment's own real bytes whenever a different + ACK bucket under the same buffer ID happened to cover the same range + first. + + ``gap`` is kept in the same **absolute, inclusive sequence number** + convention as ``conflict`` above (and as ``hdl``'s own hole + descriptors), rather than as a per-octet marker aligned with ``raw``. + That is what lets it survive ``isn`` being revised downwards by a + reach-back segment: a per-octet marker aligned with ``raw`` has to be + re-prefixed in lockstep with every such revision, while an absolute + interval needs no shifting at all. It also means a fragment with no + holes carries an empty list instead of a ``raw``-sized marker. + .. note:: TCP reassembly has **no** timeout by default: no specification gives diff --git a/pcapkit/foundation/reassembly/data/tcp.py b/pcapkit/foundation/reassembly/data/tcp.py index 907c2a77f9..df58ee1134 100644 --- a/pcapkit/foundation/reassembly/data/tcp.py +++ b/pcapkit/foundation/reassembly/data/tcp.py @@ -107,6 +107,19 @@ class Datagram(DeferredPacket, Info, Generic[_AT]): #: Parsed TCP payload. Analysed on first read rather than at construction; #: a :class:`Deferred` may be passed in its place. packet: 'Optional[Protocol]' + #: Sequence ranges on which two segments disagreed, i.e. where an arriving + #: segment overlapped bytes already buffered but did not repeat them. + #: Each entry is ``(first, last)``, absolute TCP sequence numbers and both + #: **inclusive** -- the same convention as :attr:`Packet.first` and + #: :attr:`Packet.last`. Empty when the stream never saw a contested byte. + #: + #: Resolution keeps the already-buffered bytes and discards the + #: conflicting portion of whichever segment arrived later, per + #: :rfc:`9293#section-3.10` ("we reconstruct the segment to contain just + #: the new data"); this field is what lets a caller tell a clean stream + #: from a contested one now that :attr:`completed` no longer does, since a + #: contested range does not, on its own, leave a hole. + conflict: 'tuple[tuple[int, int], ...]' if TYPE_CHECKING: # NOTE: one signature rather than a pair of ``@overload``\\ s keyed on @@ -114,7 +127,7 @@ class Datagram(DeferredPacket, Info, Generic[_AT]): # :class:`~pcapkit.foundation.reassembly.data.ip.Datagram`, which applies # here identically: ``strict=False`` reports an incomplete payload buffer as # one contiguous ``bytes`` and analyses it. - def __init__(self, completed: 'Completion', id: 'DatagramID[_AT]', index: 'tuple[int, ...]', header: 'bytes', payload: 'bytes | tuple[bytes, ...]', packet: 'Optional[Protocol | Deferred]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + def __init__(self, completed: 'Completion', id: 'DatagramID[_AT]', index: 'tuple[int, ...]', header: 'bytes', payload: 'bytes | tuple[bytes, ...]', packet: 'Optional[Protocol | Deferred]', conflict: 'tuple[tuple[int, int], ...]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @info_final @@ -156,9 +169,51 @@ class Fragment(Info): len: 'int' #: Reassembled payload holes set to b'\x00'. raw: 'bytearray' + #: Sequence ranges, absolute and inclusive, still zero-fill placeholder in + #: :attr:`raw` rather than an actually-received byte *of this fragment*. + #: Only the two gap-creating sites in :meth:`TCP.reassembly + #: ` -- the forward + #: append and the reach-back prepend, the only places that ever splice a + #: ``bytearray(GAP)`` filler into :attr:`raw` -- add an entry; an overlap + #: merge only ever shrinks or removes one, filling it from the arriving + #: segment. + #: + #: This is deliberately **not** derived from + #: :attr:`Buffer.hdl `. + #: ``hdl`` is one list shared by every acknowledgement number under the + #: same buffer ID, so a segment landing in a *different* fragment can + #: close a hole in ``hdl`` that this fragment's own :attr:`raw` never + #: filled -- and consulting ``hdl`` to decide whether an overlapping + #: position here was "already received" then answers a question about + #: the wrong fragment. Tracking gaps on the fragment itself is what keeps + #: the merge in :meth:`TCP.reassembly + #: ` from discarding + #: this fragment's own real bytes because some *other* fragment happened + #: to have received something at the same absolute sequence numbers. + #: + #: Absolute sequence numbers rather than offsets into :attr:`raw` -- + #: :attr:`conflict` below uses the same convention -- for two reasons: + #: it is what :meth:`TCP.reassembly + #: ` already computes + #: (``GAP = PSN - (ISN + LEN)`` and its mirror), so this reuses an + #: existing concept rather than adding a second one; and it means a gap + #: entry never needs shifting when :attr:`isn` is revised downward by the + #: reach-back path, unlike an offset-based or a per-octet representation. + #: A typical fragment carries zero or a handful of entries, against a + #: per-octet marker the length of the whole payload -- the difference + #: that matters on a clean stream, where the per-octet form pays a + #: buffer-sized cost to record that nothing is missing at all. + gap: 'list[tuple[int, int]]' + #: Sequence ranges, absolute and inclusive, on which an arriving segment + #: disagreed with bytes already held in :attr:`raw`. Accumulated across + #: every merge into this fragment, in the order the conflicts were found; + #: carried onto :attr:`Datagram.conflict + #: ` verbatim + #: when the buffer is submitted. + conflict: 'list[tuple[int, int]]' if TYPE_CHECKING: - def __init__(self, ind: 'list[int]', isn: 'int', len: 'int', raw: 'bytearray') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + def __init__(self, ind: 'list[int]', isn: 'int', len: 'int', raw: 'bytearray', gap: 'list[tuple[int, int]]', conflict: 'list[tuple[int, int]]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @info_final diff --git a/pcapkit/foundation/reassembly/tcp.py b/pcapkit/foundation/reassembly/tcp.py index 35e818747b..fdd4df2a3f 100644 --- a/pcapkit/foundation/reassembly/tcp.py +++ b/pcapkit/foundation/reassembly/tcp.py @@ -159,6 +159,9 @@ def reassembly(self, info: 'Packet') -> 'None': isn=PSN, len=info.len, raw=info.payload, + # this segment's own payload is, by definition, all real + gap=[], + conflict=[], ), }, timestamp=TS, @@ -173,6 +176,8 @@ def reassembly(self, info: 'Packet') -> 'None': isn=PSN, len=info.len, raw=info.payload, + gap=[], + conflict=[], ) else: # put header into header buffer @@ -183,31 +188,11 @@ def reassembly(self, info: 'Packet') -> 'None': self._buffer[BUFID].ack[ACK].ind.append(info.num) # record fragment payload - ISN = self._buffer[BUFID].ack[ACK].isn # Initial Sequence Number - RAW = self._buffer[BUFID].ack[ACK].raw # Raw Payload Data - if PSN >= ISN: # if fragment goes after existing payload - LEN = self._buffer[BUFID].ack[ACK].len - GAP = PSN - (ISN + LEN) # gap length between payloads - if GAP >= 0: # if fragment goes after existing payload - RAW += bytearray(GAP) + info.payload - else: # if fragment partially overlaps existing payload - RAW[PSN - ISN:PSN - ISN + info.len] = info.payload - else: # if fragment exceeds existing payload - LEN = info.len - GAP = ISN - (PSN + LEN) # gap length between payloads - self._buffer[BUFID].ack[ACK].__update__( - isn=PSN, - ) - if GAP >= 0: # if fragment exceeds existing payload - RAW = info.payload + bytearray(GAP) + RAW - else: # if fragment partially overlaps existing payload - RAW = info.payload + RAW[-GAP:] - #self._buffer[BUFID].ack[ACK].raw = RAW # update payload datagram - #self._buffer[BUFID].ack[ACK].len = len(RAW) # update payload length - self._buffer[BUFID].ack[ACK].__update__( - raw=RAW, # update payload datagram - len=len(RAW), # update payload length - ) + fragment = self._buffer[BUFID].ack[ACK] + if PSN >= fragment.isn: # if fragment goes after existing payload + self._reassemble_append(info, fragment, PSN) + else: # if fragment exceeds existing payload + self._reassemble_prepend(info, fragment, PSN) # update hole descriptor list # @@ -218,28 +203,7 @@ def reassembly(self, info: 'Packet') -> 'None': # into two adjacent holes covering the very same octets, growing # the list without bound on a long-lived connection. if info.len > 0: - HDL = self._buffer[BUFID].hdl # HDL alias - for (index, hole) in enumerate(HDL): # step one - if info.first > hole.last: # step two - continue - if info.last < hole.first: # step three - continue - del HDL[index] # step four - if info.first > hole.first: # step five - new_hole = HoleDescriptor( - first=hole.first, - last=info.first - 1, - ) - HDL.insert(index, new_hole) - index += 1 - if info.last < hole.last and not FIN and not RST: # step six - new_hole = HoleDescriptor( - first=info.last + 1, - last=hole.last - ) - HDL.insert(index, new_hole) - break # step seven - #self._buffer[BUFID].hdl = HDL # update HDL + self._update_hole_descriptors(info, BUFID, FIN, RST) # when FIN/RST is set, submit buffer of this session if FIN or RST: @@ -247,6 +211,244 @@ def reassembly(self, info: 'Packet') -> 'None': self.submit(self._buffer.pop(BUFID), bufid=BUFID) ) + def _reassemble_append(self, info: 'Packet', fragment: 'Fragment', PSN: 'int') -> 'None': + """Merge a segment that starts at or after the buffered payload's end. + + Covers both an ordinary (or zero-length) forward gap and a tail-side + overlap. Mutates ``fragment`` in place: its ``gap`` list, its + ``conflict`` list, and finally its ``raw``/``len`` via + :meth:`~pcapkit.foundation.reassembly.data.data.Info.__update__`. + + Arguments: + info: :term:`info ` dict of the arriving segment + fragment: this ACK bucket's own + :class:`~pcapkit.foundation.reassembly.data.tcp.Fragment` + PSN: payload sequence number of the arriving segment + + """ + ISN = fragment.isn # Initial Sequence Number + RAW = fragment.raw # Raw Payload Data + GAPS = fragment.gap # this fragment's own gap list + LEN = fragment.len + GAP = PSN - (ISN + LEN) # gap length between payloads + if GAP >= 0: # if fragment goes after existing payload + if GAP > 0: + GAPS.append((ISN + LEN, PSN - 1)) + RAW += bytearray(GAP) + info.payload + else: + # Fragment partially overlaps existing payload. Per + # :rfc:`9293#section-3.10` ("we reconstruct the segment + # to contain just the new data"), an already-*received* + # byte wins over a conflicting arriving one; only a + # position this *fragment* has not received yet (per + # its own ``gap`` list, not the buffer-wide ``HDL`` -- + # see + # :attr:`~pcapkit.foundation.reassembly.data.tcp.Fragment.gap`) + # has nothing to disagree with, so the arriving byte + # is simply accepted there -- an ordinary gap fill, + # not a conflict. + OFFSET = PSN - ISN # index into RAW where the overlap begins + OVERLAP = min(info.len, LEN - OFFSET) # length of the overlapping range + merged, conflicts = self._merge_overlap( + GAPS, + bytes(RAW[OFFSET:OFFSET + OVERLAP]), bytes(info.payload[:OVERLAP]), + PSN, + ) + RAW[OFFSET:OFFSET + OVERLAP] = merged + fragment.conflict.extend(conflicts) + if info.len > OVERLAP: # fragment reaches past the buffered end + RAW += info.payload[OVERLAP:] + fragment.__update__( + raw=RAW, # update payload datagram + len=len(RAW), # update payload length + ) + + def _reassemble_prepend(self, info: 'Packet', fragment: 'Fragment', PSN: 'int') -> 'None': + """Merge a segment that starts before the buffered payload's own ``isn``. + + Revises this fragment's ``isn`` down to ``PSN``, then covers both an + ordinary (or zero-length) reach-back gap and a head-side overlap -- + which may also extend a new tail past the buffered end in the same + call. Mutates ``fragment`` in place: its ``isn``, its ``gap`` list, + its ``conflict`` list, and finally its ``raw``/``len`` via + :meth:`~pcapkit.foundation.reassembly.data.data.Info.__update__`. + + Arguments: + info: :term:`info ` dict of the arriving segment + fragment: this ACK bucket's own + :class:`~pcapkit.foundation.reassembly.data.tcp.Fragment` + PSN: payload sequence number of the arriving segment + + """ + ISN = fragment.isn # Initial Sequence Number, before revision + RAW = fragment.raw # Raw Payload Data + GAPS = fragment.gap # this fragment's own gap list + LEN = info.len + GAP = ISN - (PSN + LEN) # gap length between payloads + fragment.__update__( + isn=PSN, + ) + if GAP >= 0: # if fragment exceeds existing payload + if GAP > 0: + GAPS.append((PSN + LEN, ISN - 1)) + RAW = info.payload + bytearray(GAP) + RAW + else: + # Mirrored reach-back case: the fragment starts before + # ``ISN`` and its tail overlaps the start of the + # already-buffered payload. Same resolution -- keep + # already-received bytes, fill any gap among them from + # the arriving segment, and prepend the genuinely new + # head. The new head-prepend below is *not* mutually + # exclusive with the rare case of also appending a new + # tail past the buffered end (full engulfment plus + # extension) -- both can happen in the same call, since + # they come from independent ends of the fragment. What + # *is* mutually exclusive is which one of the old tail + # (``RAW[OVERLAP:]``) or a genuinely new one + # (``info.payload[OFFSET + OVERLAP:]``) is non-empty -- + # never both, since ``OVERLAP`` is capped at whichever + # of the two is shorter. + # + # ``GAPS`` is not touched here at all: it is kept in + # absolute sequence numbers, so revising ``isn`` + # downward above does not require shifting or + # re-prefixing a single entry in it. + OFFSET = ISN - PSN # index into info.payload where the overlap begins + OVERLAP = min(len(RAW), LEN - OFFSET) # length of the overlapping range + merged, conflicts = self._merge_overlap( + GAPS, + bytes(RAW[:OVERLAP]), bytes(info.payload[OFFSET:OFFSET + OVERLAP]), + ISN, + ) + RAW[:OVERLAP] = merged + fragment.conflict.extend(conflicts) + RAW = info.payload[:OFFSET] + RAW + info.payload[OFFSET + OVERLAP:] + fragment.__update__( + raw=RAW, # update payload datagram + len=len(RAW), # update payload length + ) + + def _update_hole_descriptors(self, info: 'Packet', BUFID: 'BufferID', + FIN: 'bool', RST: 'bool') -> 'None': + """Update the buffer-wide hole descriptor list per :rfc:`815`. + + Called only for a segment that carries payload (``info.len > 0``); + a bare acknowledgement, SYN, FIN or RST fills no hole, and running + one through this would split whichever hole contains it into two + adjacent holes covering the very same octets, growing the list + without bound on a long-lived connection. + + Arguments: + info: :term:`info ` dict of the arriving segment + BUFID: buffer identifier of the session this fragment belongs to + FIN: finish flag (termination) of the arriving segment + RST: reset connection flag (termination) of the arriving segment + + """ + HDL = self._buffer[BUFID].hdl # HDL alias + for (index, hole) in enumerate(HDL): # step one + if info.first > hole.last: # step two + continue + if info.last < hole.first: # step three + continue + del HDL[index] # step four + if info.first > hole.first: # step five + new_hole = HoleDescriptor( + first=hole.first, + last=info.first - 1, + ) + HDL.insert(index, new_hole) + index += 1 + if info.last < hole.last and not FIN and not RST: # step six + new_hole = HoleDescriptor( + first=info.last + 1, + last=hole.last + ) + HDL.insert(index, new_hole) + break # step seven + #self._buffer[BUFID].hdl = HDL # update HDL + + @staticmethod + def _merge_overlap(gap: 'list[tuple[int, int]]', old: 'bytes', new: 'bytes', + start: 'int') -> 'tuple[bytes, list[tuple[int, int]]]': + """Merge an arriving segment into an overlapping range of buffered bytes. + + Arguments: + gap: this *fragment's own* + :attr:`~pcapkit.foundation.reassembly.data.tcp.Fragment.gap` + list -- absolute, inclusive sequence ranges still zero-fill + placeholder in ``old``. **Mutated in place**: whichever + portion of a gap entry falls inside ``[start, start + + len(old) - 1]`` is filled from ``new`` and removed (or + trimmed, if only part of the entry falls inside the range). + Deliberately **not** derived from + :attr:`~pcapkit.foundation.reassembly.data.tcp.Buffer.hdl`, + which is shared across every acknowledgement number under the + same buffer ID: a *different* fragment closing a hole there + says nothing about what *this* fragment has received, and + using it here previously discarded this fragment's own real + bytes whenever another fragment happened to cover the same + absolute sequence numbers first. + old: already-buffered bytes of this fragment over the range. + new: the arriving segment's bytes over the same range. + start: absolute sequence number of ``old[0]``/``new[0]``, which + cover the same range by construction -- see the two call + sites in :meth:`reassembly`. + + Returns: + The bytes to keep for the range, and any ``(first, last)`` + absolute sequence ranges -- inclusive, same convention as + :attr:`gap` and :attr:`~pcapkit.foundation.reassembly.data.tcp.Fragment.conflict` + -- where already-*received* bytes disagreed with the arriving + segment. + + A position still covered by a ``gap`` entry has no already-received + byte to disagree with, so the arriving segment's byte is simply + accepted there and that slice of the gap is closed: an ordinary gap + fill, not a conflict. Only a position outside every gap -- already + received -- can conflict, per :rfc:`9293#section-3.10`: the + already-received byte wins and the arriving one is discarded, but the + disagreement itself is what this records for :attr:`Fragment.conflict + `. + + """ + length = len(old) + end = start + length - 1 # inclusive + merged = bytearray(old) + + # A position is a hole exactly while some gap entry covers it; build + # that once, per this call, from the compact interval list rather + # than keeping a per-octet marker between calls. Any gap entry (or + # remaining slice of one) outside ``[start, end]`` is untouched. + received = bytearray(b'\x01' * length) # scratch for this call only + still_gap = [] # type: list[tuple[int, int]] + for (first, last) in gap: + lo, hi = max(first, start), min(last, end) + if lo > hi: # this entry misses the range entirely + still_gap.append((first, last)) + continue + rel_lo, rel_hi = lo - start, hi - start # inclusive + merged[rel_lo:rel_hi + 1] = new[rel_lo:rel_hi + 1] + received[rel_lo:rel_hi + 1] = bytes(rel_hi - rel_lo + 1) + if first < lo: # a leading slice of the entry survives + still_gap.append((first, lo - 1)) + if last > hi: # a trailing slice of the entry survives + still_gap.append((hi + 1, last)) + gap[:] = still_gap + + conflicts = [] # type: list[tuple[int, int]] + index = 0 + while index < length: + if not received[index] or old[index] == new[index]: + index += 1 + continue + stop = index + while stop < length and received[stop] and old[stop] != new[stop]: + stop += 1 + conflicts.append((start + index, start + stop - 1)) + index = stop + return bytes(merged), conflicts + def submit(self, buf: 'Buffer', *, bufid: 'BufferID', # type: ignore[override] # pylint: disable=arguments-differ timeout: 'bool' = False) -> 'list[Datagram]': """Submit reassembled payload. @@ -316,6 +518,7 @@ def submit(self, buf: 'Buffer', *, bufid: 'BufferID', # type: ignore[override] header=buf.hdr, payload=tuple(data), packet=None, + conflict=tuple(buffer.conflict), ) datagram.append(packet) @@ -342,6 +545,7 @@ def submit(self, buf: 'Buffer', *, bufid: 'BufferID', # type: ignore[override] header=buf.hdr, payload=bytes(payload), packet=Deferred(self.protocol.analyze, (bufid[1], bufid[3]), bytes(payload)), + conflict=tuple(buffer.conflict), ) datagram.append(packet) diff --git a/tests/foundation/reassembly/data/test_models.py b/tests/foundation/reassembly/data/test_models.py index 33fed7da14..c2b0fa5fa6 100644 --- a/tests/foundation/reassembly/data/test_models.py +++ b/tests/foundation/reassembly/data/test_models.py @@ -104,14 +104,15 @@ def test_tcp_data_models_and_package_aliases(self) -> None: datagram_id = DatagramID((src, 12345), (dst, 443), 200) datagram = Datagram(Completion.COMPLETE, datagram_id, (3,), b'tcp-header', b'hello', - {'parsed': True}) + {'parsed': True}, ()) self.assertIsInstance(datagram.id, TCP_DatagramID) self.assertIsInstance(datagram, TCP_Datagram) self.assertTrue(datagram.completed) self.assertIs(datagram.completed, Completion.COMPLETE) + self.assertEqual(datagram.conflict, ()) hole = HoleDescriptor(5, 10) - fragment = Fragment([3], 100, 5, bytearray(b'hello')) + fragment = Fragment([3], 100, 5, bytearray(b'hello'), [], []) buffer = Buffer([hole], b'tcp-header', {200: fragment}, 1000.0) self.assertIsInstance(hole, TCP_HoleDescriptor) self.assertIsInstance(fragment, TCP_Fragment) diff --git a/tests/foundation/reassembly/test_tcp.py b/tests/foundation/reassembly/test_tcp.py index 1c5e4d77c9..60e0f35d91 100644 --- a/tests/foundation/reassembly/test_tcp.py +++ b/tests/foundation/reassembly/test_tcp.py @@ -118,7 +118,7 @@ class TestTCP(TCP): [HoleDescriptor(0, 4), HoleDescriptor(20, 30), HoleDescriptor(40, sys.maxsize)], b'', { - 500: Fragment([1], 10, 10, bytearray(b'0123456789')), + 500: Fragment([1], 10, 10, bytearray(b'0123456789'), [], []), }, 1000.0, ) @@ -126,14 +126,25 @@ class TestTCP(TCP): reasm(self._packet(num=2, dsn=25, payload=b'after-gap', first=10, last=18)) self.assertEqual(reasm._buffer[bufid].ack[500].raw, bytearray(b'0123456789\x00\x00\x00\x00\x00after-gap')) + # ``OVERLAP`` (abs 15-21) disagrees with the already-received ``56789`` + # (abs 15-19) -- kept, per first-write-wins, and recorded as a + # conflict -- but abs 20-21 is still a hole per the seeded HDL, not + # already-received data, so the arriving segment's bytes there + # (``AP``) are an ordinary gap fill, not a conflict. reasm(self._packet(num=3, dsn=15, payload=b'OVERLAP', first=15, last=21)) self.assertEqual(reasm._buffer[bufid].ack[500].raw, - bytearray(b'01234OVERLAP\x00\x00\x00after-gap')) + bytearray(b'0123456789AP\x00\x00\x00after-gap')) + self.assertEqual(reasm._buffer[bufid].ack[500].conflict, [(15, 19)]) + # The reach-back branch: abs 10-12 (``012``) is already-received data + # and disagrees with the arriving ``klm`` -- kept, and a second + # conflict recorded -- while abs 0-9 is genuinely new (before the + # existing ISN) and merges in untouched. reasm(self._packet(num=4, dsn=0, payload=b'abcdefghijklm', first=22, last=24)) self.assertEqual(reasm._buffer[bufid].ack[500].isn, 0) self.assertEqual(reasm._buffer[bufid].ack[500].raw, - bytearray(b'abcdefghijklm34OVERLAP\x00\x00\x00after-gap')) + bytearray(b'abcdefghij0123456789AP\x00\x00\x00after-gap')) + self.assertEqual(reasm._buffer[bufid].ack[500].conflict, [(15, 19), (10, 12)]) self.assertEqual([(hole.first, hole.last) for hole in reasm._buffer[bufid].hdl], [(0, 4), (25, 30), (40, sys.maxsize)]) @@ -144,12 +155,13 @@ class TestTCP(TCP): before_gap._buffer[bufid] = Buffer( [HoleDescriptor(50, sys.maxsize)], b'', - {500: Fragment([1], 10, 5, bytearray(b'world'))}, + {500: Fragment([1], 10, 5, bytearray(b'world'), [], [])}, 1000.0, ) before_gap(self._packet(num=2, dsn=0, payload=b'hello', first=40, last=44)) self.assertEqual(before_gap._buffer[bufid].ack[500].raw, bytearray(b'hello\x00\x00\x00\x00\x00world')) + self.assertEqual(before_gap._buffer[bufid].ack[500].conflict, []) def test_submit_incomplete_strict_complete_strict_false_and_empty_buffers(self) -> None: from pcapkit.foundation.reassembly.data.data import Completion @@ -173,7 +185,7 @@ class TestTCP(TCP): Buffer( [HoleDescriptor(2, 3), HoleDescriptor(7, 8), HoleDescriptor(99, 100)], b'tcp-header', - {500: Fragment([1, 2], 0, 10, bytearray(b'abcdefghij'))}, + {500: Fragment([1, 2], 0, 10, bytearray(b'abcdefghij'), [], [])}, 1000.0, ), bufid=bufid, @@ -185,14 +197,15 @@ class TestTCP(TCP): self.assertIs(datagram.completed, Completion.PARTIAL) self.assertEqual(datagram.payload, (bytearray(b'ab'), bytearray(b'efg'), bytearray(b'j'))) self.assertIsNone(datagram.packet) + self.assertEqual(datagram.conflict, ()) mixed = strict.submit( Buffer( [HoleDescriptor(0, 0), HoleDescriptor(4, 5), HoleDescriptor(7, 7)], b'tcp-header', { - 500: Fragment([], 0, 0, bytearray()), - 501: Fragment([9], 0, 9, bytearray(b'abcdefghi')), + 500: Fragment([], 0, 0, bytearray(), [], []), + 501: Fragment([9], 0, 9, bytearray(b'abcdefghi'), [], [(2, 3)]), }, 1000.0, ), @@ -200,6 +213,9 @@ class TestTCP(TCP): ) self.assertEqual(len(mixed), 1) self.assertEqual(mixed[0].payload, (bytearray(b'bcd'), bytearray(b'g'), b'i')) + # ``conflict`` passes through from the fragment untouched -- ``submit`` + # only reads it, the merge logic in ``reassembly`` is what populates it + self.assertEqual(mixed[0].conflict, ((2, 3),)) # ``strict=False`` reports the payload buffer as one contiguous blob with # its holes zero-filled -- which is what @@ -212,7 +228,7 @@ class TestTCP(TCP): Buffer( [HoleDescriptor(2, 3), HoleDescriptor(7, 8), HoleDescriptor(99, 100)], b'tcp-header', - {500: Fragment([3], 0, 3, bytearray(b'abc'))}, + {500: Fragment([3], 0, 3, bytearray(b'abc'), [], [])}, 1000.0, ), bufid=bufid, @@ -221,9 +237,10 @@ class TestTCP(TCP): self.assertIs(completed[0].completed, Completion.PARTIAL) self.assertEqual(completed[0].payload, bytearray(b'abc')) self.assertEqual(completed[0].packet, b'abc') + self.assertEqual(completed[0].conflict, ()) self.assertEqual(Analyzer.calls[-1], ((12345, 443), b'abc')) - self.assertEqual(loose.submit(Buffer([], b'', {500: Fragment([], 0, 0, bytearray())}, + self.assertEqual(loose.submit(Buffer([], b'', {500: Fragment([], 0, 0, bytearray(), [], [])}, 1000.0), bufid=bufid), []) @@ -552,5 +569,334 @@ def test_payload_free_segments_leave_the_hole_list_alone(self) -> None: # workflow runs without the generated fixtures. +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class TCPReassemblyConflictTests(unittest.TestCase): + """Regression tests for GitHub issue #443. + + Two segments claiming the same sequence range but carrying *different* + bytes used to resolve last-write-wins, silently, with ``completed`` still + reporting the datagram whole. Per :rfc:`9293#section-3.10` ("we + reconstruct the segment to contain just the new data") the resolution is + first-write-wins instead: the already-buffered bytes are kept, the + conflicting portion of whichever segment arrived later is discarded, and + the sequence range on which they disagreed is recorded on + :attr:`~pcapkit.foundation.reassembly.data.tcp.Datagram.conflict` -- + additive, so ``completed`` is untouched and existing callers are + unaffected. + + Every test drives the public API only -- build :class:`Packet`, feed + :class:`TCP`, read back the :class:`Datagram` from + :meth:`~pcapkit.foundation.reassembly.reassembly.Reassembly.fetch` -- + the same shape as the reproduction in the issue itself, which this class + starts from verbatim. + + """ + + def setUp(self) -> None: + purge_modules(['pcapkit']) + + def _bufid(self): + return (ip_address('192.0.2.1'), 12345, ip_address('198.51.100.2'), 443) + + def _packet(self, *, num: int, dsn: int, payload: bytes = b'', ack: int = 1000, + timestamp: float = 0.0): + """One segment, described exactly as the issue's own reproduction does.""" + from pcapkit.foundation.reassembly.data.tcp import Packet + + return Packet(self._bufid(), dsn, ack, num, False, False, False, len(payload), + dsn, dsn + len(payload) - 1, b'hdr', bytearray(payload), timestamp) + + def _run(self, *packets): + from pcapkit.foundation.reassembly.tcp import TCP + + reasm = TCP() + for packet in packets: + reasm(packet) + datagram, = reasm.fetch() + return datagram + + #: Base sequence number for every scenario below -- deliberately not + #: zero, and not the same as the coordinate-system tests' own ``ISN``, + #: so a conflict range that happened to be computed as an offset rather + #: than an absolute sequence number would not go unnoticed by accident. + BASE = 0x7EED0000 + 100 + + def test_identical_retransmission_stays_uncontested_and_unchanged(self) -> None: + """A conforming retransmission -- same bytes -- reports no conflict at all.""" + base = self.BASE + datagram = self._run( + self._packet(num=1, dsn=base, payload=b'AAAAAAAA'), + self._packet(num=2, dsn=base, payload=b'AAAAAAAA'), + self._packet(num=3, dsn=base + 8, payload=b'CCCC'), + ) + self.assertTrue(datagram.completed) + self.assertEqual(datagram.payload, b'AAAAAAAACCCC') + self.assertEqual(datagram.conflict, ()) + + def test_conflicting_full_overlap_keeps_the_first_segment(self) -> None: + """The issue's own reproduction: first-write-wins, and the range is recorded. + + Before the fix this returned ``completed=True`` and + ``payload=b'BBBBBBBBCCCC'`` -- the later, conflicting segment silently + won. The RFC-conformant answer keeps the first segment's bytes. + + """ + base = self.BASE + datagram = self._run( + self._packet(num=1, dsn=base, payload=b'AAAAAAAA'), + self._packet(num=2, dsn=base, payload=b'BBBBBBBB'), + self._packet(num=3, dsn=base + 8, payload=b'CCCC'), + ) + self.assertTrue(datagram.completed) + self.assertEqual(datagram.payload, b'AAAAAAAACCCC') + self.assertEqual(datagram.conflict, ((base, base + 7),)) + self.assertEqual(datagram.index, (1, 2, 3)) + + def test_conflicting_partial_overlap_on_the_tail_side(self) -> None: + """A segment that overlaps the buffered tail and then extends past it. + + Exercises the non-reach-back overlap branch + (:meth:`~pcapkit.foundation.reassembly.tcp.TCP.reassembly`, the branch + guarded by ``PSN >= ISN``) when the arriving segment's end lies past + the already-buffered end: the overlapping half is contested and + resolved first-write-wins, the non-overlapping half is genuinely new + and is merged in untouched. + + """ + base = self.BASE + datagram = self._run( + self._packet(num=1, dsn=base, payload=b'AAAAAAAA'), # base .. base+7 + self._packet(num=2, dsn=base + 4, payload=b'XXXXXXXX'), # base+4 .. base+11 + ) + self.assertTrue(datagram.completed) + self.assertEqual(datagram.payload, b'AAAAAAAAXXXX') + self.assertEqual(datagram.conflict, ((base + 4, base + 7),)) + + def test_conflicting_partial_overlap_on_the_head_side_reach_back(self) -> None: + """The mirrored reach-back branch, at the line the issue calls out at :204. + + A segment arriving with a *lower* sequence number than the buffer's + current ISN, whose tail overlaps the buffer's head: the overlapping + half is contested and resolved first-write-wins, and the segment's + own leading bytes -- which lie before the existing ISN -- are + genuinely new and are prepended untouched. + + """ + base = self.BASE + datagram = self._run( + self._packet(num=1, dsn=base + 4, payload=b'BBBBBBBB'), # base+4 .. base+11 + self._packet(num=2, dsn=base, payload=b'YYYYYYYY'), # base .. base+7 + ) + self.assertTrue(datagram.completed) + self.assertEqual(datagram.payload, b'YYYYBBBBBBBB') + self.assertEqual(datagram.conflict, ((base + 4, base + 7),)) + + def test_three_way_conflict_keeps_the_first_and_records_every_disagreement(self) -> None: + """Three segments claiming the same range, each disagreeing with the buffer. + + First-write-wins is decided once, by the first segment to arrive; + every later arrival that disagrees with what is already buffered is + its own conflict, not just the first one. + + """ + base = self.BASE + datagram = self._run( + self._packet(num=1, dsn=base, payload=b'AAAAAAAA'), + self._packet(num=2, dsn=base, payload=b'BBBBBBBB'), + self._packet(num=3, dsn=base, payload=b'CCCCCCCC'), + self._packet(num=4, dsn=base + 8, payload=b'DDDD'), + ) + self.assertTrue(datagram.completed) + self.assertEqual(datagram.payload, b'AAAAAAAADDDD') + self.assertEqual(datagram.conflict, ((base, base + 7), (base, base + 7))) + + def test_conflict_persists_once_a_later_segment_completes_the_datagram(self) -> None: + """A conflict recorded while the stream is still partial survives to completion. + + The gap between the two original segments is a real hole -- unlike + the overlap in the other tests here, filling it is an ordinary gap + fill, not a conflict -- and once it closes the datagram reports + :attr:`~pcapkit.foundation.reassembly.data.data.Completion.COMPLETE`, + per the decision to leave ``completed`` alone: the conflict recorded + earlier is still there, on the completed datagram, rather than being + dropped or blocking completion. + + """ + base = self.BASE + from pcapkit.foundation.reassembly.data.data import Completion + from pcapkit.foundation.reassembly.tcp import TCP + + reasm = TCP() + reasm(self._packet(num=1, dsn=base, payload=b'AAAAAAAA')) # base .. base+7 + reasm(self._packet(num=2, dsn=base + 20, payload=b'CCCCCCCC')) # base+20 .. base+27 + + partial, = reasm.fetch() + self.assertIs(partial.completed, Completion.PARTIAL) + self.assertEqual(partial.payload, (b'AAAAAAAA', b'CCCCCCCC')) + self.assertEqual(partial.conflict, ()) + + # conflicts with segment 1 over base..base+7 -- already-received data + reasm(self._packet(num=3, dsn=base, payload=b'BBBBBBBB')) + # fills the base+8..base+19 gap exactly -- an ordinary gap fill, not a conflict + reasm(self._packet(num=4, dsn=base + 8, payload=b'D' * 12)) + + complete, = reasm.fetch() + self.assertIs(complete.completed, Completion.COMPLETE) + self.assertEqual(complete.payload, b'AAAAAAAA' + b'D' * 12 + b'CCCCCCCC') + self.assertEqual(complete.conflict, ((base, base + 7),)) + + def test_one_ack_buckets_hole_closing_does_not_leak_receipt_into_another(self) -> None: + """A hole closed in one ACK bucket must not look received in a different one. + + :attr:`~pcapkit.foundation.reassembly.data.tcp.Buffer.hdl` is one hole + descriptor list shared by every ACK bucket under the same BUFID, while + each bucket's own :attr:`~pcapkit.foundation.reassembly.data.tcp.Fragment.raw` + is private to that bucket. Bucket 2000 here fills the *shared* hole at + base+4..base+9 with its own, entirely unrelated data; that must not + make bucket 1000's later, real segment for the very same absolute + range look like a conflicting retransmission of something bucket 1000 + already had -- it is bucket 1000's *first* receipt there, and it must + survive intact with no conflict recorded. + + """ + base = self.BASE + from pcapkit.foundation.reassembly.tcp import TCP + + reasm = TCP() + reasm(self._packet(num=1, dsn=base, payload=b'AAAA', ack=1000)) # bucket 1000: base..base+3 + reasm(self._packet(num=2, dsn=base + 10, payload=b'DDDD', ack=1000)) # bucket 1000: base+10..base+13, + # gap base+4..base+9 in the shared hdl + reasm(self._packet(num=3, dsn=base + 4, payload=b'X' * 6, ack=2000)) # bucket 2000 closes the SHARED hole + reasm(self._packet(num=4, dsn=base + 4, payload=b'C' * 6, ack=1000)) # bucket 1000's own first receipt there + reasm(self._packet(num=5, dsn=base + 14, payload=b'EEEE', ack=1000)) + + datagrams = {d.id.ack: d for d in reasm.fetch()} + self.assertTrue(datagrams[1000].completed) + self.assertEqual(datagrams[1000].payload, b'AAAA' + b'C' * 6 + b'DDDDEEEE') + self.assertEqual(datagrams[1000].conflict, ()) + self.assertTrue(datagrams[2000].completed) + self.assertEqual(datagrams[2000].payload, b'X' * 6) + self.assertEqual(datagrams[2000].conflict, ()) + + def test_a_genuine_gap_fill_through_the_overlap_merge_is_never_a_conflict(self) -> None: + """A hole filled by the overlap-merge path is a gap fill, not a conflict. + + The arriving segment here straddles a real hole *and* touches + already-received bytes on both sides of it in the same merge call -- + the case :meth:`~pcapkit.foundation.reassembly.tcp.TCP._merge_overlap` + has to get right on a single call, not just across separate ones. The + already-received edges carry bytes identical to what is buffered (a + conforming overlap), so nothing there conflicts either; only the + hole in the middle is genuinely new, and filling it must not appear + in ``conflict`` at all. + + """ + base = self.BASE + datagram = self._run( + self._packet(num=1, dsn=base, payload=b'AAAA'), # base..base+3 + self._packet(num=2, dsn=base + 10, payload=b'CCCC'), # base+10..base+13, gap base+4..base+9 + # base+2..base+11: 'AA' matches the buffered tail of segment 1, + # 'XXXXXX' fills the gap, 'CC' matches the buffered head of + # segment 2 -- none of the three is a disagreement + self._packet(num=3, dsn=base + 2, payload=b'AA' + b'X' * 6 + b'CC'), + ) + self.assertTrue(datagram.completed) + self.assertEqual(datagram.payload, b'AAAA' + b'X' * 6 + b'CCCC') + self.assertEqual(datagram.conflict, ()) + + def test_a_simultaneous_head_prepend_and_tail_extension_in_one_overlap_call(self) -> None: + """Full engulfment plus extension: a new head *and* a new tail in the same call. + + Corrects a wrong comment that used to sit on the reach-back branch, + claiming a head-prepend and a tail-append "cannot both happen at + once". They can: the old buffer (``isn=100``, ``len=10``) is fully + inside the arriving segment's range (``dsn=90``, ``len=30``), so the + arriving segment supplies genuinely new bytes *before* ``isn`` + (90..99) and genuinely new bytes *past* the old buffer's end + (110..119) in the very same :meth:`~pcapkit.foundation.reassembly.tcp.TCP._merge_overlap` + call. What actually is mutually exclusive is only which one of the + old buffer's own tail or a genuinely new one survives -- never both -- + and that is unaffected by the head also being new. + + The overlapping middle (100..109) deliberately disagrees with the + already-buffered bytes there, so the same call also proves + first-write-wins and the new head/tail merge correctly coexist. + + """ + base = self.BASE + old_isn = base + 100 + datagram = self._run( + self._packet(num=1, dsn=old_isn, payload=b'0123456789'), # isn=base+100, len=10 + self._packet(num=2, dsn=old_isn - 10, # dsn=base+90, len=30 + payload=b'A' * 10 + b'X' * 10 + b'C' * 10), + ) + self.assertTrue(datagram.completed) + self.assertEqual(datagram.payload, b'A' * 10 + b'0123456789' + b'C' * 10) + self.assertEqual(datagram.conflict, ((old_isn, old_isn + 9),)) + + def test_gap_list_matches_the_zero_filled_positions_across_200_random_trials(self) -> None: + """Property test: ``gap`` always names exactly the positions ``raw`` still has as fill. + + This is the correctness check the absolute-interval design is meant + to make trivial: since ``gap`` never shifts when ``isn`` moves, the + set of octets it names should equal the set of octets in ``raw`` that + no segment has ever supplied a real byte for -- checked directly + against ``raw``, not inferred from the merge arithmetic, so a bug + that got the merged *bytes* right but the *bookkeeping* wrong would + still be caught. 200 trials, random overlapping and out-of-order + segments fed into a single ACK bucket, fixed seed for a reproducible + run. + + Every synthetic payload avoids the zero byte, so an unfilled octet of + ``raw`` -- ``b'\\x00'`` -- is unambiguous: it is covered by a ``gap`` + entry if and only if no segment has ever placed a real byte there. + + """ + import random + + from pcapkit.foundation.reassembly.tcp import TCP + + rng = random.Random(20260918) + bufid = self._bufid() + + for trial in range(200): + with self.subTest(trial=trial): + base = rng.randrange(0, 2 ** 31) + reasm = TCP() + + segments = [] + cursor = 0 + for num in range(1, rng.randint(2, 8) + 1): + offset = cursor + rng.randint(-5, 10) + length = rng.randint(1, 20) + payload = bytes(rng.randint(1, 255) for _ in range(length)) # never 0x00 + segments.append((num, base + offset, payload)) + cursor = offset + length + rng.shuffle(segments) # out of order delivery + for (num, dsn, payload) in segments: + reasm(self._packet(num=num, dsn=dsn, payload=payload)) + + fragment = next(iter(reasm._buffer[bufid].ack.values())) + raw, isn, gap = fragment.raw, fragment.isn, fragment.gap + + # every gap entry is itself all-zero in raw, and the entries + # are pairwise disjoint + covered = set() + for (first, last) in gap: + self.assertLessEqual(first, last) + for seq in range(first, last + 1): + self.assertNotIn(seq, covered, 'gap entries overlap') + covered.add(seq) + self.assertEqual(raw[seq - isn], 0) + + # and every zero byte in raw is covered by some gap entry -- + # i.e. gap is not merely disjoint from real data, it is + # *exactly* the zero-filled positions, nothing more and + # nothing less + for offset in range(len(raw)): + if raw[offset] == 0: + self.assertIn(isn + offset, covered) + + if __name__ == '__main__': unittest.main()