From c5bbadf97792791329a1dd96d7ad212d4804093b Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Wed, 7 Oct 2026 17:06:42 -0400 Subject: [PATCH] fix(reassembly): TCP completion from per-buffer gaps, mod-2**32 sequence numbers, FIN waits for reordered data - Judge each payload buffer's completion from its own gap list, not the buffer-wide hole list, which missed gaps below the first segment and was cleared by other ACK buckets (#1354). - Read every sequence number modulo 2**32 relative to the buffered data (RFC 9293 section 3.4); conflicts are reported reduced back mod 2**32 (#1355). - A FIN only fixes the end of the stream: the hole list is cut back to it and the buffer is submitted once every octet below the FIN has arrived, or at end of capture. RST still submits at once (RFC 9293 section 3.10.7.4) (#1356). - A segment that would zero-fill a gap wider than TCP.__window__ (16 MiB) splits the buffer instead. Without a SYN it starts a new stream; with one, both halves record the lost range as a gap and are reported PARTIAL. - The RFC 815 update now fills every hole a segment spans, not only the first. Closes #1354 Closes #1355 Closes #1356 --- pcapkit/foundation/reassembly/data/tcp.py | 28 +- pcapkit/foundation/reassembly/tcp.py | 348 ++++++++++++++---- tests/foundation/reassembly/test_tcp.py | 8 +- .../test_tcp_reassembly_roundtrip_unit.py | 205 +++++++++++ 4 files changed, 503 insertions(+), 86 deletions(-) create mode 100644 tests/foundation/reassembly/test_tcp_reassembly_roundtrip_unit.py diff --git a/pcapkit/foundation/reassembly/data/tcp.py b/pcapkit/foundation/reassembly/data/tcp.py index ee6f17ceb5..0eed7391a4 100644 --- a/pcapkit/foundation/reassembly/data/tcp.py +++ b/pcapkit/foundation/reassembly/data/tcp.py @@ -112,6 +112,7 @@ class Datagram(DeferredPacket, Info, Generic[_AT]): #: 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. + #: A range that crosses ``2 ** 32`` therefore has ``last < first``. #: #: Resolution keeps the already-buffered bytes and discards the #: conflicting portion of whichever segment arrived later, per @@ -134,13 +135,15 @@ def __init__(self, completed: 'Completion', id: 'DatagramID[_AT]', index: 'tuple class HoleDescriptor(Info): """Data model for :term:`TCP ` hole descriptor. - Both bounds are **absolute TCP sequence numbers** and both are - **inclusive**, so a hole covers ``last - first + 1`` octets. They are not - offsets into :attr:`Fragment.raw`: the descriptor list is kept once per - buffer ID, whereas each acknowledgement number's payload buffer carries an - initial sequence number of its own, so only - :meth:`TCP.submit ` -- which - knows which buffer it is looking at -- can convert one to the other. + Both bounds are **unwrapped TCP sequence numbers** -- read modulo + ``2 ** 32`` relative to the data already buffered, so a stream crossing + ``2 ** 32`` continues past it -- and both are **inclusive**, so a hole + covers ``last - first + 1`` octets. They are not offsets into + :attr:`Fragment.raw`: the descriptor list is kept once per buffer ID, + across every acknowledgement number's payload buffer, and describes the + stream as a whole. It decides when a stream whose FIN has been seen is + whole; a hole whose ``first`` is ``-sys.maxsize`` stands for whatever + preceded a capture that missed the handshake, and does not count. """ @@ -163,7 +166,9 @@ class Fragment(Info): #: buffer is indexed from: ``raw[n]`` holds the octet whose sequence number #: is ``isn + n``. Revised downwards whenever a segment turns up below the #: data already buffered, so it is not necessarily the connection's own - #: initial sequence number. + #: initial sequence number. Like every sequence number held in this + #: fragment -- :attr:`gap` and :attr:`conflict` included -- it is + #: unwrapped, and so may exceed ``2 ** 32`` on a stream that crossed it. isn: 'int' #: Length of payload buffer. len: 'int' @@ -176,7 +181,12 @@ class Fragment(Info): #: 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. + #: segment. The one exception lies wholly *outside* :attr:`raw`: when a + #: stream whose SYN was captured is split rather than padded across a gap + #: wider than :attr:`TCP.__window__ + #: `, both halves record + #: the range left between them. Any entry at all makes the fragment + #: incomplete. #: #: This is deliberately **not** derived from #: :attr:`Buffer.hdl `. diff --git a/pcapkit/foundation/reassembly/tcp.py b/pcapkit/foundation/reassembly/tcp.py index 38e1556faa..1611bdb0ff 100644 --- a/pcapkit/foundation/reassembly/tcp.py +++ b/pcapkit/foundation/reassembly/tcp.py @@ -50,21 +50,44 @@ class TCP(ReassemblyBase[Packet, Datagram, BufferID, Buffer]): Note: There are two coordinate systems in play here, and keeping them apart - matters. The :term:`hole descriptor list ` of - :rfc:`815` is kept in **absolute TCP sequence numbers**, inclusive of - both bounds, because a hole belongs to the connection's sequence space - for that direction and not to any one payload buffer: the list is held - once per buffer ID, while each acknowledgement number gets a payload - buffer of its own with an initial sequence number of its own, and that - initial sequence number is revised whenever a segment turns up below - the data already buffered. A payload buffer, on the other hand, is - indexed from zero, such that + matters. Sequence numbers -- in the :term:`hole descriptor list + ` of :rfc:`815`, and in each payload buffer's own + :attr:`~pcapkit.foundation.reassembly.data.tcp.Fragment.gap` and + :attr:`~pcapkit.foundation.reassembly.data.tcp.Fragment.conflict` lists + -- are kept **unwrapped**, inclusive of both bounds: every arriving + sequence number is read modulo ``2 ** 32`` relative to the data + already buffered under its buffer ID, per :rfc:`9293#section-3.4`, and + placed on one unbounded integer line. A connection that crosses + ``2 ** 32`` therefore continues at ``2 ** 32`` rather than back at + zero. A payload buffer, on the other hand, is indexed from zero, such + that :attr:`buffer.raw[n] ` holds the octet with sequence number :attr:`buffer.isn ` - ``+ n``. :meth:`submit` is therefore the one place that converts - between the two, subtracting that buffer's initial sequence number from - each hole bound. + ``+ n``. :meth:`submit` converts between the two, subtracting that + buffer's initial sequence number from each gap bound, and reports + conflict ranges reduced back to modulo ``2 ** 32``. + + The hole descriptor list is held once per buffer ID, across every + acknowledgement number's payload buffer, so it describes the *stream* + rather than any one payload buffer: it decides when a stream whose FIN + has been seen is whole and may be submitted. Whether each payload + buffer is complete is a question about that buffer alone, answered + from its own + :attr:`~pcapkit.foundation.reassembly.data.tcp.Fragment.gap` list. + + Warning: + **Limitation.** Without the handshake the start of a stream is unknown, + so a FIN is taken to complete the stream once every octet from the + lowest one seen so far up to the FIN has arrived. Data that arrives + later but lies *below* everything seen before that FIN therefore opens + a buffer of its own, and the first buffer has already been reported + :attr:`~pcapkit.foundation.reassembly.data.data.Completion.COMPLETE` + although the stream was not: ``[seq 1000 A*5, FIN 1005, seq 990 B*10]`` + gives ``B*10`` and ``A*5`` as two datagrams. This applies only to a + buffer begun without a SYN: once the SYN is captured, the stream's + start is the SYN's own sequence number, and a FIN completes the + stream only when every octet from there up to the FIN has arrived. """ if TYPE_CHECKING: @@ -96,6 +119,34 @@ class TCP(ReassemblyBase[Packet, Datagram, BufferID, Buffer]): #: ask for one, e.g. ``2 * 120`` for 2·MSL if that is the policy wanted. __timeout__ = math.inf + #: int: Widest gap, in octets, that merging a segment into the payload + #: buffer it is read against may zero-fill: 16 MiB. The gap is measured on + #: whichever side the segment lands, from the buffer's end to the + #: segment's start, or from the segment's end to the buffer's start. + #: + #: A segment can only legitimately open a gap as wide as the data in flight, + #: which the receive window bounds. :rfc:`7323` allows windows up to + #: ``2 ** 30``, but zero-filling a gap that wide costs 1 GiB, and real + #: windows are far smaller -- Linux's default ``net.ipv4.tcp_rmem`` ceiling + #: is 6 MiB -- so 16 MiB clears them with room to spare. A segment that + #: would open a wider gap is not padded to; the buffer held so far is + #: submitted and a fresh one opened for the segment instead. What that + #: means depends on whether the buffer's SYN was captured: + #: + #: * **No SYN** -- the segment is taken to start a new stream on a reused + #: 4-tuple, and the two buffers are reported independently. + #: * **SYN captured** -- the SYN proves both are the same stream with the + #: range in between lost, so that range is recorded in the + #: :attr:`~pcapkit.foundation.reassembly.data.tcp.Fragment.gap` of the + #: payload buffer the segment was read against and of the new one, and + #: both are reported + #: :attr:`~pcapkit.foundation.reassembly.data.data.Completion.PARTIAL` + #: (in strict mode, as a single run each). The new buffer keeps the + #: SYN's header and counts as having seen it. + #: + #: Override in a subclass to change the bound. + __window__ = 1 << 24 + ########################################################################## # Methods. ########################################################################## @@ -138,20 +189,59 @@ def reassembly(self, info: 'Packet') -> 'None': self.submit(self._buffer.pop(BUFID), bufid=BUFID) ) + # Read the sequence number modulo 2**32 relative to the end of the data + # it most plausibly continues -- this ACK's own payload buffer, or else + # the most recently opened one -- per :rfc:`9293#section-3.4`. Plain + # integer comparison reads a stream crossing 2**32 as a gap of about + # 4 GiB. A segment landing farther than ``__window__`` outside that + # buffer starts a new stream instead -- see ``__window__``. + SHIFT = 0 + HDR = info.header if SYN else b'' # header of the buffer opened below, if any + ANCHORED = SYN # whether that buffer's stream start is known + MISSING = [] # type: list[tuple[int, int]] # range a split left out of it + if BUFID in self._buffer: + buffer = self._buffer[BUFID] + fragment = buffer.ack.get(ACK) + if fragment is None: + fragment = next(reversed(buffer.ack.values())) + END = fragment.isn + fragment.len + UNWRAPPED = self._unwrap(PSN, END) + # the zero-filled gap merging this segment would open, either side + if max(UNWRAPPED - END, fragment.isn - (UNWRAPPED + info.len)) > self.__window__: + old = self._buffer.pop(BUFID) + if self._anchored(old): + # The SYN proves this is still the same stream, with the + # range in between lost. Both halves record that range as + # missing, so each is reported PARTIAL rather than COMPLETE. + MISSING.append((END, UNWRAPPED - 1) if UNWRAPPED > END + else (UNWRAPPED + info.len, fragment.isn - 1)) + fragment.gap.extend(MISSING) + HDR, ANCHORED, PSN = old.hdr, True, UNWRAPPED + self._dtgram.extend(self.submit(old, bufid=BUFID)) + else: + SHIFT = UNWRAPPED - PSN + PSN = UNWRAPPED + # initialise buffer with BUFID & ACK if BUFID not in self._buffer: + hdl = [ + HoleDescriptor( + # everything from the octet after this segment onwards is + # still missing + first=PSN + info.len, + last=sys.maxsize, + ), + ] + if not ANCHORED: + # Without the handshake the stream's start is unknown, so what + # lies below this segment is open too. The sentinel lower bound + # marks this one hole as "before the capture began" rather than + # as data missing from the stream; it is what lets a segment + # arriving *below* this one leave a real hole between the two. + hdl.insert(0, HoleDescriptor(first=-sys.maxsize, last=PSN - 1)) self._buffer[BUFID] = Buffer( - hdl=[ - HoleDescriptor( - # everything from the octet after this segment onwards - # is still missing -- in absolute sequence numbers, so - # that the bound stays valid for every payload buffer - # under this buffer ID - first=PSN + info.len, - last=sys.maxsize, - ), - ], - hdr=info.header if SYN else b'', + hdl=hdl, + hdr=HDR, ack={ ACK: Fragment( ind=[ @@ -160,8 +250,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=[], + # this segment's own payload is, by definition, all + # real; a split stream also records what it left out + gap=MISSING, conflict=[], ), }, @@ -169,8 +260,8 @@ def reassembly(self, info: 'Packet') -> 'None': ) else: # initialise buffer with ACK - if ACK not in self._buffer[BUFID].ack: - self._buffer[BUFID].ack[ACK] = Fragment( + if ACK not in buffer.ack: + buffer.ack[ACK] = Fragment( ind=[ info.num, ], @@ -183,13 +274,13 @@ def reassembly(self, info: 'Packet') -> 'None': else: # put header into header buffer if SYN: # pragma: no cover - self._buffer[BUFID].__update__(hdr=info.header) + buffer.__update__(hdr=info.header) # append packet index - self._buffer[BUFID].ack[ACK].ind.append(info.num) + fragment = buffer.ack[ACK] + fragment.ind.append(info.num) # record fragment payload - 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 @@ -204,14 +295,88 @@ 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: - self._update_hole_descriptors(info, BUFID, FIN, RST) - - # when FIN/RST is set, submit buffer of this session - if FIN or RST: + self._update_hole_descriptors(BUFID, info.first + SHIFT, info.last + SHIFT) + + # A FIN fixes the end of the stream at the sequence number it occupies, + # and nothing more: data sent before it may still arrive after it. So + # the open-ended hole is cut back to end there, and the buffer is + # submitted once every octet below the FIN has arrived -- here, or on + # whichever later segment closes the last hole -- or else when the + # capture ends. + if FIN: + self._close_hole_descriptors(BUFID, PSN + info.len) + + # An RST, unlike a FIN, aborts the connection: :rfc:`9293#section-3.10.7.4` + # has the receiver flush its segment queues and enter CLOSED, so it + # accepts nothing sent before the RST that arrives after it. The buffer + # is therefore submitted at once, as the receiver last saw it. + if RST or self._stream_whole(BUFID): self._dtgram.extend( self.submit(self._buffer.pop(BUFID), bufid=BUFID) ) + @staticmethod + def _unwrap(seq: 'int', ref: 'int') -> 'int': + """Place a 32-bit sequence number on the unwrapped line near ``ref``. + + Arguments: + seq: sequence number as carried on the wire, i.e. modulo 2**32 + ref: unwrapped sequence number the segment is expected to be near + + Returns: + The integer congruent to ``seq`` modulo 2**32 that lies within + 2**31 of ``ref``, which is how :rfc:`9293#section-3.4` compares + sequence numbers. + + """ + return ref + (seq - ref + 0x80000000) % 0x100000000 - 0x80000000 + + @staticmethod + def _anchored(buf: 'Buffer') -> 'bool': + """Whether a buffer's stream start is known, i.e. its SYN was captured. + + Arguments: + buf: :term:`buffer ` to look at + + Returns: + :data:`False` exactly when the hole descriptor list still opens + with the sentinel hole that a buffer begun without a SYN carries. + + """ + return not (buf.hdl and buf.hdl[0].first == -sys.maxsize) + + def _stream_whole(self, BUFID: 'BufferID') -> 'bool': + """Whether a stream has seen its FIN and every octet below it. + + Arguments: + BUFID: buffer identifier of the stream + + Returns: + :data:`True` once a FIN has cut back the open-ended hole and the + only hole left, if any, is the sentinel one below the first octet + of a capture that missed the handshake. + + """ + for hole in self._buffer[BUFID].hdl: + if hole.first != -sys.maxsize or hole.last == sys.maxsize: + return False + return True + + def _close_hole_descriptors(self, BUFID: 'BufferID', end: 'int') -> 'None': + """Cut the hole descriptor list back to end below a FIN. + + Arguments: + BUFID: buffer identifier of the stream + end: unwrapped sequence number the FIN occupies, i.e. one past the + last octet of the stream + + """ + HDL = self._buffer[BUFID].hdl + HDL[:] = [ + hole if hole.last < end else HoleDescriptor(first=hole.first, last=end - 1) + for hole in HDL if hole.first < end + ] + def _reassemble_append(self, info: 'Packet', fragment: 'Fragment', PSN: 'int') -> 'None': """Merge a segment that starts at or after the buffered payload's end. @@ -259,6 +424,7 @@ def _reassemble_append(self, info: 'Packet', fragment: 'Fragment', PSN: 'int') - fragment.conflict.extend(conflicts) if info.len > OVERLAP: # fragment reaches past the buffered end RAW += info.payload[OVERLAP:] + self._trim_gaps(GAPS, PSN, PSN + info.len - 1) fragment.__update__( raw=RAW, # update payload datagram len=len(RAW), # update payload length @@ -324,13 +490,13 @@ def _reassemble_prepend(self, info: 'Packet', fragment: 'Fragment', PSN: 'int') RAW[:OVERLAP] = merged fragment.conflict.extend(conflicts) RAW = info.payload[:OFFSET] + RAW + info.payload[OFFSET + OVERLAP:] + self._trim_gaps(GAPS, PSN, PSN + info.len - 1) 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': + def _update_hole_descriptors(self, BUFID: 'BufferID', first: 'int', last: 'int') -> 'None': """Update the buffer-wide hole descriptor list per :rfc:`815`. Called only for a segment that carries payload (``info.len > 0``); @@ -339,35 +505,65 @@ def _update_hole_descriptors(self, info: 'Packet', BUFID: 'BufferID', adjacent holes covering the very same octets, growing the list without bound on a long-lived connection. + The end of the stream is not settled here: a FIN cuts the open-ended + hole back afterwards, in :meth:`_close_hole_descriptors`, whether or + not it carries payload. + 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 + first: unwrapped sequence number of the segment's first octet + last: unwrapped sequence number of the segment's last octet """ 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 + new_hdl = [] # type: list[HoleDescriptor] + for hole in HDL: # step one + if first > hole.last or last < hole.first: # steps two and three + new_hdl.append(hole) continue - del HDL[index] # step four - if info.first > hole.first: # step five - new_hole = HoleDescriptor( + # step four: this hole is (partly) filled and replaced by what of + # it survives, on either side of the segment + if first > hole.first: # step five + new_hdl.append(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=first - 1, + )) + if last < hole.last: # step six + new_hdl.append(HoleDescriptor( + first=last + 1, last=hole.last - ) - HDL.insert(index, new_hole) - break # step seven - #self._buffer[BUFID].hdl = HDL # update HDL + )) + # step seven: go on to the next hole, since one segment may span + # several of them + HDL[:] = new_hdl + + @staticmethod + def _trim_gaps(gap: 'list[tuple[int, int]]', first: 'int', last: 'int') -> 'None': + """Remove a received range from a fragment's gap list, in place. + + Only an entry lying outside :attr:`~pcapkit.foundation.reassembly.data.tcp.Fragment.raw` + -- the range a split stream left out, see :attr:`TCP.__window__` -- can + still overlap a segment once it is merged; :meth:`_merge_overlap` has + already closed every entry inside ``raw`` that the segment covers. + + Arguments: + gap: the fragment's gap list + first: unwrapped sequence number of the segment's first octet + last: unwrapped sequence number of the segment's last octet + + """ + if first > last: + return + still_gap = [] # type: list[tuple[int, int]] + for (lo, hi) in gap: + if hi < first or lo > last: + still_gap.append((lo, hi)) + continue + if lo < first: + still_gap.append((lo, first - 1)) + if hi > last: + still_gap.append((last + 1, hi)) + gap[:] = still_gap @staticmethod def _merge_overlap(gap: 'list[tuple[int, int]]', old: 'bytes', new: 'bytes', @@ -466,37 +662,41 @@ def submit(self, buf: 'Buffer', *, bufid: 'BufferID', # type: ignore[override] """ datagram = [] # type: list[Datagram] # reassembled datagram - HDL = buf.hdl # hole descriptor list # check through every buffer with ACK for (ack, buffer) in buf.ack.items(): - # Translate the hole descriptor list, which is kept in absolute - # sequence numbers for the whole direction, into offsets into this - # payload buffer, which is indexed from its own initial sequence - # number. Holes lying wholly outside this buffer -- the open-ended - # one past the last octet received, and any belonging to a - # different acknowledgement number's data -- drop out here; those - # straddling an edge are clipped to it rather than being allowed to - # index from the far end of the buffer as a negative bound would. + # Translate this payload buffer's own gap list, kept in unwrapped + # sequence numbers, into offsets into the buffer, which is indexed + # from its own initial sequence number. The buffer-wide hole + # descriptor list cannot answer this: it is shared by every + # acknowledgement number, so another payload buffer filling a hole + # there says nothing about this one, and it never learns of a gap + # opened below the first segment to arrive. Entries are clipped to + # the buffer rather than being allowed to index from its far end. length = len(buffer.raw) holes = [] # type: list[tuple[int, int]] - for hole in HDL: - start = hole.first - buffer.isn # inclusive lower bound - stop = hole.last - buffer.isn + 1 # exclusive upper bound + for (first, last) in buffer.gap: + start = first - buffer.isn # inclusive lower bound + stop = last - buffer.isn + 1 # exclusive upper bound if stop <= 0 or start >= length: - continue # hole misses this buffer + continue # gap lies outside the buffer holes.append((max(start, 0), min(stop, length))) holes.sort() + # reported in sequence space proper, i.e. modulo 2**32 + conflict = tuple((first & 0xFFFFFFFF, last & 0xFFFFFFFF) + for (first, last) in buffer.conflict) # How completely this buffer came out, and why it stopped. Derived # once per buffer, so the two branches cannot disagree about it. - completion = Completion.COMPLETE if not holes else ( + # Any gap entry counts, including one lying wholly outside ``raw``: + # that is the range a split stream left out (see ``__window__``). + completion = Completion.COMPLETE if not buffer.gap else ( Completion.TIMEOUT if timeout else Completion.PARTIAL ) # if this buffer is not implemented # go through every hole and extract received payload - if holes and self._flag_s: + if completion is not Completion.COMPLETE and self._flag_s: data = [] # type: list[bytes] start = 0 for (hole_start, hole_stop) in holes: @@ -519,7 +719,7 @@ def submit(self, buf: 'Buffer', *, bufid: 'BufferID', # type: ignore[override] header=buf.hdr, payload=tuple(data), packet=None, - conflict=tuple(buffer.conflict), + conflict=conflict, ) datagram.append(packet) @@ -545,7 +745,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), + conflict=conflict, ) datagram.append(packet) diff --git a/tests/foundation/reassembly/test_tcp.py b/tests/foundation/reassembly/test_tcp.py index 6d43808ae3..3e340b8d56 100644 --- a/tests/foundation/reassembly/test_tcp.py +++ b/tests/foundation/reassembly/test_tcp.py @@ -185,7 +185,8 @@ 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'), + [(2, 3), (7, 8), (99, 100)], [])}, 1000.0, ), bufid=bufid, @@ -205,7 +206,8 @@ class TestTCP(TCP): b'tcp-header', { 500: Fragment([], 0, 0, bytearray(), [], []), - 501: Fragment([9], 0, 9, bytearray(b'abcdefghi'), [], [(2, 3)]), + 501: Fragment([9], 0, 9, bytearray(b'abcdefghi'), + [(0, 0), (4, 5), (7, 7)], [(2, 3)]), }, 1000.0, ), @@ -228,7 +230,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'), [(2, 3)], [])}, 1000.0, ), bufid=bufid, diff --git a/tests/foundation/reassembly/test_tcp_reassembly_roundtrip_unit.py b/tests/foundation/reassembly/test_tcp_reassembly_roundtrip_unit.py new file mode 100644 index 0000000000..563d4600d7 --- /dev/null +++ b/tests/foundation/reassembly/test_tcp_reassembly_roundtrip_unit.py @@ -0,0 +1,205 @@ +# -*- coding: utf-8 -*- +"""TCP reassembly reports what arrived: gaps, sequence wrap, and late data. + +GitHub issues #1354, #1355 and #1356, from the foundation round-trip audit +(#1202): + +* #1354 -- completion was judged from the buffer-wide hole list, which never + learnt of a gap opened *below* the first segment to arrive, and which another + acknowledgement number's payload buffer could clear. A zero-filled gap came + back :attr:`~pcapkit.foundation.reassembly.data.data.Completion.COMPLETE`. +* #1355 -- sequence numbers were compared as plain integers, so a stream + crossing ``2 ** 32`` read as a gap of about 4 GiB and allocated it. +* #1356 -- a FIN submitted the buffer at once, so data reordered behind it was + split off into a buffer of its own. + +Every case feeds hand-built segments and reads no capture. Classes are imported +inside each test, after :func:`~tests._support.reimport_once_per_class`. + +""" + +import unittest +from ipaddress import ip_address + +from tests._support import reimport_once_per_class + +BUFID = (ip_address('192.0.2.1'), 12345, ip_address('198.51.100.2'), 443) + + +class TestTCPReassemblyRoundTrip(unittest.TestCase): + """Pin completion, wrap-around and FIN handling of TCP reassembly.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + @staticmethod + def _seg(seq, payload=b'', *, ack=500, syn=False, fin=False, rst=False): + """Describe a segment; :meth:`_run` builds it, numbered by position.""" + return (seq, ack, syn, fin, rst, payload) + + @staticmethod + def _summary(reasm): + return [(d.completed.name, d.id.ack, d.index, d.payload) for d in reasm.datagram] + + def _run(self, *segments, strict=True): + from pcapkit.foundation.reassembly.data.tcp import Packet + from pcapkit.foundation.reassembly.tcp import TCP + + reasm = TCP(strict=strict) + for num, (seq, ack, syn, fin, rst, payload) in enumerate(segments, 1): + reasm(Packet(BUFID, seq, ack, num, syn, fin, rst, len(payload), seq, + seq + len(payload) - 1, b'H' if syn else b'', bytearray(payload), 1000.0)) + return reasm + + # #1354 ------------------------------------------------------------------ + + def test_gap_below_the_first_segment_is_partial(self) -> None: + for strict, payload in ((True, (b'A' * 100, b'B' * 100)), + (False, b'A' * 100 + bytes(100) + b'B' * 100)): + with self.subTest(strict=strict): + reasm = self._run(self._seg(200, b'B' * 100), self._seg(0, b'A' * 100), + self._seg(300, fin=True), strict=strict) + self.assertEqual(self._summary(reasm), [('PARTIAL', 500, (1, 2, 3), payload)]) + + def test_another_ack_bucket_does_not_fill_this_ones_gap(self) -> None: + reasm = self._run(self._seg(0, syn=True, ack=0), self._seg(1, b'A' * 100, ack=1000), + self._seg(201, b'C' * 100, ack=1000), self._seg(101, b'B' * 100, ack=1050), + self._seg(301, fin=True, ack=1050)) + by_ack = {ack: (completed, payload) for (completed, ack, _, payload) in self._summary(reasm)} + self.assertEqual(by_ack[1000], ('PARTIAL', (b'A' * 100, b'C' * 100))) + # the stream as a whole is covered, so the FIN submitted it + self.assertEqual(reasm._buffer, {}) + + def test_one_segment_filling_two_holes_completes_the_stream(self) -> None: + whole = b''.join(c * 10 for c in (b'A', b'B', b'C', b'D', b'E')) + reasm = self._run(self._seg(0, syn=True), self._seg(1, whole[:10]), self._seg(21, whole[20:30]), + self._seg(41, whole[40:]), self._seg(1, whole), self._seg(51, fin=True)) + # the retransmission closed both holes, so the FIN submitted the stream + self.assertEqual(reasm._buffer, {}) + self.assertEqual(self._summary(reasm), [('COMPLETE', 500, (1, 2, 3, 4, 5, 6), whole)]) + + # #1355 ------------------------------------------------------------------ + + def _no_big_allocation(self): + import pcapkit.foundation.reassembly.tcp as module + + def guarded(*args): + if args and isinstance(args[0], int) and args[0] > 1 << 20: + raise MemoryError(f'would allocate bytearray({args[0]})') + return bytearray(*args) + module.bytearray = guarded + self.addCleanup(delattr, module, 'bytearray') + + def test_sequence_numbers_wrap_modulo_2_32(self) -> None: + self._no_big_allocation() + for order in ((0, 1, 2), (1, 0, 2), (2, 1, 0)): + segs = [self._seg(0xFFFFFF00, b'A' * 256), self._seg(0, b'B' * 256), self._seg(256, fin=True)] + with self.subTest(order=order): + reasm = self._run(*(segs[i] for i in order)) + self.assertEqual([(c, p) for (c, _, _, p) in self._summary(reasm)], + [('COMPLETE', b'A' * 256 + b'B' * 256)]) + + def test_wrap_with_handshake_at_the_top_of_sequence_space(self) -> None: + self._no_big_allocation() + reasm = self._run(self._seg(0xFFFFFFFF, syn=True), self._seg(0, b'A' * 10), + self._seg(10, b'B' * 10), self._seg(20, fin=True)) + self.assertEqual(self._summary(reasm), [('COMPLETE', 500, (1, 2, 3, 4), b'A' * 10 + b'B' * 10)]) + + def test_conflict_past_the_wrap_is_reported_modulo_2_32(self) -> None: + self._no_big_allocation() + reasm = self._run(self._seg(0xFFFFFF00, b'A' * 256), self._seg(0, b'B' * 256), + self._seg(16, b'X' * 16)) + datagram, = reasm.datagram + self.assertEqual(datagram.conflict, ((16, 31),)) + self.assertEqual(datagram.payload, b'A' * 256 + b'B' * 256) + + def test_a_segment_half_the_sequence_space_away_starts_a_new_stream(self) -> None: + self._no_big_allocation() + for seq in (0x80000000, 0x7FFFFF00, 0x80000100): + with self.subTest(seq=hex(seq)): + reasm = self._run(self._seg(0, b'A' * 10), self._seg(seq, b'N' * 5)) + self.assertEqual(sorted(p for (_, _, _, p) in self._summary(reasm)), + [b'A' * 10, b'N' * 5]) + + # #1356 ------------------------------------------------------------------ + + def test_a_reused_4_tuple_behind_a_pending_fin_does_not_join_it(self) -> None: + # data, a FIN with a hole below it, then a new stream on the same + # 4-tuple whose SYN was lost: without a bound the new segment pads the + # old buffer out by some 768 MiB + self._no_big_allocation() + reasm = self._run(self._seg(1, b'A' * 5), self._seg(11, fin=True), + self._seg(0x30000000, b'N' * 5)) + self.assertEqual(self._summary(reasm), [('COMPLETE', 500, (3,), b'N' * 5), + ('PARTIAL', 500, (1, 2), (b'A' * 5,))]) + + def test_a_far_segment_after_a_captured_syn_splits_both_halves_partial(self) -> None: + # The SYN proves A and B are one stream with 20 MiB lost between them: + # neither half may come back COMPLETE, and nothing pads the gap. + self._no_big_allocation() + far = 1 + 1000 + 20 * (1 << 20) + for strict, a, b in ((True, (b'A' * 1000,), (b'B' * 1000,)), + (False, b'A' * 1000, b'B' * 1000)): + with self.subTest(strict=strict): + reasm = self._run(self._seg(0, syn=True), self._seg(1, b'A' * 1000), + self._seg(far, b'B' * 1000), strict=strict) + self.assertEqual(self._summary(reasm), [('PARTIAL', 500, (3,), b), + ('PARTIAL', 500, (1, 2), a)]) + datagram = reasm.datagram[0] + self.assertEqual(datagram.header, b'H') + + def test_a_split_halfs_missing_range_shrinks_as_data_fills_it(self) -> None: + # after the split, a segment just below B joins B and must not be + # mistaken for part of the range still missing + self._no_big_allocation() + far = 1 + 1000 + 20 * (1 << 20) + reasm = self._run(self._seg(0, syn=True), self._seg(1, b'A' * 1000), + self._seg(far, b'B' * 10), self._seg(far - 10, b'C' * 10)) + completed, _, index, payload = self._summary(reasm)[0] + self.assertEqual((completed, index, payload), ('PARTIAL', (3, 4), (b'C' * 10 + b'B' * 10,))) + self.assertEqual(reasm._buffer[BUFID].ack[500].gap, [(1001, far - 11)]) + + def test_a_far_segment_below_a_captured_syn_stream_splits_both_halves_partial(self) -> None: + self._no_big_allocation() + start = 0x30000000 + reasm = self._run(self._seg(start - 1, syn=True), self._seg(start, b'A' * 5), + self._seg(start - (1 << 25), b'B' * 5)) + self.assertEqual([c for (c, _, _, _) in self._summary(reasm)], ['PARTIAL', 'PARTIAL']) + + def test_a_segment_far_below_the_buffer_starts_a_new_stream(self) -> None: + self._no_big_allocation() + reasm = self._run(self._seg(0x30000000, b'A' * 5), self._seg(0x10000000, b'B' * 5)) + self.assertEqual(sorted(p for (_, _, _, p) in self._summary(reasm)), [b'A' * 5, b'B' * 5]) + + def test_a_gap_within_the_window_is_still_padded(self) -> None: + from pcapkit.foundation.reassembly.tcp import TCP + + far = TCP.__window__ + reasm = self._run(self._seg(0, b'A' * 5), self._seg(5 + far, b'B' * 5), strict=False) + (completed, _, index, payload), = self._summary(reasm) + self.assertEqual((completed, index, len(payload)), ('PARTIAL', (1, 2), 10 + far)) + + def test_data_reordered_behind_the_fin_joins_the_stream(self) -> None: + reasm = self._run(self._seg(100, syn=True, ack=0), self._seg(101, b'AAAAA'), + self._seg(111, fin=True), self._seg(106, b'BBBBB')) + # the segment that closed the last hole submitted the buffer + self.assertEqual(reasm._buffer, {}) + self.assertEqual(self._summary(reasm), [('COMPLETE', 500, (2, 3, 4), b'AAAAABBBBB')]) + + def test_fin_with_data_that_never_arrives_waits_for_the_end_of_capture(self) -> None: + reasm = self._run(self._seg(100, syn=True, ack=0), self._seg(101, b'AAAAA'), + self._seg(111, fin=True)) + self.assertIn(BUFID, reasm._buffer) + self.assertEqual(self._summary(reasm), [('PARTIAL', 500, (2, 3), (b'AAAAA',))]) + + def test_rst_aborts_at_once(self) -> None: + # RFC 9293 section 3.10.7.4: an RST flushes the receiver's queues, so + # data arriving after it is not part of the stream it ended + reasm = self._run(self._seg(100, syn=True, ack=0), self._seg(101, b'AAAAA'), + self._seg(111, rst=True), self._seg(106, b'BBBBB')) + self.assertEqual(self._summary(reasm), [('COMPLETE', 500, (4,), b'BBBBB'), + ('PARTIAL', 500, (2, 3), (b'AAAAA',))]) + + +if __name__ == '__main__': + unittest.main()