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