|
| 1 | +# -*- coding: utf-8 -*- |
| 2 | +"""An unfragmented datagram whose BUFID is already buffered is reported. |
| 3 | +
|
| 4 | +GitHub issue #2270. A datagram with fragment offset zero and **MF** clear whose |
| 5 | +BUFID matched a buffered one submitted that buffer and returned, so the datagram |
| 6 | +itself was never reported. IPv6 inherited this, so a colliding atomic fragment |
| 7 | +also flushed the datagram being reassembled, and its final fragment then opened |
| 8 | +a buffer of its own that never completed. |
| 9 | +
|
| 10 | +:rfc:`791#section-3.2` flushes the buffer and then submits the datagram, so |
| 11 | +IPv4 reports both, the flushed one incomplete. :rfc:`6946#section-4` says an |
| 12 | +atomic fragment is processed in isolation and the queued fragments are not |
| 13 | +discarded, so IPv6 reports the atomic fragment alone and still completes the |
| 14 | +datagram it interrupted. |
| 15 | +
|
| 16 | +Every case builds its fragments in memory and reads no capture. Classes are |
| 17 | +imported inside each test, after :func:`~tests._support.reimport_once_per_class`. |
| 18 | +
|
| 19 | +""" |
| 20 | + |
| 21 | +from ipaddress import ip_address |
| 22 | +import unittest |
| 23 | + |
| 24 | +from tests._support import reimport_once_per_class |
| 25 | + |
| 26 | + |
| 27 | +class TestIPReassemblyUnfragmentedBufidCollision(unittest.TestCase): |
| 28 | + """Pin that an unfragmented datagram sharing a buffered BUFID is reported.""" |
| 29 | + |
| 30 | + def setUp(self) -> None: |
| 31 | + reimport_once_per_class(self) |
| 32 | + |
| 33 | + def _packet(self, *, v6: bool, num: int, fo: int, mf: bool, payload: bytes): |
| 34 | + """A fragment of the one BUFID every case shares.""" |
| 35 | + from pcapkit.const.reg.transtype import TransType |
| 36 | + from pcapkit.foundation.reassembly.data.ip import Packet |
| 37 | + |
| 38 | + if v6: |
| 39 | + src, dst, ihl = ip_address('2001:db8::1'), ip_address('2001:db8::2'), 40 |
| 40 | + else: |
| 41 | + src, dst, ihl = ip_address('192.0.2.1'), ip_address('192.0.2.2'), 20 |
| 42 | + return Packet((src, dst, 9, TransType.UDP), num, fo, ihl, mf, ihl + len(payload), |
| 43 | + b'H' * ihl, bytearray(payload), 1000.0 + num) |
| 44 | + |
| 45 | + def _reassemble(self, cls, packets, strict: bool): |
| 46 | + reasm = cls(strict=strict) |
| 47 | + for packet in packets: |
| 48 | + reasm(packet) |
| 49 | + return reasm, [(datagram.index, datagram.completed, datagram.payload) for datagram in reasm.datagram] |
| 50 | + |
| 51 | + def test_ipv4_flushes_the_buffer_and_reports_the_datagram(self) -> None: |
| 52 | + from pcapkit.foundation.reassembly.data.data import Completion |
| 53 | + from pcapkit.foundation.reassembly.ipv4 import IPv4 |
| 54 | + |
| 55 | + packets = [self._packet(v6=False, num=1, fo=0, mf=True, payload=b'A' * 16), |
| 56 | + self._packet(v6=False, num=2, fo=0, mf=False, payload=b'B' * 16)] |
| 57 | + for strict in (True, False): |
| 58 | + with self.subTest(strict=strict): |
| 59 | + reasm, datagrams = self._reassemble(IPv4, packets, strict) |
| 60 | + flushed = (b'A' * 16,) if strict else b'A' * 16 |
| 61 | + self.assertEqual(datagrams, [((1,), Completion.PARTIAL, flushed), |
| 62 | + ((2,), Completion.COMPLETE, b'B' * 16)]) |
| 63 | + self.assertEqual(reasm._buffer, {}) |
| 64 | + |
| 65 | + def test_ipv6_atomic_fragment_leaves_the_buffer_alone(self) -> None: |
| 66 | + from pcapkit.foundation.reassembly.data.data import Completion |
| 67 | + from pcapkit.foundation.reassembly.ipv6 import IPv6 |
| 68 | + |
| 69 | + packets = [self._packet(v6=True, num=1, fo=0, mf=True, payload=b'A' * 16), |
| 70 | + self._packet(v6=True, num=2, fo=0, mf=False, payload=b'B' * 16), |
| 71 | + self._packet(v6=True, num=3, fo=16, mf=False, payload=b'C' * 16)] |
| 72 | + for strict in (True, False): |
| 73 | + with self.subTest(strict=strict): |
| 74 | + reasm, datagrams = self._reassemble(IPv6, packets, strict) |
| 75 | + self.assertEqual(datagrams, [((2,), Completion.COMPLETE, b'B' * 16), |
| 76 | + ((1, 3), Completion.COMPLETE, b'A' * 16 + b'C' * 16)]) |
| 77 | + self.assertEqual(reasm._buffer, {}) |
| 78 | + |
| 79 | + |
| 80 | +if __name__ == '__main__': |
| 81 | + unittest.main() |
0 commit comments