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()