test/test_reassembly.py

   1 #!/usr/bin/env python3
   2
   3 import unittest
   4 from random import shuffle, choice, randrange
   5
   6 from framework import VppTestCase, VppTestRunner
   7
   8 import scapy.compat
   9 from scapy.packet import Raw
  10 from scapy.layers.l2 import Ether, GRE
  11 from scapy.layers.inet import IP, UDP, ICMP, icmptypes
  12 from scapy.layers.inet6 import (
  13     HBHOptUnknown,
  14     ICMPv6ParamProblem,
  15     ICMPv6TimeExceeded,
  16     IPv6,
  17     IPv6ExtHdrFragment,
  18     IPv6ExtHdrHopByHop,
  19     IPv6ExtHdrDestOpt,
  20     PadN,
  21     ICMPv6EchoRequest,
  22     ICMPv6EchoReply,
  23 )
  24 from framework import VppTestCase, VppTestRunner
  25 from util import ppp, ppc, fragment_rfc791, fragment_rfc8200
  26 from vpp_gre_interface import VppGreInterface
  27 from vpp_ip import DpoProto
  28 from vpp_ip_route import VppIpRoute, VppRoutePath, FibPathProto
  29 from vpp_papi import VppEnum
  30
  31 # 35 is enough to have >257 400-byte fragments
  32 test_packet_count = 35
  33
  34
  35 class TestIPv4Reassembly(VppTestCase):
  36     """IPv4 Reassembly"""
  37
  38     @classmethod
  39     def setUpClass(cls):
  40         super().setUpClass()
  41
  42         cls.create_pg_interfaces([0, 1])
  43         cls.src_if = cls.pg0
  44         cls.dst_if = cls.pg1
  45
  46         # setup all interfaces
  47         for i in cls.pg_interfaces:
  48             i.admin_up()
  49             i.config_ip4()
  50             i.resolve_arp()
  51
  52         # packet sizes
  53         cls.packet_sizes = [64, 512, 1518, 9018]
  54         cls.padding = " abcdefghijklmn"
  55         cls.create_stream(cls.packet_sizes)
  56         cls.create_fragments()
  57
  58     @classmethod
  59     def tearDownClass(cls):
  60         super().tearDownClass()
  61
  62     def setUp(self):
  63         """Test setup - force timeout on existing reassemblies"""
  64         super().setUp()
  65         self.vapi.ip_reassembly_enable_disable(
  66             sw_if_index=self.src_if.sw_if_index, enable_ip4=True
  67         )
  68         self.vapi.ip_reassembly_set(
  69             timeout_ms=0,
  70             max_reassemblies=1000,
  71             max_reassembly_length=1000,
  72             expire_walk_interval_ms=10,
  73         )
  74         self.virtual_sleep(0.25)
  75         self.vapi.ip_reassembly_set(
  76             timeout_ms=1000000,
  77             max_reassemblies=1000,
  78             max_reassembly_length=1000,
  79             expire_walk_interval_ms=10000,
  80         )
  81
  82     def tearDown(self):
  83         self.vapi.ip_reassembly_enable_disable(
  84             sw_if_index=self.src_if.sw_if_index, enable_ip4=False
  85         )
  86         super().tearDown()
  87
  88     def show_commands_at_teardown(self):
  89         self.logger.debug(self.vapi.ppcli("show ip4-full-reassembly details"))
  90         self.logger.debug(self.vapi.ppcli("show buffers"))
  91
  92     @classmethod
  93     def create_stream(cls, packet_sizes, packet_count=test_packet_count):
  94         """Create input packet stream
  95
  96         :param list packet_sizes: Required packet sizes.
  97         """
  98         for i in range(0, packet_count):
  99             info = cls.create_packet_info(cls.src_if, cls.src_if)
 100             payload = cls.info_to_payload(info)
 101             p = (
 102                 Ether(dst=cls.src_if.local_mac, src=cls.src_if.remote_mac)
 103                 / IP(
 104                     id=info.index, src=cls.src_if.remote_ip4, dst=cls.dst_if.remote_ip4
 105                 )
 106                 / UDP(sport=1234, dport=5678)
 107                 / Raw(payload)
 108             )
 109             size = packet_sizes[(i // 2) % len(packet_sizes)]
 110             cls.extend_packet(p, size, cls.padding)
 111             info.data = p
 112
 113     @classmethod
 114     def create_fragments(cls):
 115         infos = cls._packet_infos
 116         cls.pkt_infos = []
 117         for index, info in infos.items():
 118             p = info.data
 119             # cls.logger.debug(ppp("Packet:",
 120             #                      p.__class__(scapy.compat.raw(p))))
 121             fragments_400 = fragment_rfc791(p, 400)
 122             fragments_300 = fragment_rfc791(p, 300)
 123             fragments_200 = [x for f in fragments_400 for x in fragment_rfc791(f, 200)]
 124             cls.pkt_infos.append((index, fragments_400, fragments_300, fragments_200))
 125         cls.fragments_400 = [x for (_, frags, _, _) in cls.pkt_infos for x in frags]
 126         cls.fragments_300 = [x for (_, _, frags, _) in cls.pkt_infos for x in frags]
 127         cls.fragments_200 = [x for (_, _, _, frags) in cls.pkt_infos for x in frags]
 128         cls.logger.debug(
 129             "Fragmented %s packets into %s 400-byte fragments, "
 130             "%s 300-byte fragments and %s 200-byte fragments"
 131             % (
 132                 len(infos),
 133                 len(cls.fragments_400),
 134                 len(cls.fragments_300),
 135                 len(cls.fragments_200),
 136             )
 137         )
 138
 139     def verify_capture(self, capture, dropped_packet_indexes=[]):
 140         """Verify captured packet stream.
 141
 142         :param list capture: Captured packet stream.
 143         """
 144         info = None
 145         seen = set()
 146         for packet in capture:
 147             try:
 148                 self.logger.debug(ppp("Got packet:", packet))
 149                 ip = packet[IP]
 150                 udp = packet[UDP]
 151                 payload_info = self.payload_to_info(packet[Raw])
 152                 packet_index = payload_info.index
 153                 self.assertTrue(
 154                     packet_index not in dropped_packet_indexes,
 155                     ppp("Packet received, but should be dropped:", packet),
 156                 )
 157                 if packet_index in seen:
 158                     raise Exception(ppp("Duplicate packet received", packet))
 159                 seen.add(packet_index)
 160                 self.assertEqual(payload_info.dst, self.src_if.sw_if_index)
 161                 info = self._packet_infos[packet_index]
 162                 self.assertTrue(info is not None)
 163                 self.assertEqual(packet_index, info.index)
 164                 saved_packet = info.data
 165                 self.assertEqual(ip.src, saved_packet[IP].src)
 166                 self.assertEqual(ip.dst, saved_packet[IP].dst)
 167                 self.assertEqual(udp.payload, saved_packet[UDP].payload)
 168             except Exception:
 169                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
 170                 raise
 171         for index in self._packet_infos:
 172             self.assertTrue(
 173                 index in seen or index in dropped_packet_indexes,
 174                 "Packet with packet_index %d not received" % index,
 175             )
 176
 177     def test_reassembly(self):
 178         """basic reassembly"""
 179
 180         self.pg_enable_capture()
 181         self.src_if.add_stream(self.fragments_200)
 182         self.pg_start()
 183
 184         packets = self.dst_if.get_capture(len(self.pkt_infos))
 185         self.verify_capture(packets)
 186         self.src_if.assert_nothing_captured()
 187
 188         # run it all again to verify correctness
 189         self.pg_enable_capture()
 190         self.src_if.add_stream(self.fragments_200)
 191         self.pg_start()
 192
 193         packets = self.dst_if.get_capture(len(self.pkt_infos))
 194         self.verify_capture(packets)
 195         self.src_if.assert_nothing_captured()
 196
 197     def test_verify_clear_trace_mid_reassembly(self):
 198         """verify clear trace works mid-reassembly"""
 199
 200         self.pg_enable_capture()
 201         self.src_if.add_stream(self.fragments_200[0:-1])
 202         self.pg_start()
 203
 204         self.logger.debug(self.vapi.cli("show trace"))
 205         self.vapi.cli("clear trace")
 206
 207         self.src_if.add_stream(self.fragments_200[-1])
 208         self.pg_start()
 209         packets = self.dst_if.get_capture(len(self.pkt_infos))
 210         self.verify_capture(packets)
 211
 212     def test_reversed(self):
 213         """reverse order reassembly"""
 214
 215         fragments = list(self.fragments_200)
 216         fragments.reverse()
 217
 218         self.pg_enable_capture()
 219         self.src_if.add_stream(fragments)
 220         self.pg_start()
 221
 222         packets = self.dst_if.get_capture(len(self.packet_infos))
 223         self.verify_capture(packets)
 224         self.src_if.assert_nothing_captured()
 225
 226         # run it all again to verify correctness
 227         self.pg_enable_capture()
 228         self.src_if.add_stream(fragments)
 229         self.pg_start()
 230
 231         packets = self.dst_if.get_capture(len(self.packet_infos))
 232         self.verify_capture(packets)
 233         self.src_if.assert_nothing_captured()
 234
 235     def test_long_fragment_chain(self):
 236         """long fragment chain"""
 237
 238         error_cnt_str = "/err/ip4-full-reassembly-feature/reass_fragment_chain_too_long"
 239
 240         error_cnt = self.statistics.get_err_counter(error_cnt_str)
 241
 242         self.vapi.ip_reassembly_set(
 243             timeout_ms=100,
 244             max_reassemblies=1000,
 245             max_reassembly_length=3,
 246             expire_walk_interval_ms=50,
 247         )
 248
 249         p1 = (
 250             Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac)
 251             / IP(id=1000, src=self.src_if.remote_ip4, dst=self.dst_if.remote_ip4)
 252             / UDP(sport=1234, dport=5678)
 253             / Raw(b"X" * 1000)
 254         )
 255         p2 = (
 256             Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac)
 257             / IP(id=1001, src=self.src_if.remote_ip4, dst=self.dst_if.remote_ip4)
 258             / UDP(sport=1234, dport=5678)
 259             / Raw(b"X" * 1000)
 260         )
 261         frags = fragment_rfc791(p1, 200) + fragment_rfc791(p2, 500)
 262
 263         self.pg_enable_capture()
 264         self.src_if.add_stream(frags)
 265         self.pg_start()
 266
 267         self.dst_if.get_capture(1)
 268         self.assert_error_counter_equal(error_cnt_str, error_cnt + 1)
 269
 270     def test_5737(self):
 271         """fragment length + ip header size > 65535"""
 272         self.vapi.cli("clear errors")
 273         raw = b"""E\x00\x00\x88,\xf8\x1f\xfe@\x01\x98\x00\xc0\xa8\n-\xc0\xa8\n\
 274 \x01\x08\x00\xf0J\xed\xcb\xf1\xf5Test-group: IPv4.IPv4.ipv4-message.\
 275 Ethernet-Payload.IPv4-Packet.IPv4-Header.Fragment-Offset; Test-case: 5737"""
 276         malformed_packet = Ether(
 277             dst=self.src_if.local_mac, src=self.src_if.remote_mac
 278         ) / IP(raw)
 279         p = (
 280             Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac)
 281             / IP(id=1000, src=self.src_if.remote_ip4, dst=self.dst_if.remote_ip4)
 282             / UDP(sport=1234, dport=5678)
 283             / Raw(b"X" * 1000)
 284         )
 285         valid_fragments = fragment_rfc791(p, 400)
 286
 287         counter = "/err/ip4-full-reassembly-feature/reass_malformed_packet"
 288         error_counter = self.statistics.get_err_counter(counter)
 289         self.pg_enable_capture()
 290         self.src_if.add_stream([malformed_packet] + valid_fragments)
 291         self.pg_start()
 292
 293         self.dst_if.get_capture(1)
 294         self.logger.debug(self.vapi.ppcli("show error"))
 295         self.assertEqual(self.statistics.get_err_counter(counter), error_counter + 1)
 296
 297     def test_44924(self):
 298         """compress tiny fragments"""
 299         packets = [
 300             (
 301                 Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac)
 302                 / IP(
 303                     id=24339,
 304                     flags="MF",
 305                     frag=0,
 306                     ttl=64,
 307                     src=self.src_if.remote_ip4,
 308                     dst=self.dst_if.remote_ip4,
 309                 )
 310                 / ICMP(type="echo-request", code=0, id=0x1FE6, seq=0x2407)
 311                 / Raw(load="Test-group: IPv4")
 312             ),
 313             (
 314                 Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac)
 315                 / IP(
 316                     id=24339,
 317                     flags="MF",
 318                     frag=3,
 319                     ttl=64,
 320                     src=self.src_if.remote_ip4,
 321                     dst=self.dst_if.remote_ip4,
 322                 )
 323                 / ICMP(type="echo-request", code=0, id=0x1FE6, seq=0x2407)
 324                 / Raw(load=".IPv4.Fragmentation.vali")
 325             ),
 326             (
 327                 Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac)
 328                 / IP(
 329                     id=24339,
 330                     frag=6,
 331                     ttl=64,
 332                     src=self.src_if.remote_ip4,
 333                     dst=self.dst_if.remote_ip4,
 334                 )
 335                 / ICMP(type="echo-request", code=0, id=0x1FE6, seq=0x2407)
 336                 / Raw(load="d; Test-case: 44924")
 337             ),
 338         ]
 339
 340         self.pg_enable_capture()
 341         self.src_if.add_stream(packets)
 342         self.pg_start()
 343
 344         self.dst_if.get_capture(1)
 345
 346     def test_frag_1(self):
 347         """fragment of size 1"""
 348         self.vapi.cli("clear errors")
 349         malformed_packets = [
 350             (
 351                 Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac)
 352                 / IP(
 353                     id=7,
 354                     len=21,
 355                     flags="MF",
 356                     frag=0,
 357                     ttl=64,
 358                     src=self.src_if.remote_ip4,
 359                     dst=self.dst_if.remote_ip4,
 360                 )
 361                 / ICMP(type="echo-request")
 362             ),
 363             (
 364                 Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac)
 365                 / IP(
 366                     id=7,
 367                     len=21,
 368                     frag=1,
 369                     ttl=64,
 370                     src=self.src_if.remote_ip4,
 371                     dst=self.dst_if.remote_ip4,
 372                 )
 373                 / Raw(load=b"\x08")
 374             ),
 375         ]
 376
 377         p = (
 378             Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac)
 379             / IP(id=1000, src=self.src_if.remote_ip4, dst=self.dst_if.remote_ip4)
 380             / UDP(sport=1234, dport=5678)
 381             / Raw(b"X" * 1000)
 382         )
 383         valid_fragments = fragment_rfc791(p, 400)
 384
 385         self.pg_enable_capture()
 386         self.src_if.add_stream(malformed_packets + valid_fragments)
 387         self.pg_start()
 388
 389         self.dst_if.get_capture(1)
 390
 391         self.assert_packet_counter_equal("ip4-full-reassembly-feature", 1)
 392         # TODO remove above, uncomment below once clearing of counters
 393         # is supported
 394         # self.assert_packet_counter_equal(
 395         #     "/err/ip4-full-reassembly-feature/reass_malformed_packet", 1)
 396
 397     def test_random(self):
 398         """random order reassembly"""
 399
 400         fragments = list(self.fragments_200)
 401         shuffle(fragments)
 402
 403         self.pg_enable_capture()
 404         self.src_if.add_stream(fragments)
 405         self.pg_start()
 406
 407         packets = self.dst_if.get_capture(len(self.packet_infos))
 408         self.verify_capture(packets)
 409         self.src_if.assert_nothing_captured()
 410
 411         # run it all again to verify correctness
 412         self.pg_enable_capture()
 413         self.src_if.add_stream(fragments)
 414         self.pg_start()
 415
 416         packets = self.dst_if.get_capture(len(self.packet_infos))
 417         self.verify_capture(packets)
 418         self.src_if.assert_nothing_captured()
 419
 420     def test_duplicates(self):
 421         """duplicate fragments"""
 422
 423         fragments = [
 424             x
 425             for (_, frags, _, _) in self.pkt_infos
 426             for x in frags
 427             for _ in range(0, min(2, len(frags)))
 428         ]
 429
 430         self.pg_enable_capture()
 431         self.src_if.add_stream(fragments)
 432         self.pg_start()
 433
 434         packets = self.dst_if.get_capture(len(self.pkt_infos))
 435         self.verify_capture(packets)
 436         self.src_if.assert_nothing_captured()
 437
 438     def test_overlap1(self):
 439         """overlapping fragments case #1"""
 440
 441         fragments = []
 442         for _, _, frags_300, frags_200 in self.pkt_infos:
 443             if len(frags_300) == 1:
 444                 fragments.extend(frags_300)
 445             else:
 446                 for i, j in zip(frags_200, frags_300):
 447                     fragments.extend(i)
 448                     fragments.extend(j)
 449
 450         self.pg_enable_capture()
 451         self.src_if.add_stream(fragments)
 452         self.pg_start()
 453
 454         packets = self.dst_if.get_capture(len(self.pkt_infos))
 455         self.verify_capture(packets)
 456         self.src_if.assert_nothing_captured()
 457
 458         # run it all to verify correctness
 459         self.pg_enable_capture()
 460         self.src_if.add_stream(fragments)
 461         self.pg_start()
 462
 463         packets = self.dst_if.get_capture(len(self.pkt_infos))
 464         self.verify_capture(packets)
 465         self.src_if.assert_nothing_captured()
 466
 467     def test_overlap2(self):
 468         """overlapping fragments case #2"""
 469
 470         fragments = []
 471         for _, _, frags_300, frags_200 in self.pkt_infos:
 472             if len(frags_300) == 1:
 473                 fragments.extend(frags_300)
 474             else:
 475                 # care must be taken here so that there are no fragments
 476                 # received by vpp after reassembly is finished, otherwise
 477                 # new reassemblies will be started and packet generator will
 478                 # freak out when it detects unfreed buffers
 479                 zipped = zip(frags_300, frags_200)
 480                 for i, j in zipped:
 481                     fragments.extend(i)
 482                     fragments.extend(j)
 483                 fragments.pop()
 484
 485         self.pg_enable_capture()
 486         self.src_if.add_stream(fragments)
 487         self.pg_start()
 488
 489         packets = self.dst_if.get_capture(len(self.pkt_infos))
 490         self.verify_capture(packets)
 491         self.src_if.assert_nothing_captured()
 492
 493         # run it all to verify correctness
 494         self.pg_enable_capture()
 495         self.src_if.add_stream(fragments)
 496         self.pg_start()
 497
 498         packets = self.dst_if.get_capture(len(self.pkt_infos))
 499         self.verify_capture(packets)
 500         self.src_if.assert_nothing_captured()
 501
 502     def test_timeout_inline(self):
 503         """timeout (inline)"""
 504
 505         dropped_packet_indexes = set(
 506             index for (index, frags, _, _) in self.pkt_infos if len(frags) > 1
 507         )
 508
 509         self.vapi.ip_reassembly_set(
 510             timeout_ms=0,
 511             max_reassemblies=1000,
 512             max_reassembly_length=3,
 513             expire_walk_interval_ms=10000,
 514         )
 515
 516         self.pg_enable_capture()
 517         self.src_if.add_stream(self.fragments_400)
 518         self.pg_start()
 519
 520         packets = self.dst_if.get_capture(
 521             len(self.pkt_infos) - len(dropped_packet_indexes)
 522         )
 523         self.verify_capture(packets, dropped_packet_indexes)
 524         self.src_if.assert_nothing_captured()
 525
 526     def test_timeout_cleanup(self):
 527         """timeout (cleanup)"""
 528
 529         # whole packets + fragmented packets sans last fragment
 530         fragments = [
 531             x
 532             for (_, frags_400, _, _) in self.pkt_infos
 533             for x in frags_400[: -1 if len(frags_400) > 1 else None]
 534         ]
 535
 536         # last fragments for fragmented packets
 537         fragments2 = [
 538             frags_400[-1]
 539             for (_, frags_400, _, _) in self.pkt_infos
 540             if len(frags_400) > 1
 541         ]
 542
 543         dropped_packet_indexes = set(
 544             index for (index, frags_400, _, _) in self.pkt_infos if len(frags_400) > 1
 545         )
 546
 547         self.vapi.ip_reassembly_set(
 548             timeout_ms=100,
 549             max_reassemblies=1000,
 550             max_reassembly_length=1000,
 551             expire_walk_interval_ms=50,
 552         )
 553
 554         self.pg_enable_capture()
 555         self.src_if.add_stream(fragments)
 556         self.pg_start()
 557
 558         self.virtual_sleep(0.25, "wait before sending rest of fragments")
 559
 560         self.src_if.add_stream(fragments2)
 561         self.pg_start()
 562
 563         packets = self.dst_if.get_capture(
 564             len(self.pkt_infos) - len(dropped_packet_indexes)
 565         )
 566         self.verify_capture(packets, dropped_packet_indexes)
 567         self.src_if.assert_nothing_captured()
 568
 569     def test_disabled(self):
 570         """reassembly disabled"""
 571
 572         dropped_packet_indexes = set(
 573             index for (index, frags_400, _, _) in self.pkt_infos if len(frags_400) > 1
 574         )
 575
 576         self.vapi.ip_reassembly_set(
 577             timeout_ms=1000,
 578             max_reassemblies=0,
 579             max_reassembly_length=3,
 580             expire_walk_interval_ms=10000,
 581         )
 582
 583         self.pg_enable_capture()
 584         self.src_if.add_stream(self.fragments_400)
 585         self.pg_start()
 586
 587         packets = self.dst_if.get_capture(
 588             len(self.pkt_infos) - len(dropped_packet_indexes)
 589         )
 590         self.verify_capture(packets, dropped_packet_indexes)
 591         self.src_if.assert_nothing_captured()
 592
 593     def test_local_enable_disable(self):
 594         """local reassembly enabled/disable"""
 595         self.vapi.ip_reassembly_enable_disable(
 596             sw_if_index=self.src_if.sw_if_index, enable_ip4=False
 597         )
 598         self.vapi.ip_local_reass_enable_disable(enable_ip4=True)
 599         p = (
 600             Ether(src=self.src_if.remote_mac, dst=self.src_if.local_mac)
 601             / IP(src=self.src_if.remote_ip4, dst=self.src_if.local_ip4)
 602             / ICMP(id=1234, type="echo-request")
 603             / Raw("x" * 1000)
 604         )
 605         frags = fragment_rfc791(p, 400)
 606         r = self.send_and_expect(self.src_if, frags, self.src_if, n_rx=1)[0]
 607         self.assertEqual(1234, r[ICMP].id)
 608         self.assertEqual(icmptypes[r[ICMP].type], "echo-reply")
 609         self.vapi.ip_local_reass_enable_disable()
 610
 611         self.send_and_assert_no_replies(self.src_if, frags)
 612         self.vapi.ip_local_reass_enable_disable(enable_ip4=True)
 613
 614
 615 class TestIPv4SVReassembly(VppTestCase):
 616     """IPv4 Shallow Virtual Reassembly"""
 617
 618     @classmethod
 619     def setUpClass(cls):
 620         super().setUpClass()
 621
 622         cls.create_pg_interfaces([0, 1])
 623         cls.src_if = cls.pg0
 624         cls.dst_if = cls.pg1
 625
 626         # setup all interfaces
 627         for i in cls.pg_interfaces:
 628             i.admin_up()
 629             i.config_ip4()
 630             i.resolve_arp()
 631
 632     def setUp(self):
 633         """Test setup - force timeout on existing reassemblies"""
 634         super().setUp()
 635         self.vapi.ip_reassembly_enable_disable(
 636             sw_if_index=self.src_if.sw_if_index,
 637             enable_ip4=True,
 638             type=VppEnum.vl_api_ip_reass_type_t.IP_REASS_TYPE_SHALLOW_VIRTUAL,
 639         )
 640         self.vapi.ip_reassembly_set(
 641             timeout_ms=0,
 642             max_reassemblies=1000,
 643             max_reassembly_length=1000,
 644             type=VppEnum.vl_api_ip_reass_type_t.IP_REASS_TYPE_SHALLOW_VIRTUAL,
 645             expire_walk_interval_ms=10,
 646         )
 647         self.virtual_sleep(0.25)
 648         self.vapi.ip_reassembly_set(
 649             timeout_ms=1000000,
 650             max_reassemblies=1000,
 651             max_reassembly_length=1000,
 652             type=VppEnum.vl_api_ip_reass_type_t.IP_REASS_TYPE_SHALLOW_VIRTUAL,
 653             expire_walk_interval_ms=10000,
 654         )
 655
 656     def tearDown(self):
 657         super().tearDown()
 658         self.logger.debug(self.vapi.ppcli("show ip4-sv-reassembly details"))
 659         self.logger.debug(self.vapi.ppcli("show buffers"))
 660
 661     def test_basic(self):
 662         """basic reassembly"""
 663         payload_len = 1000
 664         payload = ""
 665         counter = 0
 666         while len(payload) < payload_len:
 667             payload += "%u " % counter
 668             counter += 1
 669
 670         p = (
 671             Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac)
 672             / IP(id=1, src=self.src_if.remote_ip4, dst=self.dst_if.remote_ip4)
 673             / UDP(sport=1234, dport=5678)
 674             / Raw(payload)
 675         )
 676         fragments = fragment_rfc791(p, payload_len / 4)
 677
 678         # send fragment #2 - should be cached inside reassembly
 679         self.pg_enable_capture()
 680         self.src_if.add_stream(fragments[1])
 681         self.pg_start()
 682         self.logger.debug(self.vapi.ppcli("show ip4-sv-reassembly details"))
 683         self.logger.debug(self.vapi.ppcli("show buffers"))
 684         self.logger.debug(self.vapi.ppcli("show trace"))
 685         self.dst_if.assert_nothing_captured()
 686
 687         # send fragment #1 - reassembly is finished now and both fragments
 688         # forwarded
 689         self.pg_enable_capture()
 690         self.src_if.add_stream(fragments[0])
 691         self.pg_start()
 692         self.logger.debug(self.vapi.ppcli("show ip4-sv-reassembly details"))
 693         self.logger.debug(self.vapi.ppcli("show buffers"))
 694         self.logger.debug(self.vapi.ppcli("show trace"))
 695         c = self.dst_if.get_capture(2)
 696         for sent, recvd in zip([fragments[1], fragments[0]], c):
 697             self.assertEqual(sent[IP].src, recvd[IP].src)
 698             self.assertEqual(sent[IP].dst, recvd[IP].dst)
 699             self.assertEqual(sent[Raw].payload, recvd[Raw].payload)
 700
 701         # send rest of fragments - should be immediately forwarded
 702         self.pg_enable_capture()
 703         self.src_if.add_stream(fragments[2:])
 704         self.pg_start()
 705         c = self.dst_if.get_capture(len(fragments[2:]))
 706         for sent, recvd in zip(fragments[2:], c):
 707             self.assertEqual(sent[IP].src, recvd[IP].src)
 708             self.assertEqual(sent[IP].dst, recvd[IP].dst)
 709             self.assertEqual(sent[Raw].payload, recvd[Raw].payload)
 710
 711     def test_verify_clear_trace_mid_reassembly(self):
 712         """verify clear trace works mid-reassembly"""
 713         payload_len = 1000
 714         payload = ""
 715         counter = 0
 716         while len(payload) < payload_len:
 717             payload += "%u " % counter
 718             counter += 1
 719
 720         p = (
 721             Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac)
 722             / IP(id=1, src=self.src_if.remote_ip4, dst=self.dst_if.remote_ip4)
 723             / UDP(sport=1234, dport=5678)
 724             / Raw(payload)
 725         )
 726         fragments = fragment_rfc791(p, payload_len / 4)
 727
 728         self.pg_enable_capture()
 729         self.src_if.add_stream(fragments[1])
 730         self.pg_start()
 731
 732         self.logger.debug(self.vapi.cli("show trace"))
 733         self.vapi.cli("clear trace")
 734
 735         self.pg_enable_capture()
 736         self.src_if.add_stream(fragments[0])
 737         self.pg_start()
 738         self.dst_if.get_capture(2)
 739
 740         self.logger.debug(self.vapi.cli("show trace"))
 741         self.vapi.cli("clear trace")
 742
 743         self.pg_enable_capture()
 744         self.src_if.add_stream(fragments[2:])
 745         self.pg_start()
 746         self.dst_if.get_capture(len(fragments[2:]))
 747
 748     def test_timeout(self):
 749         """reassembly timeout"""
 750         payload_len = 1000
 751         payload = ""
 752         counter = 0
 753         while len(payload) < payload_len:
 754             payload += "%u " % counter
 755             counter += 1
 756
 757         p = (
 758             Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac)
 759             / IP(id=1, src=self.src_if.remote_ip4, dst=self.dst_if.remote_ip4)
 760             / UDP(sport=1234, dport=5678)
 761             / Raw(payload)
 762         )
 763         fragments = fragment_rfc791(p, payload_len / 4)
 764
 765         self.vapi.ip_reassembly_set(
 766             timeout_ms=100,
 767             max_reassemblies=1000,
 768             max_reassembly_length=1000,
 769             expire_walk_interval_ms=50,
 770             type=VppEnum.vl_api_ip_reass_type_t.IP_REASS_TYPE_SHALLOW_VIRTUAL,
 771         )
 772
 773         # send fragments #2 and #1 - should be forwarded
 774         self.pg_enable_capture()
 775         self.src_if.add_stream(fragments[0:2])
 776         self.pg_start()
 777         self.logger.debug(self.vapi.ppcli("show ip4-sv-reassembly details"))
 778         self.logger.debug(self.vapi.ppcli("show buffers"))
 779         self.logger.debug(self.vapi.ppcli("show trace"))
 780         c = self.dst_if.get_capture(2)
 781         for sent, recvd in zip([fragments[1], fragments[0]], c):
 782             self.assertEqual(sent[IP].src, recvd[IP].src)
 783             self.assertEqual(sent[IP].dst, recvd[IP].dst)
 784             self.assertEqual(sent[Raw].payload, recvd[Raw].payload)
 785
 786         # wait for cleanup
 787         self.virtual_sleep(0.25, "wait before sending rest of fragments")
 788
 789         # send rest of fragments - shouldn't be forwarded
 790         self.pg_enable_capture()
 791         self.src_if.add_stream(fragments[2:])
 792         self.pg_start()
 793         self.dst_if.assert_nothing_captured()
 794
 795     def test_lru(self):
 796         """reassembly reuses LRU element"""
 797
 798         self.vapi.ip_reassembly_set(
 799             timeout_ms=1000000,
 800             max_reassemblies=1,
 801             max_reassembly_length=1000,
 802             type=VppEnum.vl_api_ip_reass_type_t.IP_REASS_TYPE_SHALLOW_VIRTUAL,
 803             expire_walk_interval_ms=10000,
 804         )
 805
 806         payload_len = 1000
 807         payload = ""
 808         counter = 0
 809         while len(payload) < payload_len:
 810             payload += "%u " % counter
 811             counter += 1
 812
 813         packet_count = 10
 814
 815         fragments = [
 816             f
 817             for i in range(packet_count)
 818             for p in (
 819                 Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac)
 820                 / IP(id=i, src=self.src_if.remote_ip4, dst=self.dst_if.remote_ip4)
 821                 / UDP(sport=1234, dport=5678)
 822                 / Raw(payload)
 823             )
 824             for f in fragment_rfc791(p, payload_len / 4)
 825         ]
 826
 827         self.pg_enable_capture()
 828         self.src_if.add_stream(fragments)
 829         self.pg_start()
 830         c = self.dst_if.get_capture(len(fragments))
 831         for sent, recvd in zip(fragments, c):
 832             self.assertEqual(sent[IP].src, recvd[IP].src)
 833             self.assertEqual(sent[IP].dst, recvd[IP].dst)
 834             self.assertEqual(sent[Raw].payload, recvd[Raw].payload)
 835
 836     def send_mixed_and_verify_capture(self, traffic):
 837         stream = []
 838         for t in traffic:
 839             for c in range(t["count"]):
 840                 stream.append(
 841                     (
 842                         Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac)
 843                         / IP(
 844                             id=self.counter,
 845                             flags=t["flags"],
 846                             src=self.src_if.remote_ip4,
 847                             dst=self.dst_if.remote_ip4,
 848                         )
 849                         / UDP(sport=1234, dport=5678)
 850                         / Raw("abcdef")
 851                     )
 852                 )
 853                 self.counter = self.counter + 1
 854
 855         self.pg_enable_capture()
 856         self.src_if.add_stream(stream)
 857         self.pg_start()
 858         self.logger.debug(self.vapi.ppcli("show ip4-sv-reassembly details"))
 859         self.logger.debug(self.vapi.ppcli("show buffers"))
 860         self.logger.debug(self.vapi.ppcli("show trace"))
 861         self.dst_if.get_capture(len(stream))
 862
 863     def test_mixed(self):
 864         """mixed traffic correctly passes through SVR"""
 865         self.counter = 1
 866
 867         self.send_mixed_and_verify_capture([{"count": 1, "flags": ""}])
 868         self.send_mixed_and_verify_capture([{"count": 2, "flags": ""}])
 869         self.send_mixed_and_verify_capture([{"count": 3, "flags": ""}])
 870         self.send_mixed_and_verify_capture([{"count": 8, "flags": ""}])
 871         self.send_mixed_and_verify_capture([{"count": 257, "flags": ""}])
 872
 873         self.send_mixed_and_verify_capture([{"count": 1, "flags": "MF"}])
 874         self.send_mixed_and_verify_capture([{"count": 2, "flags": "MF"}])
 875         self.send_mixed_and_verify_capture([{"count": 3, "flags": "MF"}])
 876         self.send_mixed_and_verify_capture([{"count": 8, "flags": "MF"}])
 877         self.send_mixed_and_verify_capture([{"count": 257, "flags": "MF"}])
 878
 879         self.send_mixed_and_verify_capture(
 880             [{"count": 1, "flags": ""}, {"count": 1, "flags": "MF"}]
 881         )
 882         self.send_mixed_and_verify_capture(
 883             [{"count": 2, "flags": ""}, {"count": 2, "flags": "MF"}]
 884         )
 885         self.send_mixed_and_verify_capture(
 886             [{"count": 3, "flags": ""}, {"count": 3, "flags": "MF"}]
 887         )
 888         self.send_mixed_and_verify_capture(
 889             [{"count": 8, "flags": ""}, {"count": 8, "flags": "MF"}]
 890         )
 891         self.send_mixed_and_verify_capture(
 892             [{"count": 129, "flags": ""}, {"count": 129, "flags": "MF"}]
 893         )
 894
 895         self.send_mixed_and_verify_capture(
 896             [
 897                 {"count": 1, "flags": ""},
 898                 {"count": 1, "flags": "MF"},
 899                 {"count": 1, "flags": ""},
 900                 {"count": 1, "flags": "MF"},
 901             ]
 902         )
 903         self.send_mixed_and_verify_capture(
 904             [
 905                 {"count": 2, "flags": ""},
 906                 {"count": 2, "flags": "MF"},
 907                 {"count": 2, "flags": ""},
 908                 {"count": 2, "flags": "MF"},
 909             ]
 910         )
 911         self.send_mixed_and_verify_capture(
 912             [
 913                 {"count": 3, "flags": ""},
 914                 {"count": 3, "flags": "MF"},
 915                 {"count": 3, "flags": ""},
 916                 {"count": 3, "flags": "MF"},
 917             ]
 918         )
 919         self.send_mixed_and_verify_capture(
 920             [
 921                 {"count": 8, "flags": ""},
 922                 {"count": 8, "flags": "MF"},
 923                 {"count": 8, "flags": ""},
 924                 {"count": 8, "flags": "MF"},
 925             ]
 926         )
 927         self.send_mixed_and_verify_capture(
 928             [
 929                 {"count": 65, "flags": ""},
 930                 {"count": 65, "flags": "MF"},
 931                 {"count": 65, "flags": ""},
 932                 {"count": 65, "flags": "MF"},
 933             ]
 934         )
 935
 936
 937 class TestIPv4MWReassembly(VppTestCase):
 938     """IPv4 Reassembly (multiple workers)"""
 939
 940     vpp_worker_count = 3
 941
 942     @classmethod
 943     def setUpClass(cls):
 944         super().setUpClass()
 945
 946         cls.create_pg_interfaces(range(cls.vpp_worker_count + 1))
 947         cls.src_if = cls.pg0
 948         cls.send_ifs = cls.pg_interfaces[:-1]
 949         cls.dst_if = cls.pg_interfaces[-1]
 950
 951         # setup all interfaces
 952         for i in cls.pg_interfaces:
 953             i.admin_up()
 954             i.config_ip4()
 955             i.resolve_arp()
 956
 957         # packets sizes reduced here because we are generating packets without
 958         # Ethernet headers, which are added later (diff fragments go via
 959         # different interfaces)
 960         cls.packet_sizes = [
 961             64 - len(Ether()),
 962             512 - len(Ether()),
 963             1518 - len(Ether()),
 964             9018 - len(Ether()),
 965         ]
 966         cls.padding = " abcdefghijklmn"
 967         cls.create_stream(cls.packet_sizes)
 968         cls.create_fragments()
 969
 970     @classmethod
 971     def tearDownClass(cls):
 972         super().tearDownClass()
 973
 974     def setUp(self):
 975         """Test setup - force timeout on existing reassemblies"""
 976         super().setUp()
 977         for intf in self.send_ifs:
 978             self.vapi.ip_reassembly_enable_disable(
 979                 sw_if_index=intf.sw_if_index, enable_ip4=True
 980             )
 981         self.vapi.ip_reassembly_set(
 982             timeout_ms=0,
 983             max_reassemblies=1000,
 984             max_reassembly_length=1000,
 985             expire_walk_interval_ms=10,
 986         )
 987         self.virtual_sleep(0.25)
 988         self.vapi.ip_reassembly_set(
 989             timeout_ms=1000000,
 990             max_reassemblies=1000,
 991             max_reassembly_length=1000,
 992             expire_walk_interval_ms=10000,
 993         )
 994
 995     def tearDown(self):
 996         for intf in self.send_ifs:
 997             self.vapi.ip_reassembly_enable_disable(
 998                 sw_if_index=intf.sw_if_index, enable_ip4=False
 999             )
1000         super().tearDown()
1001
1002     def show_commands_at_teardown(self):
1003         self.logger.debug(self.vapi.ppcli("show ip4-full-reassembly details"))
1004         self.logger.debug(self.vapi.ppcli("show buffers"))
1005
1006     @classmethod
1007     def create_stream(cls, packet_sizes, packet_count=test_packet_count):
1008         """Create input packet stream
1009
1010         :param list packet_sizes: Required packet sizes.
1011         """
1012         for i in range(0, packet_count):
1013             info = cls.create_packet_info(cls.src_if, cls.src_if)
1014             payload = cls.info_to_payload(info)
1015             p = (
1016                 IP(id=info.index, src=cls.src_if.remote_ip4, dst=cls.dst_if.remote_ip4)
1017                 / UDP(sport=1234, dport=5678)
1018                 / Raw(payload)
1019             )
1020             size = packet_sizes[(i // 2) % len(packet_sizes)]
1021             cls.extend_packet(p, size, cls.padding)
1022             info.data = p
1023
1024     @classmethod
1025     def create_fragments(cls):
1026         infos = cls._packet_infos
1027         cls.pkt_infos = []
1028         for index, info in infos.items():
1029             p = info.data
1030             # cls.logger.debug(ppp("Packet:",
1031             #                      p.__class__(scapy.compat.raw(p))))
1032             fragments_400 = fragment_rfc791(p, 400)
1033             cls.pkt_infos.append((index, fragments_400))
1034         cls.fragments_400 = [x for (_, frags) in cls.pkt_infos for x in frags]
1035         cls.logger.debug(
1036             "Fragmented %s packets into %s 400-byte fragments, "
1037             % (len(infos), len(cls.fragments_400))
1038         )
1039
1040     def verify_capture(self, capture, dropped_packet_indexes=[]):
1041         """Verify captured packet stream.
1042
1043         :param list capture: Captured packet stream.
1044         """
1045         info = None
1046         seen = set()
1047         for packet in capture:
1048             try:
1049                 self.logger.debug(ppp("Got packet:", packet))
1050                 ip = packet[IP]
1051                 udp = packet[UDP]
1052                 payload_info = self.payload_to_info(packet[Raw])
1053                 packet_index = payload_info.index
1054                 self.assertTrue(
1055                     packet_index not in dropped_packet_indexes,
1056                     ppp("Packet received, but should be dropped:", packet),
1057                 )
1058                 if packet_index in seen:
1059                     raise Exception(ppp("Duplicate packet received", packet))
1060                 seen.add(packet_index)
1061                 self.assertEqual(payload_info.dst, self.src_if.sw_if_index)
1062                 info = self._packet_infos[packet_index]
1063                 self.assertTrue(info is not None)
1064                 self.assertEqual(packet_index, info.index)
1065                 saved_packet = info.data
1066                 self.assertEqual(ip.src, saved_packet[IP].src)
1067                 self.assertEqual(ip.dst, saved_packet[IP].dst)
1068                 self.assertEqual(udp.payload, saved_packet[UDP].payload)
1069             except Exception:
1070                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
1071                 raise
1072         for index in self._packet_infos:
1073             self.assertTrue(
1074                 index in seen or index in dropped_packet_indexes,
1075                 "Packet with packet_index %d not received" % index,
1076             )
1077
1078     def send_packets(self, packets):
1079         for counter in range(self.vpp_worker_count):
1080             if 0 == len(packets[counter]):
1081                 continue
1082             send_if = self.send_ifs[counter]
1083             send_if.add_stream(
1084                 (
1085                     Ether(dst=send_if.local_mac, src=send_if.remote_mac) / x
1086                     for x in packets[counter]
1087                 ),
1088                 worker=counter,
1089             )
1090         self.pg_start()
1091
1092     def test_worker_conflict(self):
1093         """1st and FO=0 fragments on different workers"""
1094
1095         # in first wave we send fragments which don't start at offset 0
1096         # then we send fragments with offset 0 on a different thread
1097         # then the rest of packets on a random thread
1098         first_packets = [[] for n in range(self.vpp_worker_count)]
1099         second_packets = [[] for n in range(self.vpp_worker_count)]
1100         rest_of_packets = [[] for n in range(self.vpp_worker_count)]
1101         for (_, p) in self.pkt_infos:
1102             wi = randrange(self.vpp_worker_count)
1103             second_packets[wi].append(p[0])
1104             if len(p) <= 1:
1105                 continue
1106             wi2 = wi
1107             while wi2 == wi:
1108                 wi2 = randrange(self.vpp_worker_count)
1109             first_packets[wi2].append(p[1])
1110             wi3 = randrange(self.vpp_worker_count)
1111             rest_of_packets[wi3].extend(p[2:])
1112
1113         self.pg_enable_capture()
1114         self.send_packets(first_packets)
1115         self.send_packets(second_packets)
1116         self.send_packets(rest_of_packets)
1117
1118         packets = self.dst_if.get_capture(len(self.pkt_infos))
1119         self.verify_capture(packets)
1120         for send_if in self.send_ifs:
1121             send_if.assert_nothing_captured()
1122
1123         self.logger.debug(self.vapi.ppcli("show trace"))
1124         self.logger.debug(self.vapi.ppcli("show ip4-full-reassembly details"))
1125         self.logger.debug(self.vapi.ppcli("show buffers"))
1126         self.vapi.cli("clear trace")
1127
1128         self.pg_enable_capture()
1129         self.send_packets(first_packets)
1130         self.send_packets(second_packets)
1131         self.send_packets(rest_of_packets)
1132
1133         packets = self.dst_if.get_capture(len(self.pkt_infos))
1134         self.verify_capture(packets)
1135         for send_if in self.send_ifs:
1136             send_if.assert_nothing_captured()
1137
1138
1139 class TestIPv6Reassembly(VppTestCase):
1140     """IPv6 Reassembly"""
1141
1142     @classmethod
1143     def setUpClass(cls):
1144         super().setUpClass()
1145
1146         cls.create_pg_interfaces([0, 1])
1147         cls.src_if = cls.pg0
1148         cls.dst_if = cls.pg1
1149
1150         # setup all interfaces
1151         for i in cls.pg_interfaces:
1152             i.admin_up()
1153             i.config_ip6()
1154             i.resolve_ndp()
1155
1156         # packet sizes
1157         cls.packet_sizes = [64, 512, 1518, 9018]
1158         cls.padding = " abcdefghijklmn"
1159         cls.create_stream(cls.packet_sizes)
1160         cls.create_fragments()
1161
1162     @classmethod
1163     def tearDownClass(cls):
1164         super().tearDownClass()
1165
1166     def setUp(self):
1167         """Test setup - force timeout on existing reassemblies"""
1168         super().setUp()
1169         self.vapi.ip_reassembly_enable_disable(
1170             sw_if_index=self.src_if.sw_if_index, enable_ip6=True
1171         )
1172         self.vapi.ip_reassembly_set(
1173             timeout_ms=0,
1174             max_reassemblies=1000,
1175             max_reassembly_length=1000,
1176             expire_walk_interval_ms=10,
1177             is_ip6=1,
1178         )
1179         self.virtual_sleep(0.25)
1180         self.vapi.ip_reassembly_set(
1181             timeout_ms=1000000,
1182             max_reassemblies=1000,
1183             max_reassembly_length=1000,
1184             expire_walk_interval_ms=10000,
1185             is_ip6=1,
1186         )
1187         self.logger.debug(self.vapi.ppcli("show ip6-full-reassembly details"))
1188         self.logger.debug(self.vapi.ppcli("show buffers"))
1189
1190     def tearDown(self):
1191         self.vapi.ip_reassembly_enable_disable(
1192             sw_if_index=self.src_if.sw_if_index, enable_ip6=False
1193         )
1194         super().tearDown()
1195
1196     def show_commands_at_teardown(self):
1197         self.logger.debug(self.vapi.ppcli("show ip6-full-reassembly details"))
1198         self.logger.debug(self.vapi.ppcli("show buffers"))
1199
1200     @classmethod
1201     def create_stream(cls, packet_sizes, packet_count=test_packet_count):
1202         """Create input packet stream for defined interface.
1203
1204         :param list packet_sizes: Required packet sizes.
1205         """
1206         for i in range(0, packet_count):
1207             info = cls.create_packet_info(cls.src_if, cls.src_if)
1208             payload = cls.info_to_payload(info)
1209             p = (
1210                 Ether(dst=cls.src_if.local_mac, src=cls.src_if.remote_mac)
1211                 / IPv6(src=cls.src_if.remote_ip6, dst=cls.dst_if.remote_ip6)
1212                 / UDP(sport=1234, dport=5678)
1213                 / Raw(payload)
1214             )
1215             size = packet_sizes[(i // 2) % len(packet_sizes)]
1216             cls.extend_packet(p, size, cls.padding)
1217             info.data = p
1218
1219     @classmethod
1220     def create_fragments(cls):
1221         infos = cls._packet_infos
1222         cls.pkt_infos = []
1223         for index, info in infos.items():
1224             p = info.data
1225             # cls.logger.debug(ppp("Packet:",
1226             #                      p.__class__(scapy.compat.raw(p))))
1227             fragments_400 = fragment_rfc8200(p, info.index, 400)
1228             fragments_300 = fragment_rfc8200(p, info.index, 300)
1229             cls.pkt_infos.append((index, fragments_400, fragments_300))
1230         cls.fragments_400 = [x for _, frags, _ in cls.pkt_infos for x in frags]
1231         cls.fragments_300 = [x for _, _, frags in cls.pkt_infos for x in frags]
1232         cls.logger.debug(
1233             "Fragmented %s packets into %s 400-byte fragments, "
1234             "and %s 300-byte fragments"
1235             % (len(infos), len(cls.fragments_400), len(cls.fragments_300))
1236         )
1237
1238     def verify_capture(self, capture, dropped_packet_indexes=[]):
1239         """Verify captured packet strea .
1240
1241         :param list capture: Captured packet stream.
1242         """
1243         info = None
1244         seen = set()
1245         for packet in capture:
1246             try:
1247                 self.logger.debug(ppp("Got packet:", packet))
1248                 ip = packet[IPv6]
1249                 udp = packet[UDP]
1250                 payload_info = self.payload_to_info(packet[Raw])
1251                 packet_index = payload_info.index
1252                 self.assertTrue(
1253                     packet_index not in dropped_packet_indexes,
1254                     ppp("Packet received, but should be dropped:", packet),
1255                 )
1256                 if packet_index in seen:
1257                     raise Exception(ppp("Duplicate packet received", packet))
1258                 seen.add(packet_index)
1259                 self.assertEqual(payload_info.dst, self.src_if.sw_if_index)
1260                 info = self._packet_infos[packet_index]
1261                 self.assertTrue(info is not None)
1262                 self.assertEqual(packet_index, info.index)
1263                 saved_packet = info.data
1264                 self.assertEqual(ip.src, saved_packet[IPv6].src)
1265                 self.assertEqual(ip.dst, saved_packet[IPv6].dst)
1266                 self.assertEqual(udp.payload, saved_packet[UDP].payload)
1267             except Exception:
1268                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
1269                 raise
1270         for index in self._packet_infos:
1271             self.assertTrue(
1272                 index in seen or index in dropped_packet_indexes,
1273                 "Packet with packet_index %d not received" % index,
1274             )
1275
1276     def test_reassembly(self):
1277         """basic reassembly"""
1278
1279         self.pg_enable_capture()
1280         self.src_if.add_stream(self.fragments_400)
1281         self.pg_start()
1282
1283         packets = self.dst_if.get_capture(len(self.pkt_infos))
1284         self.verify_capture(packets)
1285         self.src_if.assert_nothing_captured()
1286
1287         # run it all again to verify correctness
1288         self.pg_enable_capture()
1289         self.src_if.add_stream(self.fragments_400)
1290         self.pg_start()
1291
1292         packets = self.dst_if.get_capture(len(self.pkt_infos))
1293         self.verify_capture(packets)
1294         self.src_if.assert_nothing_captured()
1295
1296     def test_buffer_boundary(self):
1297         """fragment header crossing buffer boundary"""
1298
1299         p = (
1300             Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac)
1301             / IPv6(src=self.src_if.remote_ip6, dst=self.src_if.local_ip6)
1302             / IPv6ExtHdrHopByHop(options=[HBHOptUnknown(otype=0xFF, optlen=0)] * 1000)
1303             / IPv6ExtHdrFragment(m=1)
1304             / UDP(sport=1234, dport=5678)
1305             / Raw()
1306         )
1307         self.pg_enable_capture()
1308         self.src_if.add_stream([p])
1309         self.pg_start()
1310         self.src_if.assert_nothing_captured()
1311         self.dst_if.assert_nothing_captured()
1312
1313     def test_verify_clear_trace_mid_reassembly(self):
1314         """verify clear trace works mid-reassembly"""
1315
1316         self.pg_enable_capture()
1317         self.src_if.add_stream(self.fragments_400[0:-1])
1318         self.pg_start()
1319
1320         self.logger.debug(self.vapi.cli("show trace"))
1321         self.vapi.cli("clear trace")
1322
1323         self.src_if.add_stream(self.fragments_400[-1])
1324         self.pg_start()
1325         packets = self.dst_if.get_capture(len(self.pkt_infos))
1326         self.verify_capture(packets)
1327
1328     def test_reversed(self):
1329         """reverse order reassembly"""
1330
1331         fragments = list(self.fragments_400)
1332         fragments.reverse()
1333
1334         self.pg_enable_capture()
1335         self.src_if.add_stream(fragments)
1336         self.pg_start()
1337
1338         packets = self.dst_if.get_capture(len(self.pkt_infos))
1339         self.verify_capture(packets)
1340         self.src_if.assert_nothing_captured()
1341
1342         # run it all again to verify correctness
1343         self.pg_enable_capture()
1344         self.src_if.add_stream(fragments)
1345         self.pg_start()
1346
1347         packets = self.dst_if.get_capture(len(self.pkt_infos))
1348         self.verify_capture(packets)
1349         self.src_if.assert_nothing_captured()
1350
1351     def test_random(self):
1352         """random order reassembly"""
1353
1354         fragments = list(self.fragments_400)
1355         shuffle(fragments)
1356
1357         self.pg_enable_capture()
1358         self.src_if.add_stream(fragments)
1359         self.pg_start()
1360
1361         packets = self.dst_if.get_capture(len(self.pkt_infos))
1362         self.verify_capture(packets)
1363         self.src_if.assert_nothing_captured()
1364
1365         # run it all again to verify correctness
1366         self.pg_enable_capture()
1367         self.src_if.add_stream(fragments)
1368         self.pg_start()
1369
1370         packets = self.dst_if.get_capture(len(self.pkt_infos))
1371         self.verify_capture(packets)
1372         self.src_if.assert_nothing_captured()
1373
1374     def test_duplicates(self):
1375         """duplicate fragments"""
1376
1377         fragments = [
1378             x
1379             for (_, frags, _) in self.pkt_infos
1380             for x in frags
1381             for _ in range(0, min(2, len(frags)))
1382         ]
1383
1384         self.pg_enable_capture()
1385         self.src_if.add_stream(fragments)
1386         self.pg_start()
1387
1388         packets = self.dst_if.get_capture(len(self.pkt_infos))
1389         self.verify_capture(packets)
1390         self.src_if.assert_nothing_captured()
1391
1392     def test_long_fragment_chain(self):
1393         """long fragment chain"""
1394
1395         error_cnt_str = "/err/ip6-full-reassembly-feature/reass_fragment_chain_too_long"
1396
1397         error_cnt = self.statistics.get_err_counter(error_cnt_str)
1398
1399         self.vapi.ip_reassembly_set(
1400             timeout_ms=100,
1401             max_reassemblies=1000,
1402             max_reassembly_length=3,
1403             expire_walk_interval_ms=50,
1404             is_ip6=1,
1405         )
1406
1407         p = (
1408             Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac)
1409             / IPv6(src=self.src_if.remote_ip6, dst=self.dst_if.remote_ip6)
1410             / UDP(sport=1234, dport=5678)
1411             / Raw(b"X" * 1000)
1412         )
1413         frags = fragment_rfc8200(p, 1, 300) + fragment_rfc8200(p, 2, 500)
1414
1415         self.pg_enable_capture()
1416         self.src_if.add_stream(frags)
1417         self.pg_start()
1418
1419         self.dst_if.get_capture(1)
1420         self.assert_error_counter_equal(error_cnt_str, error_cnt + 1)
1421
1422     def test_overlap1(self):
1423         """overlapping fragments case #1"""
1424
1425         fragments = []
1426         for _, frags_400, frags_300 in self.pkt_infos:
1427             if len(frags_300) == 1:
1428                 fragments.extend(frags_400)
1429             else:
1430                 for i, j in zip(frags_300, frags_400):
1431                     fragments.extend(i)
1432                     fragments.extend(j)
1433
1434         dropped_packet_indexes = set(
1435             index for (index, _, frags) in self.pkt_infos if len(frags) > 1
1436         )
1437
1438         self.pg_enable_capture()
1439         self.src_if.add_stream(fragments)
1440         self.pg_start()
1441
1442         packets = self.dst_if.get_capture(
1443             len(self.pkt_infos) - len(dropped_packet_indexes)
1444         )
1445         self.verify_capture(packets, dropped_packet_indexes)
1446         self.src_if.assert_nothing_captured()
1447
1448     def test_overlap2(self):
1449         """overlapping fragments case #2"""
1450
1451         fragments = []
1452         for _, frags_400, frags_300 in self.pkt_infos:
1453             if len(frags_400) == 1:
1454                 fragments.extend(frags_400)
1455             else:
1456                 # care must be taken here so that there are no fragments
1457                 # received by vpp after reassembly is finished, otherwise
1458                 # new reassemblies will be started and packet generator will
1459                 # freak out when it detects unfreed buffers
1460                 zipped = zip(frags_400, frags_300)
1461                 for i, j in zipped:
1462                     fragments.extend(i)
1463                     fragments.extend(j)
1464                 fragments.pop()
1465
1466         dropped_packet_indexes = set(
1467             index for (index, _, frags) in self.pkt_infos if len(frags) > 1
1468         )
1469
1470         self.pg_enable_capture()
1471         self.src_if.add_stream(fragments)
1472         self.pg_start()
1473
1474         packets = self.dst_if.get_capture(
1475             len(self.pkt_infos) - len(dropped_packet_indexes)
1476         )
1477         self.verify_capture(packets, dropped_packet_indexes)
1478         self.src_if.assert_nothing_captured()
1479
1480     def test_timeout_inline(self):
1481         """timeout (inline)"""
1482
1483         dropped_packet_indexes = set(
1484             index for (index, frags, _) in self.pkt_infos if len(frags) > 1
1485         )
1486
1487         self.vapi.ip_reassembly_set(
1488             timeout_ms=0,
1489             max_reassemblies=1000,
1490             max_reassembly_length=3,
1491             expire_walk_interval_ms=10000,
1492             is_ip6=1,
1493         )
1494
1495         self.pg_enable_capture()
1496         self.src_if.add_stream(self.fragments_400)
1497         self.pg_start()
1498
1499         packets = self.dst_if.get_capture(
1500             len(self.pkt_infos) - len(dropped_packet_indexes)
1501         )
1502         self.verify_capture(packets, dropped_packet_indexes)
1503         pkts = self.src_if._get_capture(1)
1504         for icmp in pkts:
1505             self.assertIn(ICMPv6TimeExceeded, icmp)
1506             self.assertIn(IPv6ExtHdrFragment, icmp)
1507             self.assertIn(icmp[IPv6ExtHdrFragment].id, dropped_packet_indexes)
1508             dropped_packet_indexes.remove(icmp[IPv6ExtHdrFragment].id)
1509
1510     def test_timeout_cleanup(self):
1511         """timeout (cleanup)"""
1512
1513         # whole packets + fragmented packets sans last fragment
1514         fragments = [
1515             x
1516             for (_, frags_400, _) in self.pkt_infos
1517             for x in frags_400[: -1 if len(frags_400) > 1 else None]
1518         ]
1519
1520         # last fragments for fragmented packets
1521         fragments2 = [
1522             frags_400[-1] for (_, frags_400, _) in self.pkt_infos if len(frags_400) > 1
1523         ]
1524
1525         dropped_packet_indexes = set(
1526             index for (index, frags_400, _) in self.pkt_infos if len(frags_400) > 1
1527         )
1528
1529         self.vapi.ip_reassembly_set(
1530             timeout_ms=100,
1531             max_reassemblies=1000,
1532             max_reassembly_length=1000,
1533             expire_walk_interval_ms=50,
1534         )
1535
1536         self.vapi.ip_reassembly_set(
1537             timeout_ms=100,
1538             max_reassemblies=1000,
1539             max_reassembly_length=1000,
1540             expire_walk_interval_ms=50,
1541             is_ip6=1,
1542         )
1543
1544         self.pg_enable_capture()
1545         self.src_if.add_stream(fragments)
1546         self.pg_start()
1547
1548         self.virtual_sleep(0.25, "wait before sending rest of fragments")
1549
1550         self.src_if.add_stream(fragments2)
1551         self.pg_start()
1552
1553         packets = self.dst_if.get_capture(
1554             len(self.pkt_infos) - len(dropped_packet_indexes)
1555         )
1556         self.verify_capture(packets, dropped_packet_indexes)
1557         pkts = self.src_if._get_capture(1)
1558         for icmp in pkts:
1559             self.assertIn(ICMPv6TimeExceeded, icmp)
1560             self.assertIn(IPv6ExtHdrFragment, icmp)
1561             self.assertIn(icmp[IPv6ExtHdrFragment].id, dropped_packet_indexes)
1562             dropped_packet_indexes.remove(icmp[IPv6ExtHdrFragment].id)
1563
1564     def test_disabled(self):
1565         """reassembly disabled"""
1566
1567         dropped_packet_indexes = set(
1568             index for (index, frags_400, _) in self.pkt_infos if len(frags_400) > 1
1569         )
1570
1571         self.vapi.ip_reassembly_set(
1572             timeout_ms=1000,
1573             max_reassemblies=0,
1574             max_reassembly_length=3,
1575             expire_walk_interval_ms=10000,
1576             is_ip6=1,
1577         )
1578
1579         self.pg_enable_capture()
1580         self.src_if.add_stream(self.fragments_400)
1581         self.pg_start()
1582
1583         packets = self.dst_if.get_capture(
1584             len(self.pkt_infos) - len(dropped_packet_indexes)
1585         )
1586         self.verify_capture(packets, dropped_packet_indexes)
1587         self.src_if.assert_nothing_captured()
1588
1589     def test_missing_upper(self):
1590         """missing upper layer"""
1591         optdata = "\x00" * 100
1592         p = (
1593             Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac)
1594             / IPv6(src=self.src_if.remote_ip6, dst=self.src_if.local_ip6)
1595             / IPv6ExtHdrFragment(m=1)
1596             / IPv6ExtHdrDestOpt(
1597                 nh=17, options=PadN(optdata="\101" * 255) / PadN(optdata="\102" * 255)
1598             )
1599         )
1600
1601         self.pg_enable_capture()
1602         self.src_if.add_stream([p])
1603         self.pg_start()
1604         pkts = self.src_if.get_capture(expected_count=1)
1605         icmp = pkts[0]
1606         self.assertIn(ICMPv6ParamProblem, icmp)
1607         self.assert_equal(icmp[ICMPv6ParamProblem].code, 3, "ICMP code")
1608
1609     def test_truncated_fragment(self):
1610         """truncated fragment"""
1611         pkt = (
1612             Ether(src=self.pg0.local_mac, dst=self.pg0.remote_mac)
1613             / IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6, nh=44, plen=2)
1614             / IPv6ExtHdrFragment(nh=6)
1615         )
1616
1617         self.send_and_assert_no_replies(self.pg0, [pkt], self.pg0)
1618
1619     def test_invalid_frag_size(self):
1620         """fragment size not a multiple of 8"""
1621         p = (
1622             Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac)
1623             / IPv6(src=self.src_if.remote_ip6, dst=self.src_if.local_ip6)
1624             / UDP(sport=1234, dport=5678)
1625             / Raw()
1626         )
1627         self.extend_packet(p, 1000, self.padding)
1628         fragments = fragment_rfc8200(p, 1, 500)
1629         bad_fragment = fragments[0]
1630         self.extend_packet(bad_fragment, len(bad_fragment) + 5)
1631         self.pg_enable_capture()
1632         self.src_if.add_stream([bad_fragment])
1633         self.pg_start()
1634         pkts = self.src_if.get_capture(expected_count=1)
1635         icmp = pkts[0]
1636         self.assertIn(ICMPv6ParamProblem, icmp)
1637         self.assert_equal(icmp[ICMPv6ParamProblem].code, 0, "ICMP code")
1638
1639     def test_invalid_packet_size(self):
1640         """total packet size > 65535"""
1641         p = (
1642             Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac)
1643             / IPv6(src=self.src_if.remote_ip6, dst=self.src_if.local_ip6)
1644             / UDP(sport=1234, dport=5678)
1645             / Raw()
1646         )
1647         self.extend_packet(p, 1000, self.padding)
1648         fragments = fragment_rfc8200(p, 1, 500)
1649         bad_fragment = fragments[1]
1650         bad_fragment[IPv6ExtHdrFragment].offset = 65500
1651         self.pg_enable_capture()
1652         self.src_if.add_stream([bad_fragment])
1653         self.pg_start()
1654         pkts = self.src_if.get_capture(expected_count=1)
1655         icmp = pkts[0]
1656         self.assertIn(ICMPv6ParamProblem, icmp)
1657         self.assert_equal(icmp[ICMPv6ParamProblem].code, 0, "ICMP code")
1658
1659     def test_atomic_fragment(self):
1660         """IPv6 atomic fragment"""
1661         pkt = (
1662             Ether(src=self.pg0.local_mac, dst=self.pg0.remote_mac)
1663             / IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6, nh=44, plen=65535)
1664             / IPv6ExtHdrFragment(
1665                 offset=8191, m=1, res1=0xFF, res2=0xFF, nh=255, id=0xFFFF
1666             )
1667             / ("X" * 1452)
1668         )
1669
1670         rx = self.send_and_expect(self.pg0, [pkt], self.pg0)
1671         self.assertIn(ICMPv6ParamProblem, rx[0])
1672
1673     def test_truncated_fragment(self):
1674         """IPv6 truncated fragment header"""
1675         pkt = (
1676             Ether(src=self.pg0.local_mac, dst=self.pg0.remote_mac)
1677             / IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6, nh=44, plen=2)
1678             / IPv6ExtHdrFragment(nh=6)
1679         )
1680
1681         self.send_and_assert_no_replies(self.pg0, [pkt])
1682
1683         pkt = (
1684             Ether(src=self.pg0.local_mac, dst=self.pg0.remote_mac)
1685             / IPv6(src=self.pg0.remote_ip6, dst=self.pg0.remote_ip6)
1686             / ICMPv6EchoRequest()
1687         )
1688         rx = self.send_and_expect(self.pg0, [pkt], self.pg0)
1689
1690     def test_one_fragment(self):
1691         """whole packet in one fragment processed independently"""
1692         pkt = (
1693             Ether(src=self.pg0.local_mac, dst=self.pg0.remote_mac)
1694             / IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6)
1695             / ICMPv6EchoRequest()
1696             / Raw("X" * 1600)
1697         )
1698         frags = fragment_rfc8200(pkt, 1, 400)
1699
1700         # send a fragment with known id
1701         self.send_and_assert_no_replies(self.pg0, [frags[0]])
1702
1703         # send an atomic fragment with same id - should be reassembled
1704         pkt = (
1705             Ether(src=self.pg0.local_mac, dst=self.pg0.remote_mac)
1706             / IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6)
1707             / IPv6ExtHdrFragment(id=1)
1708             / ICMPv6EchoRequest()
1709         )
1710         rx = self.send_and_expect(self.pg0, [pkt], self.pg0)
1711         self.assertNotIn(IPv6ExtHdrFragment, rx)
1712
1713         # now finish the original reassembly, this should still be possible
1714         rx = self.send_and_expect(self.pg0, frags[1:], self.pg0, n_rx=1)
1715         self.assertNotIn(IPv6ExtHdrFragment, rx)
1716
1717     def test_bunch_of_fragments(self):
1718         """valid fragments followed by rogue fragments and atomic fragment"""
1719         pkt = (
1720             Ether(src=self.pg0.local_mac, dst=self.pg0.remote_mac)
1721             / IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6)
1722             / ICMPv6EchoRequest()
1723             / Raw("X" * 1600)
1724         )
1725         frags = fragment_rfc8200(pkt, 1, 400)
1726         self.send_and_expect(self.pg0, frags, self.pg0, n_rx=1)
1727
1728         inc_frag = (
1729             Ether(src=self.pg0.local_mac, dst=self.pg0.remote_mac)
1730             / IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6)
1731             / IPv6ExtHdrFragment(id=1, nh=58, offset=608)
1732             / Raw("X" * 308)
1733         )
1734
1735         self.send_and_assert_no_replies(self.pg0, inc_frag * 604)
1736
1737         pkt = (
1738             Ether(src=self.pg0.local_mac, dst=self.pg0.remote_mac)
1739             / IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6)
1740             / IPv6ExtHdrFragment(id=1)
1741             / ICMPv6EchoRequest()
1742         )
1743         rx = self.send_and_expect(self.pg0, [pkt], self.pg0)
1744         self.assertNotIn(IPv6ExtHdrFragment, rx)
1745
1746     def test_local_enable_disable(self):
1747         """local reassembly enabled/disable"""
1748         self.vapi.ip_reassembly_enable_disable(
1749             sw_if_index=self.src_if.sw_if_index, enable_ip6=False
1750         )
1751         self.vapi.ip_local_reass_enable_disable(enable_ip6=True)
1752         pkt = (
1753             Ether(src=self.src_if.local_mac, dst=self.src_if.remote_mac)
1754             / IPv6(src=self.src_if.remote_ip6, dst=self.src_if.local_ip6)
1755             / ICMPv6EchoRequest(id=1234)
1756             / Raw("X" * 1600)
1757         )
1758         frags = fragment_rfc8200(pkt, 1, 400)
1759         r = self.send_and_expect(self.src_if, frags, self.src_if, n_rx=1)[0]
1760         self.assertEqual(1234, r[ICMPv6EchoReply].id)
1761         self.vapi.ip_local_reass_enable_disable()
1762
1763         self.send_and_assert_no_replies(self.src_if, frags)
1764         self.vapi.ip_local_reass_enable_disable(enable_ip6=True)
1765
1766
1767 class TestIPv6MWReassembly(VppTestCase):
1768     """IPv6 Reassembly (multiple workers)"""
1769
1770     vpp_worker_count = 3
1771
1772     @classmethod
1773     def setUpClass(cls):
1774         super().setUpClass()
1775
1776         cls.create_pg_interfaces(range(cls.vpp_worker_count + 1))
1777         cls.src_if = cls.pg0
1778         cls.send_ifs = cls.pg_interfaces[:-1]
1779         cls.dst_if = cls.pg_interfaces[-1]
1780
1781         # setup all interfaces
1782         for i in cls.pg_interfaces:
1783             i.admin_up()
1784             i.config_ip6()
1785             i.resolve_ndp()
1786
1787         # packets sizes reduced here because we are generating packets without
1788         # Ethernet headers, which are added later (diff fragments go via
1789         # different interfaces)
1790         cls.packet_sizes = [
1791             64 - len(Ether()),
1792             512 - len(Ether()),
1793             1518 - len(Ether()),
1794             9018 - len(Ether()),
1795         ]
1796         cls.padding = " abcdefghijklmn"
1797         cls.create_stream(cls.packet_sizes)
1798         cls.create_fragments()
1799
1800     @classmethod
1801     def tearDownClass(cls):
1802         super().tearDownClass()
1803
1804     def setUp(self):
1805         """Test setup - force timeout on existing reassemblies"""
1806         super().setUp()
1807         for intf in self.send_ifs:
1808             self.vapi.ip_reassembly_enable_disable(
1809                 sw_if_index=intf.sw_if_index, enable_ip6=True
1810             )
1811         self.vapi.ip_reassembly_set(
1812             timeout_ms=0,
1813             max_reassemblies=1000,
1814             max_reassembly_length=1000,
1815             expire_walk_interval_ms=10,
1816             is_ip6=1,
1817         )
1818         self.virtual_sleep(0.25)
1819         self.vapi.ip_reassembly_set(
1820             timeout_ms=1000000,
1821             max_reassemblies=1000,
1822             max_reassembly_length=1000,
1823             expire_walk_interval_ms=1000,
1824             is_ip6=1,
1825         )
1826
1827     def tearDown(self):
1828         for intf in self.send_ifs:
1829             self.vapi.ip_reassembly_enable_disable(
1830                 sw_if_index=intf.sw_if_index, enable_ip6=False
1831             )
1832         super().tearDown()
1833
1834     def show_commands_at_teardown(self):
1835         self.logger.debug(self.vapi.ppcli("show ip6-full-reassembly details"))
1836         self.logger.debug(self.vapi.ppcli("show buffers"))
1837
1838     @classmethod
1839     def create_stream(cls, packet_sizes, packet_count=test_packet_count):
1840         """Create input packet stream
1841
1842         :param list packet_sizes: Required packet sizes.
1843         """
1844         for i in range(0, packet_count):
1845             info = cls.create_packet_info(cls.src_if, cls.src_if)
1846             payload = cls.info_to_payload(info)
1847             p = (
1848                 IPv6(src=cls.src_if.remote_ip6, dst=cls.dst_if.remote_ip6)
1849                 / UDP(sport=1234, dport=5678)
1850                 / Raw(payload)
1851             )
1852             size = packet_sizes[(i // 2) % len(packet_sizes)]
1853             cls.extend_packet(p, size, cls.padding)
1854             info.data = p
1855
1856     @classmethod
1857     def create_fragments(cls):
1858         infos = cls._packet_infos
1859         cls.pkt_infos = []
1860         for index, info in infos.items():
1861             p = info.data
1862             # cls.logger.debug(ppp("Packet:",
1863             #                      p.__class__(scapy.compat.raw(p))))
1864             fragments_400 = fragment_rfc8200(p, index, 400)
1865             cls.pkt_infos.append((index, fragments_400))
1866         cls.fragments_400 = [x for (_, frags) in cls.pkt_infos for x in frags]
1867         cls.logger.debug(
1868             "Fragmented %s packets into %s 400-byte fragments, "
1869             % (len(infos), len(cls.fragments_400))
1870         )
1871
1872     def verify_capture(self, capture, dropped_packet_indexes=[]):
1873         """Verify captured packet strea .
1874
1875         :param list capture: Captured packet stream.
1876         """
1877         info = None
1878         seen = set()
1879         for packet in capture:
1880             try:
1881                 self.logger.debug(ppp("Got packet:", packet))
1882                 ip = packet[IPv6]
1883                 udp = packet[UDP]
1884                 payload_info = self.payload_to_info(packet[Raw])
1885                 packet_index = payload_info.index
1886                 self.assertTrue(
1887                     packet_index not in dropped_packet_indexes,
1888                     ppp("Packet received, but should be dropped:", packet),
1889                 )
1890                 if packet_index in seen:
1891                     raise Exception(ppp("Duplicate packet received", packet))
1892                 seen.add(packet_index)
1893                 self.assertEqual(payload_info.dst, self.src_if.sw_if_index)
1894                 info = self._packet_infos[packet_index]
1895                 self.assertTrue(info is not None)
1896                 self.assertEqual(packet_index, info.index)
1897                 saved_packet = info.data
1898                 self.assertEqual(ip.src, saved_packet[IPv6].src)
1899                 self.assertEqual(ip.dst, saved_packet[IPv6].dst)
1900                 self.assertEqual(udp.payload, saved_packet[UDP].payload)
1901             except Exception:
1902                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
1903                 raise
1904         for index in self._packet_infos:
1905             self.assertTrue(
1906                 index in seen or index in dropped_packet_indexes,
1907                 "Packet with packet_index %d not received" % index,
1908             )
1909
1910     def send_packets(self, packets):
1911         for counter in range(self.vpp_worker_count):
1912             if 0 == len(packets[counter]):
1913                 continue
1914             send_if = self.send_ifs[counter]
1915             send_if.add_stream(
1916                 (
1917                     Ether(dst=send_if.local_mac, src=send_if.remote_mac) / x
1918                     for x in packets[counter]
1919                 ),
1920                 worker=counter,
1921             )
1922         self.pg_start()
1923
1924     def test_worker_conflict(self):
1925         """1st and FO=0 fragments on different workers"""
1926
1927         # in first wave we send fragments which don't start at offset 0
1928         # then we send fragments with offset 0 on a different thread
1929         # then the rest of packets on a random thread
1930         first_packets = [[] for n in range(self.vpp_worker_count)]
1931         second_packets = [[] for n in range(self.vpp_worker_count)]
1932         rest_of_packets = [[] for n in range(self.vpp_worker_count)]
1933         for (_, p) in self.pkt_infos:
1934             wi = randrange(self.vpp_worker_count)
1935             second_packets[wi].append(p[0])
1936             if len(p) <= 1:
1937                 continue
1938             wi2 = wi
1939             while wi2 == wi:
1940                 wi2 = randrange(self.vpp_worker_count)
1941             first_packets[wi2].append(p[1])
1942             wi3 = randrange(self.vpp_worker_count)
1943             rest_of_packets[wi3].extend(p[2:])
1944
1945         self.pg_enable_capture()
1946         self.send_packets(first_packets)
1947         self.send_packets(second_packets)
1948         self.send_packets(rest_of_packets)
1949
1950         packets = self.dst_if.get_capture(len(self.pkt_infos))
1951         self.verify_capture(packets)
1952         for send_if in self.send_ifs:
1953             send_if.assert_nothing_captured()
1954
1955         self.logger.debug(self.vapi.ppcli("show trace"))
1956         self.logger.debug(self.vapi.ppcli("show ip6-full-reassembly details"))
1957         self.logger.debug(self.vapi.ppcli("show buffers"))
1958         self.vapi.cli("clear trace")
1959
1960         self.pg_enable_capture()
1961         self.send_packets(first_packets)
1962         self.send_packets(second_packets)
1963         self.send_packets(rest_of_packets)
1964
1965         packets = self.dst_if.get_capture(len(self.pkt_infos))
1966         self.verify_capture(packets)
1967         for send_if in self.send_ifs:
1968             send_if.assert_nothing_captured()
1969
1970
1971 class TestIPv6SVReassembly(VppTestCase):
1972     """IPv6 Shallow Virtual Reassembly"""
1973
1974     @classmethod
1975     def setUpClass(cls):
1976         super().setUpClass()
1977
1978         cls.create_pg_interfaces([0, 1])
1979         cls.src_if = cls.pg0
1980         cls.dst_if = cls.pg1
1981
1982         # setup all interfaces
1983         for i in cls.pg_interfaces:
1984             i.admin_up()
1985             i.config_ip6()
1986             i.resolve_ndp()
1987
1988     def setUp(self):
1989         """Test setup - force timeout on existing reassemblies"""
1990         super().setUp()
1991         self.vapi.ip_reassembly_enable_disable(
1992             sw_if_index=self.src_if.sw_if_index,
1993             enable_ip6=True,
1994             type=VppEnum.vl_api_ip_reass_type_t.IP_REASS_TYPE_SHALLOW_VIRTUAL,
1995         )
1996         self.vapi.ip_reassembly_set(
1997             timeout_ms=0,
1998             max_reassemblies=1000,
1999             max_reassembly_length=1000,
2000             type=VppEnum.vl_api_ip_reass_type_t.IP_REASS_TYPE_SHALLOW_VIRTUAL,
2001             expire_walk_interval_ms=10,
2002             is_ip6=1,
2003         )
2004         self.virtual_sleep(0.25)
2005         self.vapi.ip_reassembly_set(
2006             timeout_ms=1000000,
2007             max_reassemblies=1000,
2008             max_reassembly_length=1000,
2009             type=VppEnum.vl_api_ip_reass_type_t.IP_REASS_TYPE_SHALLOW_VIRTUAL,
2010             expire_walk_interval_ms=10000,
2011             is_ip6=1,
2012         )
2013
2014     def tearDown(self):
2015         super().tearDown()
2016         self.logger.debug(self.vapi.ppcli("show ip6-sv-reassembly details"))
2017         self.logger.debug(self.vapi.ppcli("show buffers"))
2018
2019     def test_basic(self):
2020         """basic reassembly"""
2021         payload_len = 1000
2022         payload = ""
2023         counter = 0
2024         while len(payload) < payload_len:
2025             payload += "%u " % counter
2026             counter += 1
2027
2028         p = (
2029             Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac)
2030             / IPv6(src=self.src_if.remote_ip6, dst=self.dst_if.remote_ip6)
2031             / UDP(sport=1234, dport=5678)
2032             / Raw(payload)
2033         )
2034         fragments = fragment_rfc8200(p, 1, payload_len / 4)
2035
2036         # send fragment #2 - should be cached inside reassembly
2037         self.pg_enable_capture()
2038         self.src_if.add_stream(fragments[1])
2039         self.pg_start()
2040         self.logger.debug(self.vapi.ppcli("show ip6-sv-reassembly details"))
2041         self.logger.debug(self.vapi.ppcli("show buffers"))
2042         self.logger.debug(self.vapi.ppcli("show trace"))
2043         self.dst_if.assert_nothing_captured()
2044
2045         # send fragment #1 - reassembly is finished now and both fragments
2046         # forwarded
2047         self.pg_enable_capture()
2048         self.src_if.add_stream(fragments[0])
2049         self.pg_start()
2050         self.logger.debug(self.vapi.ppcli("show ip6-sv-reassembly details"))
2051         self.logger.debug(self.vapi.ppcli("show buffers"))
2052         self.logger.debug(self.vapi.ppcli("show trace"))
2053         c = self.dst_if.get_capture(2)
2054         for sent, recvd in zip([fragments[1], fragments[0]], c):
2055             self.assertEqual(sent[IPv6].src, recvd[IPv6].src)
2056             self.assertEqual(sent[IPv6].dst, recvd[IPv6].dst)
2057             self.assertEqual(sent[Raw].payload, recvd[Raw].payload)
2058
2059         # send rest of fragments - should be immediately forwarded
2060         self.pg_enable_capture()
2061         self.src_if.add_stream(fragments[2:])
2062         self.pg_start()
2063         c = self.dst_if.get_capture(len(fragments[2:]))
2064         for sent, recvd in zip(fragments[2:], c):
2065             self.assertEqual(sent[IPv6].src, recvd[IPv6].src)
2066             self.assertEqual(sent[IPv6].dst, recvd[IPv6].dst)
2067             self.assertEqual(sent[Raw].payload, recvd[Raw].payload)
2068
2069     def test_verify_clear_trace_mid_reassembly(self):
2070         """verify clear trace works mid-reassembly"""
2071         payload_len = 1000
2072         payload = ""
2073         counter = 0
2074         while len(payload) < payload_len:
2075             payload += "%u " % counter
2076             counter += 1
2077
2078         p = (
2079             Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac)
2080             / IPv6(src=self.src_if.remote_ip6, dst=self.dst_if.remote_ip6)
2081             / UDP(sport=1234, dport=5678)
2082             / Raw(payload)
2083         )
2084         fragments = fragment_rfc8200(p, 1, payload_len / 4)
2085
2086         self.pg_enable_capture()
2087         self.src_if.add_stream(fragments[1])
2088         self.pg_start()
2089
2090         self.logger.debug(self.vapi.cli("show trace"))
2091         self.vapi.cli("clear trace")
2092
2093         self.pg_enable_capture()
2094         self.src_if.add_stream(fragments[0])
2095         self.pg_start()
2096         self.dst_if.get_capture(2)
2097
2098         self.logger.debug(self.vapi.cli("show trace"))
2099         self.vapi.cli("clear trace")
2100
2101         self.pg_enable_capture()
2102         self.src_if.add_stream(fragments[2:])
2103         self.pg_start()
2104         self.dst_if.get_capture(len(fragments[2:]))
2105
2106     def test_timeout(self):
2107         """reassembly timeout"""
2108         payload_len = 1000
2109         payload = ""
2110         counter = 0
2111         while len(payload) < payload_len:
2112             payload += "%u " % counter
2113             counter += 1
2114
2115         p = (
2116             Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac)
2117             / IPv6(src=self.src_if.remote_ip6, dst=self.dst_if.remote_ip6)
2118             / UDP(sport=1234, dport=5678)
2119             / Raw(payload)
2120         )
2121         fragments = fragment_rfc8200(p, 1, payload_len / 4)
2122
2123         self.vapi.ip_reassembly_set(
2124             timeout_ms=100,
2125             max_reassemblies=1000,
2126             max_reassembly_length=1000,
2127             expire_walk_interval_ms=50,
2128             is_ip6=1,
2129             type=VppEnum.vl_api_ip_reass_type_t.IP_REASS_TYPE_SHALLOW_VIRTUAL,
2130         )
2131
2132         # send fragments #2 and #1 - should be forwarded
2133         self.pg_enable_capture()
2134         self.src_if.add_stream(fragments[0:2])
2135         self.pg_start()
2136         self.logger.debug(self.vapi.ppcli("show ip4-sv-reassembly details"))
2137         self.logger.debug(self.vapi.ppcli("show buffers"))
2138         self.logger.debug(self.vapi.ppcli("show trace"))
2139         c = self.dst_if.get_capture(2)
2140         for sent, recvd in zip([fragments[1], fragments[0]], c):
2141             self.assertEqual(sent[IPv6].src, recvd[IPv6].src)
2142             self.assertEqual(sent[IPv6].dst, recvd[IPv6].dst)
2143             self.assertEqual(sent[Raw].payload, recvd[Raw].payload)
2144
2145         # wait for cleanup
2146         self.virtual_sleep(0.25, "wait before sending rest of fragments")
2147
2148         # send rest of fragments - shouldn't be forwarded
2149         self.pg_enable_capture()
2150         self.src_if.add_stream(fragments[2:])
2151         self.pg_start()
2152         self.dst_if.assert_nothing_captured()
2153
2154     def test_lru(self):
2155         """reassembly reuses LRU element"""
2156
2157         self.vapi.ip_reassembly_set(
2158             timeout_ms=1000000,
2159             max_reassemblies=1,
2160             max_reassembly_length=1000,
2161             type=VppEnum.vl_api_ip_reass_type_t.IP_REASS_TYPE_SHALLOW_VIRTUAL,
2162             is_ip6=1,
2163             expire_walk_interval_ms=10000,
2164         )
2165
2166         payload_len = 1000
2167         payload = ""
2168         counter = 0
2169         while len(payload) < payload_len:
2170             payload += "%u " % counter
2171             counter += 1
2172
2173         packet_count = 10
2174
2175         fragments = [
2176             f
2177             for i in range(packet_count)
2178             for p in (
2179                 Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac)
2180                 / IPv6(src=self.src_if.remote_ip6, dst=self.dst_if.remote_ip6)
2181                 / UDP(sport=1234, dport=5678)
2182                 / Raw(payload)
2183             )
2184             for f in fragment_rfc8200(p, i, payload_len / 4)
2185         ]
2186
2187         self.pg_enable_capture()
2188         self.src_if.add_stream(fragments)
2189         self.pg_start()
2190         c = self.dst_if.get_capture(len(fragments))
2191         for sent, recvd in zip(fragments, c):
2192             self.assertEqual(sent[IPv6].src, recvd[IPv6].src)
2193             self.assertEqual(sent[IPv6].dst, recvd[IPv6].dst)
2194             self.assertEqual(sent[Raw].payload, recvd[Raw].payload)
2195
2196     def test_one_fragment(self):
2197         """whole packet in one fragment processed independently"""
2198         pkt = (
2199             Ether(src=self.src_if.local_mac, dst=self.src_if.remote_mac)
2200             / IPv6(src=self.src_if.remote_ip6, dst=self.dst_if.remote_ip6)
2201             / ICMPv6EchoRequest()
2202             / Raw("X" * 1600)
2203         )
2204         frags = fragment_rfc8200(pkt, 1, 400)
2205
2206         # send a fragment with known id
2207         self.send_and_expect(self.src_if, [frags[0]], self.dst_if)
2208
2209         # send an atomic fragment with same id - should be reassembled
2210         pkt = (
2211             Ether(src=self.src_if.local_mac, dst=self.src_if.remote_mac)
2212             / IPv6(src=self.src_if.remote_ip6, dst=self.dst_if.remote_ip6)
2213             / IPv6ExtHdrFragment(id=1)
2214             / ICMPv6EchoRequest()
2215         )
2216         rx = self.send_and_expect(self.src_if, [pkt], self.dst_if)
2217
2218         # now forward packets matching original reassembly, should still work
2219         rx = self.send_and_expect(self.src_if, frags[1:], self.dst_if)
2220
2221     def test_bunch_of_fragments(self):
2222         """valid fragments followed by rogue fragments and atomic fragment"""
2223         pkt = (
2224             Ether(src=self.src_if.local_mac, dst=self.src_if.remote_mac)
2225             / IPv6(src=self.src_if.remote_ip6, dst=self.dst_if.remote_ip6)
2226             / ICMPv6EchoRequest()
2227             / Raw("X" * 1600)
2228         )
2229         frags = fragment_rfc8200(pkt, 1, 400)
2230         rx = self.send_and_expect(self.src_if, frags, self.dst_if)
2231
2232         rogue = (
2233             Ether(src=self.src_if.local_mac, dst=self.src_if.remote_mac)
2234             / IPv6(src=self.src_if.remote_ip6, dst=self.dst_if.remote_ip6)
2235             / IPv6ExtHdrFragment(id=1, nh=58, offset=608)
2236             / Raw("X" * 308)
2237         )
2238
2239         self.send_and_expect(self.src_if, rogue * 604, self.dst_if)
2240
2241         pkt = (
2242             Ether(src=self.src_if.local_mac, dst=self.src_if.remote_mac)
2243             / IPv6(src=self.src_if.remote_ip6, dst=self.dst_if.remote_ip6)
2244             / IPv6ExtHdrFragment(id=1)
2245             / ICMPv6EchoRequest()
2246         )
2247         rx = self.send_and_expect(self.src_if, [pkt], self.dst_if)
2248
2249     def test_truncated_fragment(self):
2250         """truncated fragment"""
2251         pkt = (
2252             Ether(src=self.pg0.local_mac, dst=self.pg0.remote_mac)
2253             / IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6, nh=44, plen=2)
2254             / IPv6ExtHdrFragment(nh=6)
2255         )
2256
2257         self.send_and_assert_no_replies(self.pg0, [pkt], self.pg0)
2258
2259
2260 class TestIPv4ReassemblyLocalNode(VppTestCase):
2261     """IPv4 Reassembly for packets coming to ip4-local node"""
2262
2263     @classmethod
2264     def setUpClass(cls):
2265         super().setUpClass()
2266
2267         cls.create_pg_interfaces([0])
2268         cls.src_dst_if = cls.pg0
2269
2270         # setup all interfaces
2271         for i in cls.pg_interfaces:
2272             i.admin_up()
2273             i.config_ip4()
2274             i.resolve_arp()
2275
2276         cls.padding = " abcdefghijklmn"
2277         cls.create_stream()
2278         cls.create_fragments()
2279
2280     @classmethod
2281     def tearDownClass(cls):
2282         super().tearDownClass()
2283
2284     def setUp(self):
2285         """Test setup - force timeout on existing reassemblies"""
2286         super().setUp()
2287         self.vapi.ip_reassembly_set(
2288             timeout_ms=0,
2289             max_reassemblies=1000,
2290             max_reassembly_length=1000,
2291             expire_walk_interval_ms=10,
2292         )
2293         self.virtual_sleep(0.25)
2294         self.vapi.ip_reassembly_set(
2295             timeout_ms=1000000,
2296             max_reassemblies=1000,
2297             max_reassembly_length=1000,
2298             expire_walk_interval_ms=10000,
2299         )
2300
2301     def tearDown(self):
2302         super().tearDown()
2303
2304     def show_commands_at_teardown(self):
2305         self.logger.debug(self.vapi.ppcli("show ip4-full-reassembly details"))
2306         self.logger.debug(self.vapi.ppcli("show buffers"))
2307
2308     @classmethod
2309     def create_stream(cls, packet_count=test_packet_count):
2310         """Create input packet stream for defined interface.
2311
2312         :param list packet_sizes: Required packet sizes.
2313         """
2314         for i in range(0, packet_count):
2315             info = cls.create_packet_info(cls.src_dst_if, cls.src_dst_if)
2316             payload = cls.info_to_payload(info)
2317             p = (
2318                 Ether(dst=cls.src_dst_if.local_mac, src=cls.src_dst_if.remote_mac)
2319                 / IP(
2320                     id=info.index,
2321                     src=cls.src_dst_if.remote_ip4,
2322                     dst=cls.src_dst_if.local_ip4,
2323                 )
2324                 / ICMP(type="echo-request", id=1234)
2325                 / Raw(payload)
2326             )
2327             cls.extend_packet(p, 1518, cls.padding)
2328             info.data = p
2329
2330     @classmethod
2331     def create_fragments(cls):
2332         infos = cls._packet_infos
2333         cls.pkt_infos = []
2334         for index, info in infos.items():
2335             p = info.data
2336             # cls.logger.debug(ppp("Packet:",
2337             #                      p.__class__(scapy.compat.raw(p))))
2338             fragments_300 = fragment_rfc791(p, 300)
2339             cls.pkt_infos.append((index, fragments_300))
2340         cls.fragments_300 = [x for (_, frags) in cls.pkt_infos for x in frags]
2341         cls.logger.debug(
2342             "Fragmented %s packets into %s 300-byte fragments"
2343             % (len(infos), len(cls.fragments_300))
2344         )
2345
2346     def verify_capture(self, capture):
2347         """Verify captured packet stream.
2348
2349         :param list capture: Captured packet stream.
2350         """
2351         info = None
2352         seen = set()
2353         for packet in capture:
2354             try:
2355                 self.logger.debug(ppp("Got packet:", packet))
2356                 ip = packet[IP]
2357                 icmp = packet[ICMP]
2358                 payload_info = self.payload_to_info(packet[Raw])
2359                 packet_index = payload_info.index
2360                 if packet_index in seen:
2361                     raise Exception(ppp("Duplicate packet received", packet))
2362                 seen.add(packet_index)
2363                 self.assertEqual(payload_info.dst, self.src_dst_if.sw_if_index)
2364                 info = self._packet_infos[packet_index]
2365                 self.assertIsNotNone(info)
2366                 self.assertEqual(packet_index, info.index)
2367                 saved_packet = info.data
2368                 self.assertEqual(ip.src, saved_packet[IP].dst)
2369                 self.assertEqual(ip.dst, saved_packet[IP].src)
2370                 self.assertEqual(icmp.type, 0)  # echo reply
2371                 self.assertEqual(icmp.id, saved_packet[ICMP].id)
2372                 self.assertEqual(icmp.payload, saved_packet[ICMP].payload)
2373             except Exception:
2374                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
2375                 raise
2376         for index in self._packet_infos:
2377             self.assertIn(
2378                 index, seen, "Packet with packet_index %d not received" % index
2379             )
2380
2381     def test_reassembly(self):
2382         """basic reassembly"""
2383
2384         self.pg_enable_capture()
2385         self.src_dst_if.add_stream(self.fragments_300)
2386         self.pg_start()
2387
2388         packets = self.src_dst_if.get_capture(len(self.pkt_infos))
2389         self.verify_capture(packets)
2390
2391         # run it all again to verify correctness
2392         self.pg_enable_capture()
2393         self.src_dst_if.add_stream(self.fragments_300)
2394         self.pg_start()
2395
2396         packets = self.src_dst_if.get_capture(len(self.pkt_infos))
2397         self.verify_capture(packets)
2398
2399
2400 class TestFIFReassembly(VppTestCase):
2401     """Fragments in fragments reassembly"""
2402
2403     @classmethod
2404     def setUpClass(cls):
2405         super().setUpClass()
2406
2407         cls.create_pg_interfaces([0, 1])
2408         cls.src_if = cls.pg0
2409         cls.dst_if = cls.pg1
2410         for i in cls.pg_interfaces:
2411             i.admin_up()
2412             i.config_ip4()
2413             i.resolve_arp()
2414             i.config_ip6()
2415             i.resolve_ndp()
2416
2417         cls.packet_sizes = [64, 512, 1518, 9018]
2418         cls.padding = " abcdefghijklmn"
2419
2420     @classmethod
2421     def tearDownClass(cls):
2422         super().tearDownClass()
2423
2424     def setUp(self):
2425         """Test setup - force timeout on existing reassemblies"""
2426         super().setUp()
2427         self.vapi.ip_reassembly_enable_disable(
2428             sw_if_index=self.src_if.sw_if_index, enable_ip4=True, enable_ip6=True
2429         )
2430         self.vapi.ip_reassembly_enable_disable(
2431             sw_if_index=self.dst_if.sw_if_index, enable_ip4=True, enable_ip6=True
2432         )
2433         self.vapi.ip_reassembly_set(
2434             timeout_ms=0,
2435             max_reassemblies=1000,
2436             max_reassembly_length=1000,
2437             expire_walk_interval_ms=10,
2438         )
2439         self.vapi.ip_reassembly_set(
2440             timeout_ms=0,
2441             max_reassemblies=1000,
2442             max_reassembly_length=1000,
2443             expire_walk_interval_ms=10,
2444             is_ip6=1,
2445         )
2446         self.virtual_sleep(0.25)
2447         self.vapi.ip_reassembly_set(
2448             timeout_ms=1000000,
2449             max_reassemblies=1000,
2450             max_reassembly_length=1000,
2451             expire_walk_interval_ms=10000,
2452         )
2453         self.vapi.ip_reassembly_set(
2454             timeout_ms=1000000,
2455             max_reassemblies=1000,
2456             max_reassembly_length=1000,
2457             expire_walk_interval_ms=10000,
2458             is_ip6=1,
2459         )
2460
2461     def tearDown(self):
2462         super().tearDown()
2463
2464     def show_commands_at_teardown(self):
2465         self.logger.debug(self.vapi.ppcli("show ip4-full-reassembly details"))
2466         self.logger.debug(self.vapi.ppcli("show ip6-full-reassembly details"))
2467         self.logger.debug(self.vapi.ppcli("show buffers"))
2468
2469     def verify_capture(self, capture, ip_class, dropped_packet_indexes=[]):
2470         """Verify captured packet stream.
2471
2472         :param list capture: Captured packet stream.
2473         """
2474         info = None
2475         seen = set()
2476         for packet in capture:
2477             try:
2478                 self.logger.debug(ppp("Got packet:", packet))
2479                 ip = packet[ip_class]
2480                 udp = packet[UDP]
2481                 payload_info = self.payload_to_info(packet[Raw])
2482                 packet_index = payload_info.index
2483                 self.assertTrue(
2484                     packet_index not in dropped_packet_indexes,
2485                     ppp("Packet received, but should be dropped:", packet),
2486                 )
2487                 if packet_index in seen:
2488                     raise Exception(ppp("Duplicate packet received", packet))
2489                 seen.add(packet_index)
2490                 self.assertEqual(payload_info.dst, self.dst_if.sw_if_index)
2491                 info = self._packet_infos[packet_index]
2492                 self.assertTrue(info is not None)
2493                 self.assertEqual(packet_index, info.index)
2494                 saved_packet = info.data
2495                 self.assertEqual(ip.src, saved_packet[ip_class].src)
2496                 self.assertEqual(ip.dst, saved_packet[ip_class].dst)
2497                 self.assertEqual(udp.payload, saved_packet[UDP].payload)
2498             except Exception:
2499                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
2500                 raise
2501         for index in self._packet_infos:
2502             self.assertTrue(
2503                 index in seen or index in dropped_packet_indexes,
2504                 "Packet with packet_index %d not received" % index,
2505             )
2506
2507     def test_fif4(self):
2508         """Fragments in fragments (4o4)"""
2509
2510         # TODO this should be ideally in setUpClass, but then we hit a bug
2511         # with VppIpRoute incorrectly reporting it's present when it's not
2512         # so we need to manually remove the vpp config, thus we cannot have
2513         # it shared for multiple test cases
2514         self.tun_ip4 = "1.1.1.2"
2515
2516         self.gre4 = VppGreInterface(self, self.src_if.local_ip4, self.tun_ip4)
2517         self.gre4.add_vpp_config()
2518         self.gre4.admin_up()
2519         self.gre4.config_ip4()
2520
2521         self.vapi.ip_reassembly_enable_disable(
2522             sw_if_index=self.gre4.sw_if_index, enable_ip4=True
2523         )
2524
2525         self.route4 = VppIpRoute(
2526             self,
2527             self.tun_ip4,
2528             32,
2529             [VppRoutePath(self.src_if.remote_ip4, self.src_if.sw_if_index)],
2530         )
2531         self.route4.add_vpp_config()
2532
2533         self.reset_packet_infos()
2534         for i in range(test_packet_count):
2535             info = self.create_packet_info(self.src_if, self.dst_if)
2536             payload = self.info_to_payload(info)
2537             # Ethernet header here is only for size calculation, thus it
2538             # doesn't matter how it's initialized. This is to ensure that
2539             # reassembled packet is not > 9000 bytes, so that it's not dropped
2540             p = (
2541                 Ether()
2542                 / IP(id=i, src=self.src_if.remote_ip4, dst=self.dst_if.remote_ip4)
2543                 / UDP(sport=1234, dport=5678)
2544                 / Raw(payload)
2545             )
2546             size = self.packet_sizes[(i // 2) % len(self.packet_sizes)]
2547             self.extend_packet(p, size, self.padding)
2548             info.data = p[IP]  # use only IP part, without ethernet header
2549
2550         fragments = [
2551             x
2552             for _, p in self._packet_infos.items()
2553             for x in fragment_rfc791(p.data, 400)
2554         ]
2555
2556         encapped_fragments = [
2557             Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac)
2558             / IP(src=self.tun_ip4, dst=self.src_if.local_ip4)
2559             / GRE()
2560             / p
2561             for p in fragments
2562         ]
2563
2564         fragmented_encapped_fragments = [
2565             x for p in encapped_fragments for x in fragment_rfc791(p, 200)
2566         ]
2567
2568         self.src_if.add_stream(fragmented_encapped_fragments)
2569
2570         self.pg_enable_capture(self.pg_interfaces)
2571         self.pg_start()
2572
2573         self.src_if.assert_nothing_captured()
2574         packets = self.dst_if.get_capture(len(self._packet_infos))
2575         self.verify_capture(packets, IP)
2576
2577         # TODO remove gre vpp config by hand until VppIpRoute gets fixed
2578         # so that it's query_vpp_config() works as it should
2579         self.gre4.remove_vpp_config()
2580         self.logger.debug(self.vapi.ppcli("show interface"))
2581
2582     def test_fif6(self):
2583         """Fragments in fragments (6o6)"""
2584         # TODO this should be ideally in setUpClass, but then we hit a bug
2585         # with VppIpRoute incorrectly reporting it's present when it's not
2586         # so we need to manually remove the vpp config, thus we cannot have
2587         # it shared for multiple test cases
2588         self.tun_ip6 = "1002::1"
2589
2590         self.gre6 = VppGreInterface(self, self.src_if.local_ip6, self.tun_ip6)
2591         self.gre6.add_vpp_config()
2592         self.gre6.admin_up()
2593         self.gre6.config_ip6()
2594
2595         self.vapi.ip_reassembly_enable_disable(
2596             sw_if_index=self.gre6.sw_if_index, enable_ip6=True
2597         )
2598
2599         self.route6 = VppIpRoute(
2600             self,
2601             self.tun_ip6,
2602             128,
2603             [VppRoutePath(self.src_if.remote_ip6, self.src_if.sw_if_index)],
2604         )
2605         self.route6.add_vpp_config()
2606
2607         self.reset_packet_infos()
2608         for i in range(test_packet_count):
2609             info = self.create_packet_info(self.src_if, self.dst_if)
2610             payload = self.info_to_payload(info)
2611             # Ethernet header here is only for size calculation, thus it
2612             # doesn't matter how it's initialized. This is to ensure that
2613             # reassembled packet is not > 9000 bytes, so that it's not dropped
2614             p = (
2615                 Ether()
2616                 / IPv6(src=self.src_if.remote_ip6, dst=self.dst_if.remote_ip6)
2617                 / UDP(sport=1234, dport=5678)
2618                 / Raw(payload)
2619             )
2620             size = self.packet_sizes[(i // 2) % len(self.packet_sizes)]
2621             self.extend_packet(p, size, self.padding)
2622             info.data = p[IPv6]  # use only IPv6 part, without ethernet header
2623
2624         fragments = [
2625             x
2626             for _, i in self._packet_infos.items()
2627             for x in fragment_rfc8200(i.data, i.index, 400)
2628         ]
2629
2630         encapped_fragments = [
2631             Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac)
2632             / IPv6(src=self.tun_ip6, dst=self.src_if.local_ip6)
2633             / GRE()
2634             / p
2635             for p in fragments
2636         ]
2637
2638         fragmented_encapped_fragments = [
2639             x
2640             for p in encapped_fragments
2641             for x in (
2642                 fragment_rfc8200(
2643                     p, 2 * len(self._packet_infos) + p[IPv6ExtHdrFragment].id, 200
2644                 )
2645                 if IPv6ExtHdrFragment in p
2646                 else [p]
2647             )
2648         ]
2649
2650         self.src_if.add_stream(fragmented_encapped_fragments)
2651
2652         self.pg_enable_capture(self.pg_interfaces)
2653         self.pg_start()
2654
2655         self.src_if.assert_nothing_captured()
2656         packets = self.dst_if.get_capture(len(self._packet_infos))
2657         self.verify_capture(packets, IPv6)
2658
2659         # TODO remove gre vpp config by hand until VppIpRoute gets fixed
2660         # so that it's query_vpp_config() works as it should
2661         self.gre6.remove_vpp_config()
2662
2663
2664 if __name__ == "__main__":
2665     unittest.main(testRunner=VppTestRunner)