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
  12 from scapy.layers.inet6 import HBHOptUnknown, ICMPv6ParamProblem,\
  13     ICMPv6TimeExceeded, IPv6, IPv6ExtHdrFragment, IPv6ExtHdrHopByHop
  14 from framework import VppTestCase, VppTestRunner
  15 from util import ppp, ppc, fragment_rfc791, fragment_rfc8200
  16 from vpp_gre_interface import VppGreInterface
  17 from vpp_ip import DpoProto
  18 from vpp_ip_route import VppIpRoute, VppRoutePath, FibPathProto
  19 from vpp_papi import VppEnum
  20
  21 # 35 is enough to have >257 400-byte fragments
  22 test_packet_count = 35
  23
  24 # number of workers used for multi-worker test cases
  25 worker_count = 3
  26
  27
  28 class TestIPv4Reassembly(VppTestCase):
  29     """ IPv4 Reassembly """
  30
  31     @classmethod
  32     def setUpClass(cls):
  33         super(TestIPv4Reassembly, cls).setUpClass()
  34
  35         cls.create_pg_interfaces([0, 1])
  36         cls.src_if = cls.pg0
  37         cls.dst_if = cls.pg1
  38
  39         # setup all interfaces
  40         for i in cls.pg_interfaces:
  41             i.admin_up()
  42             i.config_ip4()
  43             i.resolve_arp()
  44
  45         # packet sizes
  46         cls.packet_sizes = [64, 512, 1518, 9018]
  47         cls.padding = " abcdefghijklmn"
  48         cls.create_stream(cls.packet_sizes)
  49         cls.create_fragments()
  50
  51     @classmethod
  52     def tearDownClass(cls):
  53         super(TestIPv4Reassembly, cls).tearDownClass()
  54
  55     def setUp(self):
  56         """ Test setup - force timeout on existing reassemblies """
  57         super(TestIPv4Reassembly, self).setUp()
  58         self.vapi.ip_reassembly_enable_disable(
  59             sw_if_index=self.src_if.sw_if_index, enable_ip4=True)
  60         self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000,
  61                                     max_reassembly_length=1000,
  62                                     expire_walk_interval_ms=10)
  63         self.sleep(.25)
  64         self.vapi.ip_reassembly_set(timeout_ms=1000000, max_reassemblies=1000,
  65                                     max_reassembly_length=1000,
  66                                     expire_walk_interval_ms=10000)
  67
  68     def tearDown(self):
  69         super(TestIPv4Reassembly, self).tearDown()
  70
  71     def show_commands_at_teardown(self):
  72         self.logger.debug(self.vapi.ppcli("show ip4-full-reassembly details"))
  73         self.logger.debug(self.vapi.ppcli("show buffers"))
  74
  75     @classmethod
  76     def create_stream(cls, packet_sizes, packet_count=test_packet_count):
  77         """Create input packet stream
  78
  79         :param list packet_sizes: Required packet sizes.
  80         """
  81         for i in range(0, packet_count):
  82             info = cls.create_packet_info(cls.src_if, cls.src_if)
  83             payload = cls.info_to_payload(info)
  84             p = (Ether(dst=cls.src_if.local_mac, src=cls.src_if.remote_mac) /
  85                  IP(id=info.index, src=cls.src_if.remote_ip4,
  86                     dst=cls.dst_if.remote_ip4) /
  87                  UDP(sport=1234, dport=5678) /
  88                  Raw(payload))
  89             size = packet_sizes[(i // 2) % len(packet_sizes)]
  90             cls.extend_packet(p, size, cls.padding)
  91             info.data = p
  92
  93     @classmethod
  94     def create_fragments(cls):
  95         infos = cls._packet_infos
  96         cls.pkt_infos = []
  97         for index, info in infos.items():
  98             p = info.data
  99             # cls.logger.debug(ppp("Packet:",
 100             #                      p.__class__(scapy.compat.raw(p))))
 101             fragments_400 = fragment_rfc791(p, 400)
 102             fragments_300 = fragment_rfc791(p, 300)
 103             fragments_200 = [
 104                 x for f in fragments_400 for x in fragment_rfc791(f, 200)]
 105             cls.pkt_infos.append(
 106                 (index, fragments_400, fragments_300, fragments_200))
 107         cls.fragments_400 = [
 108             x for (_, frags, _, _) in cls.pkt_infos for x in frags]
 109         cls.fragments_300 = [
 110             x for (_, _, frags, _) in cls.pkt_infos for x in frags]
 111         cls.fragments_200 = [
 112             x for (_, _, _, frags) in cls.pkt_infos for x in frags]
 113         cls.logger.debug("Fragmented %s packets into %s 400-byte fragments, "
 114                          "%s 300-byte fragments and %s 200-byte fragments" %
 115                          (len(infos), len(cls.fragments_400),
 116                              len(cls.fragments_300), len(cls.fragments_200)))
 117
 118     def verify_capture(self, capture, dropped_packet_indexes=[]):
 119         """Verify captured packet stream.
 120
 121         :param list capture: Captured packet stream.
 122         """
 123         info = None
 124         seen = set()
 125         for packet in capture:
 126             try:
 127                 self.logger.debug(ppp("Got packet:", packet))
 128                 ip = packet[IP]
 129                 udp = packet[UDP]
 130                 payload_info = self.payload_to_info(packet[Raw])
 131                 packet_index = payload_info.index
 132                 self.assertTrue(
 133                     packet_index not in dropped_packet_indexes,
 134                     ppp("Packet received, but should be dropped:", packet))
 135                 if packet_index in seen:
 136                     raise Exception(ppp("Duplicate packet received", packet))
 137                 seen.add(packet_index)
 138                 self.assertEqual(payload_info.dst, self.src_if.sw_if_index)
 139                 info = self._packet_infos[packet_index]
 140                 self.assertTrue(info is not None)
 141                 self.assertEqual(packet_index, info.index)
 142                 saved_packet = info.data
 143                 self.assertEqual(ip.src, saved_packet[IP].src)
 144                 self.assertEqual(ip.dst, saved_packet[IP].dst)
 145                 self.assertEqual(udp.payload, saved_packet[UDP].payload)
 146             except Exception:
 147                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
 148                 raise
 149         for index in self._packet_infos:
 150             self.assertTrue(index in seen or index in dropped_packet_indexes,
 151                             "Packet with packet_index %d not received" % index)
 152
 153     def test_reassembly(self):
 154         """ basic reassembly """
 155
 156         self.pg_enable_capture()
 157         self.src_if.add_stream(self.fragments_200)
 158         self.pg_start()
 159
 160         packets = self.dst_if.get_capture(len(self.pkt_infos))
 161         self.verify_capture(packets)
 162         self.src_if.assert_nothing_captured()
 163
 164         # run it all again to verify correctness
 165         self.pg_enable_capture()
 166         self.src_if.add_stream(self.fragments_200)
 167         self.pg_start()
 168
 169         packets = self.dst_if.get_capture(len(self.pkt_infos))
 170         self.verify_capture(packets)
 171         self.src_if.assert_nothing_captured()
 172
 173     def test_reversed(self):
 174         """ reverse order reassembly """
 175
 176         fragments = list(self.fragments_200)
 177         fragments.reverse()
 178
 179         self.pg_enable_capture()
 180         self.src_if.add_stream(fragments)
 181         self.pg_start()
 182
 183         packets = self.dst_if.get_capture(len(self.packet_infos))
 184         self.verify_capture(packets)
 185         self.src_if.assert_nothing_captured()
 186
 187         # run it all again to verify correctness
 188         self.pg_enable_capture()
 189         self.src_if.add_stream(fragments)
 190         self.pg_start()
 191
 192         packets = self.dst_if.get_capture(len(self.packet_infos))
 193         self.verify_capture(packets)
 194         self.src_if.assert_nothing_captured()
 195
 196     def test_long_fragment_chain(self):
 197         """ long fragment chain """
 198
 199         error_cnt_str = \
 200             "/err/ip4-full-reassembly-feature/fragment chain too long (drop)"
 201
 202         error_cnt = self.statistics.get_err_counter(error_cnt_str)
 203
 204         self.vapi.ip_reassembly_set(timeout_ms=100, max_reassemblies=1000,
 205                                     max_reassembly_length=3,
 206                                     expire_walk_interval_ms=50)
 207
 208         p1 = (Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac) /
 209               IP(id=1000, src=self.src_if.remote_ip4,
 210                  dst=self.dst_if.remote_ip4) /
 211               UDP(sport=1234, dport=5678) /
 212               Raw(b"X" * 1000))
 213         p2 = (Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac) /
 214               IP(id=1001, src=self.src_if.remote_ip4,
 215                  dst=self.dst_if.remote_ip4) /
 216               UDP(sport=1234, dport=5678) /
 217               Raw(b"X" * 1000))
 218         frags = fragment_rfc791(p1, 200) + fragment_rfc791(p2, 500)
 219
 220         self.pg_enable_capture()
 221         self.src_if.add_stream(frags)
 222         self.pg_start()
 223
 224         self.dst_if.get_capture(1)
 225         self.assert_error_counter_equal(error_cnt_str, error_cnt + 1)
 226
 227     def test_5737(self):
 228         """ fragment length + ip header size > 65535 """
 229         self.vapi.cli("clear errors")
 230         raw = b'''E\x00\x00\x88,\xf8\x1f\xfe@\x01\x98\x00\xc0\xa8\n-\xc0\xa8\n\
 231 \x01\x08\x00\xf0J\xed\xcb\xf1\xf5Test-group: IPv4.IPv4.ipv4-message.\
 232 Ethernet-Payload.IPv4-Packet.IPv4-Header.Fragment-Offset; Test-case: 5737'''
 233         malformed_packet = (Ether(dst=self.src_if.local_mac,
 234                                   src=self.src_if.remote_mac) /
 235                             IP(raw))
 236         p = (Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac) /
 237              IP(id=1000, src=self.src_if.remote_ip4,
 238                 dst=self.dst_if.remote_ip4) /
 239              UDP(sport=1234, dport=5678) /
 240              Raw(b"X" * 1000))
 241         valid_fragments = fragment_rfc791(p, 400)
 242
 243         counter = "/err/ip4-full-reassembly-feature/malformed packets"
 244         error_counter = self.statistics.get_err_counter(counter)
 245         self.pg_enable_capture()
 246         self.src_if.add_stream([malformed_packet] + valid_fragments)
 247         self.pg_start()
 248
 249         self.dst_if.get_capture(1)
 250         self.logger.debug(self.vapi.ppcli("show error"))
 251         self.assertEqual(self.statistics.get_err_counter(counter),
 252                          error_counter + 1)
 253
 254     def test_44924(self):
 255         """ compress tiny fragments """
 256         packets = [(Ether(dst=self.src_if.local_mac,
 257                           src=self.src_if.remote_mac) /
 258                     IP(id=24339, flags="MF", frag=0, ttl=64,
 259                        src=self.src_if.remote_ip4,
 260                        dst=self.dst_if.remote_ip4) /
 261                     ICMP(type="echo-request", code=0, id=0x1fe6, seq=0x2407) /
 262                     Raw(load='Test-group: IPv4')),
 263                    (Ether(dst=self.src_if.local_mac,
 264                           src=self.src_if.remote_mac) /
 265                     IP(id=24339, flags="MF", frag=3, ttl=64,
 266                        src=self.src_if.remote_ip4,
 267                        dst=self.dst_if.remote_ip4) /
 268                     ICMP(type="echo-request", code=0, id=0x1fe6, seq=0x2407) /
 269                     Raw(load='.IPv4.Fragmentation.vali')),
 270                    (Ether(dst=self.src_if.local_mac,
 271                           src=self.src_if.remote_mac) /
 272                     IP(id=24339, frag=6, ttl=64,
 273                        src=self.src_if.remote_ip4,
 274                        dst=self.dst_if.remote_ip4) /
 275                     ICMP(type="echo-request", code=0, id=0x1fe6, seq=0x2407) /
 276                     Raw(load='d; Test-case: 44924'))
 277                    ]
 278
 279         self.pg_enable_capture()
 280         self.src_if.add_stream(packets)
 281         self.pg_start()
 282
 283         self.dst_if.get_capture(1)
 284
 285     def test_frag_1(self):
 286         """ fragment of size 1 """
 287         self.vapi.cli("clear errors")
 288         malformed_packets = [(Ether(dst=self.src_if.local_mac,
 289                                     src=self.src_if.remote_mac) /
 290                               IP(id=7, len=21, flags="MF", frag=0, ttl=64,
 291                                  src=self.src_if.remote_ip4,
 292                                  dst=self.dst_if.remote_ip4) /
 293                               ICMP(type="echo-request")),
 294                              (Ether(dst=self.src_if.local_mac,
 295                                     src=self.src_if.remote_mac) /
 296                               IP(id=7, len=21, frag=1, ttl=64,
 297                                  src=self.src_if.remote_ip4,
 298                                  dst=self.dst_if.remote_ip4) /
 299                               Raw(load=b'\x08')),
 300                              ]
 301
 302         p = (Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac) /
 303              IP(id=1000, src=self.src_if.remote_ip4,
 304                 dst=self.dst_if.remote_ip4) /
 305              UDP(sport=1234, dport=5678) /
 306              Raw(b"X" * 1000))
 307         valid_fragments = fragment_rfc791(p, 400)
 308
 309         self.pg_enable_capture()
 310         self.src_if.add_stream(malformed_packets + valid_fragments)
 311         self.pg_start()
 312
 313         self.dst_if.get_capture(1)
 314
 315         self.assert_packet_counter_equal("ip4-full-reassembly-feature", 1)
 316         # TODO remove above, uncomment below once clearing of counters
 317         # is supported
 318         # self.assert_packet_counter_equal(
 319         #     "/err/ip4-full-reassembly-feature/malformed packets", 1)
 320
 321     def test_random(self):
 322         """ random order reassembly """
 323
 324         fragments = list(self.fragments_200)
 325         shuffle(fragments)
 326
 327         self.pg_enable_capture()
 328         self.src_if.add_stream(fragments)
 329         self.pg_start()
 330
 331         packets = self.dst_if.get_capture(len(self.packet_infos))
 332         self.verify_capture(packets)
 333         self.src_if.assert_nothing_captured()
 334
 335         # run it all again to verify correctness
 336         self.pg_enable_capture()
 337         self.src_if.add_stream(fragments)
 338         self.pg_start()
 339
 340         packets = self.dst_if.get_capture(len(self.packet_infos))
 341         self.verify_capture(packets)
 342         self.src_if.assert_nothing_captured()
 343
 344     def test_duplicates(self):
 345         """ duplicate fragments """
 346
 347         fragments = [
 348             x for (_, frags, _, _) in self.pkt_infos
 349             for x in frags
 350             for _ in range(0, min(2, len(frags)))
 351         ]
 352
 353         self.pg_enable_capture()
 354         self.src_if.add_stream(fragments)
 355         self.pg_start()
 356
 357         packets = self.dst_if.get_capture(len(self.pkt_infos))
 358         self.verify_capture(packets)
 359         self.src_if.assert_nothing_captured()
 360
 361     def test_overlap1(self):
 362         """ overlapping fragments case #1 """
 363
 364         fragments = []
 365         for _, _, frags_300, frags_200 in self.pkt_infos:
 366             if len(frags_300) == 1:
 367                 fragments.extend(frags_300)
 368             else:
 369                 for i, j in zip(frags_200, frags_300):
 370                     fragments.extend(i)
 371                     fragments.extend(j)
 372
 373         self.pg_enable_capture()
 374         self.src_if.add_stream(fragments)
 375         self.pg_start()
 376
 377         packets = self.dst_if.get_capture(len(self.pkt_infos))
 378         self.verify_capture(packets)
 379         self.src_if.assert_nothing_captured()
 380
 381         # run it all to verify correctness
 382         self.pg_enable_capture()
 383         self.src_if.add_stream(fragments)
 384         self.pg_start()
 385
 386         packets = self.dst_if.get_capture(len(self.pkt_infos))
 387         self.verify_capture(packets)
 388         self.src_if.assert_nothing_captured()
 389
 390     def test_overlap2(self):
 391         """ overlapping fragments case #2 """
 392
 393         fragments = []
 394         for _, _, frags_300, frags_200 in self.pkt_infos:
 395             if len(frags_300) == 1:
 396                 fragments.extend(frags_300)
 397             else:
 398                 # care must be taken here so that there are no fragments
 399                 # received by vpp after reassembly is finished, otherwise
 400                 # new reassemblies will be started and packet generator will
 401                 # freak out when it detects unfreed buffers
 402                 zipped = zip(frags_300, frags_200)
 403                 for i, j in zipped:
 404                     fragments.extend(i)
 405                     fragments.extend(j)
 406                 fragments.pop()
 407
 408         self.pg_enable_capture()
 409         self.src_if.add_stream(fragments)
 410         self.pg_start()
 411
 412         packets = self.dst_if.get_capture(len(self.pkt_infos))
 413         self.verify_capture(packets)
 414         self.src_if.assert_nothing_captured()
 415
 416         # run it all to verify correctness
 417         self.pg_enable_capture()
 418         self.src_if.add_stream(fragments)
 419         self.pg_start()
 420
 421         packets = self.dst_if.get_capture(len(self.pkt_infos))
 422         self.verify_capture(packets)
 423         self.src_if.assert_nothing_captured()
 424
 425     def test_timeout_inline(self):
 426         """ timeout (inline) """
 427
 428         dropped_packet_indexes = set(
 429             index for (index, frags, _, _) in self.pkt_infos if len(frags) > 1
 430         )
 431
 432         self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000,
 433                                     max_reassembly_length=3,
 434                                     expire_walk_interval_ms=10000)
 435
 436         self.pg_enable_capture()
 437         self.src_if.add_stream(self.fragments_400)
 438         self.pg_start()
 439
 440         packets = self.dst_if.get_capture(
 441             len(self.pkt_infos) - len(dropped_packet_indexes))
 442         self.verify_capture(packets, dropped_packet_indexes)
 443         self.src_if.assert_nothing_captured()
 444
 445     def test_timeout_cleanup(self):
 446         """ timeout (cleanup) """
 447
 448         # whole packets + fragmented packets sans last fragment
 449         fragments = [
 450             x for (_, frags_400, _, _) in self.pkt_infos
 451             for x in frags_400[:-1 if len(frags_400) > 1 else None]
 452         ]
 453
 454         # last fragments for fragmented packets
 455         fragments2 = [frags_400[-1]
 456                       for (_, frags_400, _, _) in self.pkt_infos
 457                       if len(frags_400) > 1]
 458
 459         dropped_packet_indexes = set(
 460             index for (index, frags_400, _, _) in self.pkt_infos
 461             if len(frags_400) > 1)
 462
 463         self.vapi.ip_reassembly_set(timeout_ms=100, max_reassemblies=1000,
 464                                     max_reassembly_length=1000,
 465                                     expire_walk_interval_ms=50)
 466
 467         self.pg_enable_capture()
 468         self.src_if.add_stream(fragments)
 469         self.pg_start()
 470
 471         self.sleep(.25, "wait before sending rest of fragments")
 472
 473         self.src_if.add_stream(fragments2)
 474         self.pg_start()
 475
 476         packets = self.dst_if.get_capture(
 477             len(self.pkt_infos) - len(dropped_packet_indexes))
 478         self.verify_capture(packets, dropped_packet_indexes)
 479         self.src_if.assert_nothing_captured()
 480
 481     def test_disabled(self):
 482         """ reassembly disabled """
 483
 484         dropped_packet_indexes = set(
 485             index for (index, frags_400, _, _) in self.pkt_infos
 486             if len(frags_400) > 1)
 487
 488         self.vapi.ip_reassembly_set(timeout_ms=1000, max_reassemblies=0,
 489                                     max_reassembly_length=3,
 490                                     expire_walk_interval_ms=10000)
 491
 492         self.pg_enable_capture()
 493         self.src_if.add_stream(self.fragments_400)
 494         self.pg_start()
 495
 496         packets = self.dst_if.get_capture(
 497             len(self.pkt_infos) - len(dropped_packet_indexes))
 498         self.verify_capture(packets, dropped_packet_indexes)
 499         self.src_if.assert_nothing_captured()
 500
 501
 502 class TestIPv4SVReassembly(VppTestCase):
 503     """ IPv4 Shallow Virtual Reassembly """
 504
 505     @classmethod
 506     def setUpClass(cls):
 507         super(TestIPv4SVReassembly, cls).setUpClass()
 508
 509         cls.create_pg_interfaces([0, 1])
 510         cls.src_if = cls.pg0
 511         cls.dst_if = cls.pg1
 512
 513         # setup all interfaces
 514         for i in cls.pg_interfaces:
 515             i.admin_up()
 516             i.config_ip4()
 517             i.resolve_arp()
 518
 519     def setUp(self):
 520         """ Test setup - force timeout on existing reassemblies """
 521         super(TestIPv4SVReassembly, self).setUp()
 522         self.vapi.ip_reassembly_enable_disable(
 523             sw_if_index=self.src_if.sw_if_index, enable_ip4=True,
 524             type=VppEnum.vl_api_ip_reass_type_t.IP_REASS_TYPE_SHALLOW_VIRTUAL)
 525         self.vapi.ip_reassembly_set(
 526             timeout_ms=0, max_reassemblies=1000,
 527             max_reassembly_length=1000,
 528             type=VppEnum.vl_api_ip_reass_type_t.IP_REASS_TYPE_SHALLOW_VIRTUAL,
 529             expire_walk_interval_ms=10)
 530         self.sleep(.25)
 531         self.vapi.ip_reassembly_set(
 532             timeout_ms=1000000, max_reassemblies=1000,
 533             max_reassembly_length=1000,
 534             type=VppEnum.vl_api_ip_reass_type_t.IP_REASS_TYPE_SHALLOW_VIRTUAL,
 535             expire_walk_interval_ms=10000)
 536
 537     def tearDown(self):
 538         super(TestIPv4SVReassembly, self).tearDown()
 539         self.logger.debug(self.vapi.ppcli("show ip4-sv-reassembly details"))
 540         self.logger.debug(self.vapi.ppcli("show buffers"))
 541
 542     def test_basic(self):
 543         """ basic reassembly """
 544         payload_len = 1000
 545         payload = ""
 546         counter = 0
 547         while len(payload) < payload_len:
 548             payload += "%u " % counter
 549             counter += 1
 550
 551         p = (Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac) /
 552              IP(id=1, src=self.src_if.remote_ip4,
 553                 dst=self.dst_if.remote_ip4) /
 554              UDP(sport=1234, dport=5678) /
 555              Raw(payload))
 556         fragments = fragment_rfc791(p, payload_len/4)
 557
 558         # send fragment #2 - should be cached inside reassembly
 559         self.pg_enable_capture()
 560         self.src_if.add_stream(fragments[1])
 561         self.pg_start()
 562         self.logger.debug(self.vapi.ppcli("show ip4-sv-reassembly details"))
 563         self.logger.debug(self.vapi.ppcli("show buffers"))
 564         self.logger.debug(self.vapi.ppcli("show trace"))
 565         self.dst_if.assert_nothing_captured()
 566
 567         # send fragment #1 - reassembly is finished now and both fragments
 568         # forwarded
 569         self.pg_enable_capture()
 570         self.src_if.add_stream(fragments[0])
 571         self.pg_start()
 572         self.logger.debug(self.vapi.ppcli("show ip4-sv-reassembly details"))
 573         self.logger.debug(self.vapi.ppcli("show buffers"))
 574         self.logger.debug(self.vapi.ppcli("show trace"))
 575         c = self.dst_if.get_capture(2)
 576         for sent, recvd in zip([fragments[1], fragments[0]], c):
 577             self.assertEqual(sent[IP].src, recvd[IP].src)
 578             self.assertEqual(sent[IP].dst, recvd[IP].dst)
 579             self.assertEqual(sent[Raw].payload, recvd[Raw].payload)
 580
 581         # send rest of fragments - should be immediately forwarded
 582         self.pg_enable_capture()
 583         self.src_if.add_stream(fragments[2:])
 584         self.pg_start()
 585         c = self.dst_if.get_capture(len(fragments[2:]))
 586         for sent, recvd in zip(fragments[2:], c):
 587             self.assertEqual(sent[IP].src, recvd[IP].src)
 588             self.assertEqual(sent[IP].dst, recvd[IP].dst)
 589             self.assertEqual(sent[Raw].payload, recvd[Raw].payload)
 590
 591     def test_timeout(self):
 592         """ reassembly timeout """
 593         payload_len = 1000
 594         payload = ""
 595         counter = 0
 596         while len(payload) < payload_len:
 597             payload += "%u " % counter
 598             counter += 1
 599
 600         p = (Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac) /
 601              IP(id=1, src=self.src_if.remote_ip4,
 602                 dst=self.dst_if.remote_ip4) /
 603              UDP(sport=1234, dport=5678) /
 604              Raw(payload))
 605         fragments = fragment_rfc791(p, payload_len/4)
 606
 607         self.vapi.ip_reassembly_set(
 608             timeout_ms=100, max_reassemblies=1000,
 609             max_reassembly_length=1000,
 610             expire_walk_interval_ms=50,
 611             type=VppEnum.vl_api_ip_reass_type_t.IP_REASS_TYPE_SHALLOW_VIRTUAL)
 612
 613         # send fragments #2 and #1 - should be forwarded
 614         self.pg_enable_capture()
 615         self.src_if.add_stream(fragments[0:2])
 616         self.pg_start()
 617         self.logger.debug(self.vapi.ppcli("show ip4-sv-reassembly details"))
 618         self.logger.debug(self.vapi.ppcli("show buffers"))
 619         self.logger.debug(self.vapi.ppcli("show trace"))
 620         c = self.dst_if.get_capture(2)
 621         for sent, recvd in zip([fragments[1], fragments[0]], c):
 622             self.assertEqual(sent[IP].src, recvd[IP].src)
 623             self.assertEqual(sent[IP].dst, recvd[IP].dst)
 624             self.assertEqual(sent[Raw].payload, recvd[Raw].payload)
 625
 626         # wait for cleanup
 627         self.sleep(.25, "wait before sending rest of fragments")
 628
 629         # send rest of fragments - shouldn't be forwarded
 630         self.pg_enable_capture()
 631         self.src_if.add_stream(fragments[2:])
 632         self.pg_start()
 633         self.dst_if.assert_nothing_captured()
 634
 635     def test_lru(self):
 636         """ reassembly reuses LRU element """
 637
 638         self.vapi.ip_reassembly_set(
 639             timeout_ms=1000000, max_reassemblies=1,
 640             max_reassembly_length=1000,
 641             type=VppEnum.vl_api_ip_reass_type_t.IP_REASS_TYPE_SHALLOW_VIRTUAL,
 642             expire_walk_interval_ms=10000)
 643
 644         payload_len = 1000
 645         payload = ""
 646         counter = 0
 647         while len(payload) < payload_len:
 648             payload += "%u " % counter
 649             counter += 1
 650
 651         packet_count = 10
 652
 653         fragments = [f
 654                      for i in range(packet_count)
 655                      for p in (Ether(dst=self.src_if.local_mac,
 656                                      src=self.src_if.remote_mac) /
 657                                IP(id=i, src=self.src_if.remote_ip4,
 658                                    dst=self.dst_if.remote_ip4) /
 659                                UDP(sport=1234, dport=5678) /
 660                                Raw(payload))
 661                      for f in fragment_rfc791(p, payload_len/4)]
 662
 663         self.pg_enable_capture()
 664         self.src_if.add_stream(fragments)
 665         self.pg_start()
 666         c = self.dst_if.get_capture(len(fragments))
 667         for sent, recvd in zip(fragments, c):
 668             self.assertEqual(sent[IP].src, recvd[IP].src)
 669             self.assertEqual(sent[IP].dst, recvd[IP].dst)
 670             self.assertEqual(sent[Raw].payload, recvd[Raw].payload)
 671
 672     def send_mixed_and_verify_capture(self, traffic):
 673         stream = []
 674         for t in traffic:
 675             for c in range(t['count']):
 676                 stream.append(
 677                     (Ether(dst=self.src_if.local_mac,
 678                            src=self.src_if.remote_mac) /
 679                      IP(id=self.counter,
 680                         flags=t['flags'],
 681                         src=self.src_if.remote_ip4,
 682                         dst=self.dst_if.remote_ip4) /
 683                      UDP(sport=1234, dport=5678) /
 684                      Raw("abcdef")))
 685                 self.counter = self.counter + 1
 686
 687         self.pg_enable_capture()
 688         self.src_if.add_stream(stream)
 689         self.pg_start()
 690         self.logger.debug(self.vapi.ppcli("show ip4-sv-reassembly details"))
 691         self.logger.debug(self.vapi.ppcli("show buffers"))
 692         self.logger.debug(self.vapi.ppcli("show trace"))
 693         self.dst_if.get_capture(len(stream))
 694
 695     def test_mixed(self):
 696         """ mixed traffic correctly passes through SVR """
 697         self.counter = 1
 698
 699         self.send_mixed_and_verify_capture([{'count': 1, 'flags': ''}])
 700         self.send_mixed_and_verify_capture([{'count': 2, 'flags': ''}])
 701         self.send_mixed_and_verify_capture([{'count': 3, 'flags': ''}])
 702         self.send_mixed_and_verify_capture([{'count': 8, 'flags': ''}])
 703         self.send_mixed_and_verify_capture([{'count': 257, 'flags': ''}])
 704
 705         self.send_mixed_and_verify_capture([{'count': 1, 'flags': 'MF'}])
 706         self.send_mixed_and_verify_capture([{'count': 2, 'flags': 'MF'}])
 707         self.send_mixed_and_verify_capture([{'count': 3, 'flags': 'MF'}])
 708         self.send_mixed_and_verify_capture([{'count': 8, 'flags': 'MF'}])
 709         self.send_mixed_and_verify_capture([{'count': 257, 'flags': 'MF'}])
 710
 711         self.send_mixed_and_verify_capture(
 712             [{'count': 1, 'flags': ''}, {'count': 1, 'flags': 'MF'}])
 713         self.send_mixed_and_verify_capture(
 714             [{'count': 2, 'flags': ''}, {'count': 2, 'flags': 'MF'}])
 715         self.send_mixed_and_verify_capture(
 716             [{'count': 3, 'flags': ''}, {'count': 3, 'flags': 'MF'}])
 717         self.send_mixed_and_verify_capture(
 718             [{'count': 8, 'flags': ''}, {'count': 8, 'flags': 'MF'}])
 719         self.send_mixed_and_verify_capture(
 720             [{'count': 129, 'flags': ''}, {'count': 129, 'flags': 'MF'}])
 721
 722         self.send_mixed_and_verify_capture(
 723             [{'count': 1, 'flags': ''}, {'count': 1, 'flags': 'MF'},
 724              {'count': 1, 'flags': ''}, {'count': 1, 'flags': 'MF'}])
 725         self.send_mixed_and_verify_capture(
 726             [{'count': 2, 'flags': ''}, {'count': 2, 'flags': 'MF'},
 727              {'count': 2, 'flags': ''}, {'count': 2, 'flags': 'MF'}])
 728         self.send_mixed_and_verify_capture(
 729             [{'count': 3, 'flags': ''}, {'count': 3, 'flags': 'MF'},
 730              {'count': 3, 'flags': ''}, {'count': 3, 'flags': 'MF'}])
 731         self.send_mixed_and_verify_capture(
 732             [{'count': 8, 'flags': ''}, {'count': 8, 'flags': 'MF'},
 733              {'count': 8, 'flags': ''}, {'count': 8, 'flags': 'MF'}])
 734         self.send_mixed_and_verify_capture(
 735             [{'count': 65, 'flags': ''}, {'count': 65, 'flags': 'MF'},
 736              {'count': 65, 'flags': ''}, {'count': 65, 'flags': 'MF'}])
 737
 738
 739 class TestIPv4MWReassembly(VppTestCase):
 740     """ IPv4 Reassembly (multiple workers) """
 741     worker_config = "workers %d" % worker_count
 742
 743     @classmethod
 744     def setUpClass(cls):
 745         super(TestIPv4MWReassembly, cls).setUpClass()
 746
 747         cls.create_pg_interfaces(range(worker_count+1))
 748         cls.src_if = cls.pg0
 749         cls.send_ifs = cls.pg_interfaces[:-1]
 750         cls.dst_if = cls.pg_interfaces[-1]
 751
 752         # setup all interfaces
 753         for i in cls.pg_interfaces:
 754             i.admin_up()
 755             i.config_ip4()
 756             i.resolve_arp()
 757
 758         # packets sizes reduced here because we are generating packets without
 759         # Ethernet headers, which are added later (diff fragments go via
 760         # different interfaces)
 761         cls.packet_sizes = [64-len(Ether()), 512-len(Ether()),
 762                             1518-len(Ether()), 9018-len(Ether())]
 763         cls.padding = " abcdefghijklmn"
 764         cls.create_stream(cls.packet_sizes)
 765         cls.create_fragments()
 766
 767     @classmethod
 768     def tearDownClass(cls):
 769         super(TestIPv4MWReassembly, cls).tearDownClass()
 770
 771     def setUp(self):
 772         """ Test setup - force timeout on existing reassemblies """
 773         super(TestIPv4MWReassembly, self).setUp()
 774         for intf in self.send_ifs:
 775             self.vapi.ip_reassembly_enable_disable(
 776                 sw_if_index=intf.sw_if_index, enable_ip4=True)
 777         self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000,
 778                                     max_reassembly_length=1000,
 779                                     expire_walk_interval_ms=10)
 780         self.sleep(.25)
 781         self.vapi.ip_reassembly_set(timeout_ms=1000000, max_reassemblies=1000,
 782                                     max_reassembly_length=1000,
 783                                     expire_walk_interval_ms=10000)
 784
 785     def tearDown(self):
 786         super(TestIPv4MWReassembly, self).tearDown()
 787
 788     def show_commands_at_teardown(self):
 789         self.logger.debug(self.vapi.ppcli("show ip4-full-reassembly details"))
 790         self.logger.debug(self.vapi.ppcli("show buffers"))
 791
 792     @classmethod
 793     def create_stream(cls, packet_sizes, packet_count=test_packet_count):
 794         """Create input packet stream
 795
 796         :param list packet_sizes: Required packet sizes.
 797         """
 798         for i in range(0, packet_count):
 799             info = cls.create_packet_info(cls.src_if, cls.src_if)
 800             payload = cls.info_to_payload(info)
 801             p = (IP(id=info.index, src=cls.src_if.remote_ip4,
 802                     dst=cls.dst_if.remote_ip4) /
 803                  UDP(sport=1234, dport=5678) /
 804                  Raw(payload))
 805             size = packet_sizes[(i // 2) % len(packet_sizes)]
 806             cls.extend_packet(p, size, cls.padding)
 807             info.data = p
 808
 809     @classmethod
 810     def create_fragments(cls):
 811         infos = cls._packet_infos
 812         cls.pkt_infos = []
 813         for index, info in infos.items():
 814             p = info.data
 815             # cls.logger.debug(ppp("Packet:",
 816             #                      p.__class__(scapy.compat.raw(p))))
 817             fragments_400 = fragment_rfc791(p, 400)
 818             cls.pkt_infos.append((index, fragments_400))
 819         cls.fragments_400 = [
 820             x for (_, frags) in cls.pkt_infos for x in frags]
 821         cls.logger.debug("Fragmented %s packets into %s 400-byte fragments, " %
 822                          (len(infos), len(cls.fragments_400)))
 823
 824     def verify_capture(self, capture, dropped_packet_indexes=[]):
 825         """Verify captured packet stream.
 826
 827         :param list capture: Captured packet stream.
 828         """
 829         info = None
 830         seen = set()
 831         for packet in capture:
 832             try:
 833                 self.logger.debug(ppp("Got packet:", packet))
 834                 ip = packet[IP]
 835                 udp = packet[UDP]
 836                 payload_info = self.payload_to_info(packet[Raw])
 837                 packet_index = payload_info.index
 838                 self.assertTrue(
 839                     packet_index not in dropped_packet_indexes,
 840                     ppp("Packet received, but should be dropped:", packet))
 841                 if packet_index in seen:
 842                     raise Exception(ppp("Duplicate packet received", packet))
 843                 seen.add(packet_index)
 844                 self.assertEqual(payload_info.dst, self.src_if.sw_if_index)
 845                 info = self._packet_infos[packet_index]
 846                 self.assertTrue(info is not None)
 847                 self.assertEqual(packet_index, info.index)
 848                 saved_packet = info.data
 849                 self.assertEqual(ip.src, saved_packet[IP].src)
 850                 self.assertEqual(ip.dst, saved_packet[IP].dst)
 851                 self.assertEqual(udp.payload, saved_packet[UDP].payload)
 852             except Exception:
 853                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
 854                 raise
 855         for index in self._packet_infos:
 856             self.assertTrue(index in seen or index in dropped_packet_indexes,
 857                             "Packet with packet_index %d not received" % index)
 858
 859     def send_packets(self, packets):
 860         for counter in range(worker_count):
 861             if 0 == len(packets[counter]):
 862                 continue
 863             send_if = self.send_ifs[counter]
 864             send_if.add_stream(
 865                 (Ether(dst=send_if.local_mac, src=send_if.remote_mac) / x
 866                  for x in packets[counter]),
 867                 worker=counter)
 868         self.pg_start()
 869
 870     def test_worker_conflict(self):
 871         """ 1st and FO=0 fragments on different workers """
 872
 873         # in first wave we send fragments which don't start at offset 0
 874         # then we send fragments with offset 0 on a different thread
 875         # then the rest of packets on a random thread
 876         first_packets = [[] for n in range(worker_count)]
 877         second_packets = [[] for n in range(worker_count)]
 878         rest_of_packets = [[] for n in range(worker_count)]
 879         for (_, p) in self.pkt_infos:
 880             wi = randrange(worker_count)
 881             second_packets[wi].append(p[0])
 882             if len(p) <= 1:
 883                 continue
 884             wi2 = wi
 885             while wi2 == wi:
 886                 wi2 = randrange(worker_count)
 887             first_packets[wi2].append(p[1])
 888             wi3 = randrange(worker_count)
 889             rest_of_packets[wi3].extend(p[2:])
 890
 891         self.pg_enable_capture()
 892         self.send_packets(first_packets)
 893         self.send_packets(second_packets)
 894         self.send_packets(rest_of_packets)
 895
 896         packets = self.dst_if.get_capture(len(self.pkt_infos))
 897         self.verify_capture(packets)
 898         for send_if in self.send_ifs:
 899             send_if.assert_nothing_captured()
 900
 901         self.logger.debug(self.vapi.ppcli("show trace"))
 902         self.logger.debug(self.vapi.ppcli("show ip4-full-reassembly details"))
 903         self.logger.debug(self.vapi.ppcli("show buffers"))
 904         self.vapi.cli("clear trace")
 905
 906         self.pg_enable_capture()
 907         self.send_packets(first_packets)
 908         self.send_packets(second_packets)
 909         self.send_packets(rest_of_packets)
 910
 911         packets = self.dst_if.get_capture(len(self.pkt_infos))
 912         self.verify_capture(packets)
 913         for send_if in self.send_ifs:
 914             send_if.assert_nothing_captured()
 915
 916
 917 class TestIPv6Reassembly(VppTestCase):
 918     """ IPv6 Reassembly """
 919
 920     @classmethod
 921     def setUpClass(cls):
 922         super(TestIPv6Reassembly, cls).setUpClass()
 923
 924         cls.create_pg_interfaces([0, 1])
 925         cls.src_if = cls.pg0
 926         cls.dst_if = cls.pg1
 927
 928         # setup all interfaces
 929         for i in cls.pg_interfaces:
 930             i.admin_up()
 931             i.config_ip6()
 932             i.resolve_ndp()
 933
 934         # packet sizes
 935         cls.packet_sizes = [64, 512, 1518, 9018]
 936         cls.padding = " abcdefghijklmn"
 937         cls.create_stream(cls.packet_sizes)
 938         cls.create_fragments()
 939
 940     @classmethod
 941     def tearDownClass(cls):
 942         super(TestIPv6Reassembly, cls).tearDownClass()
 943
 944     def setUp(self):
 945         """ Test setup - force timeout on existing reassemblies """
 946         super(TestIPv6Reassembly, self).setUp()
 947         self.vapi.ip_reassembly_enable_disable(
 948             sw_if_index=self.src_if.sw_if_index, enable_ip6=True)
 949         self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000,
 950                                     max_reassembly_length=1000,
 951                                     expire_walk_interval_ms=10, is_ip6=1)
 952         self.sleep(.25)
 953         self.vapi.ip_reassembly_set(timeout_ms=1000000, max_reassemblies=1000,
 954                                     max_reassembly_length=1000,
 955                                     expire_walk_interval_ms=10000, is_ip6=1)
 956         self.logger.debug(self.vapi.ppcli("show ip6-full-reassembly details"))
 957         self.logger.debug(self.vapi.ppcli("show buffers"))
 958
 959     def tearDown(self):
 960         super(TestIPv6Reassembly, self).tearDown()
 961
 962     def show_commands_at_teardown(self):
 963         self.logger.debug(self.vapi.ppcli("show ip6-full-reassembly details"))
 964         self.logger.debug(self.vapi.ppcli("show buffers"))
 965
 966     @classmethod
 967     def create_stream(cls, packet_sizes, packet_count=test_packet_count):
 968         """Create input packet stream for defined interface.
 969
 970         :param list packet_sizes: Required packet sizes.
 971         """
 972         for i in range(0, packet_count):
 973             info = cls.create_packet_info(cls.src_if, cls.src_if)
 974             payload = cls.info_to_payload(info)
 975             p = (Ether(dst=cls.src_if.local_mac, src=cls.src_if.remote_mac) /
 976                  IPv6(src=cls.src_if.remote_ip6,
 977                       dst=cls.dst_if.remote_ip6) /
 978                  UDP(sport=1234, dport=5678) /
 979                  Raw(payload))
 980             size = packet_sizes[(i // 2) % len(packet_sizes)]
 981             cls.extend_packet(p, size, cls.padding)
 982             info.data = p
 983
 984     @classmethod
 985     def create_fragments(cls):
 986         infos = cls._packet_infos
 987         cls.pkt_infos = []
 988         for index, info in infos.items():
 989             p = info.data
 990             # cls.logger.debug(ppp("Packet:",
 991             #                      p.__class__(scapy.compat.raw(p))))
 992             fragments_400 = fragment_rfc8200(p, info.index, 400)
 993             fragments_300 = fragment_rfc8200(p, info.index, 300)
 994             cls.pkt_infos.append((index, fragments_400, fragments_300))
 995         cls.fragments_400 = [
 996             x for _, frags, _ in cls.pkt_infos for x in frags]
 997         cls.fragments_300 = [
 998             x for _, _, frags in cls.pkt_infos for x in frags]
 999         cls.logger.debug("Fragmented %s packets into %s 400-byte fragments, "
1000                          "and %s 300-byte fragments" %
1001                          (len(infos), len(cls.fragments_400),
1002                              len(cls.fragments_300)))
1003
1004     def verify_capture(self, capture, dropped_packet_indexes=[]):
1005         """Verify captured packet strea .
1006
1007         :param list capture: Captured packet stream.
1008         """
1009         info = None
1010         seen = set()
1011         for packet in capture:
1012             try:
1013                 self.logger.debug(ppp("Got packet:", packet))
1014                 ip = packet[IPv6]
1015                 udp = packet[UDP]
1016                 payload_info = self.payload_to_info(packet[Raw])
1017                 packet_index = payload_info.index
1018                 self.assertTrue(
1019                     packet_index not in dropped_packet_indexes,
1020                     ppp("Packet received, but should be dropped:", packet))
1021                 if packet_index in seen:
1022                     raise Exception(ppp("Duplicate packet received", packet))
1023                 seen.add(packet_index)
1024                 self.assertEqual(payload_info.dst, self.src_if.sw_if_index)
1025                 info = self._packet_infos[packet_index]
1026                 self.assertTrue(info is not None)
1027                 self.assertEqual(packet_index, info.index)
1028                 saved_packet = info.data
1029                 self.assertEqual(ip.src, saved_packet[IPv6].src)
1030                 self.assertEqual(ip.dst, saved_packet[IPv6].dst)
1031                 self.assertEqual(udp.payload, saved_packet[UDP].payload)
1032             except Exception:
1033                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
1034                 raise
1035         for index in self._packet_infos:
1036             self.assertTrue(index in seen or index in dropped_packet_indexes,
1037                             "Packet with packet_index %d not received" % index)
1038
1039     def test_reassembly(self):
1040         """ basic reassembly """
1041
1042         self.pg_enable_capture()
1043         self.src_if.add_stream(self.fragments_400)
1044         self.pg_start()
1045
1046         packets = self.dst_if.get_capture(len(self.pkt_infos))
1047         self.verify_capture(packets)
1048         self.src_if.assert_nothing_captured()
1049
1050         # run it all again to verify correctness
1051         self.pg_enable_capture()
1052         self.src_if.add_stream(self.fragments_400)
1053         self.pg_start()
1054
1055         packets = self.dst_if.get_capture(len(self.pkt_infos))
1056         self.verify_capture(packets)
1057         self.src_if.assert_nothing_captured()
1058
1059     def test_buffer_boundary(self):
1060         """ fragment header crossing buffer boundary """
1061
1062         p = (Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac) /
1063              IPv6(src=self.src_if.remote_ip6,
1064                   dst=self.src_if.local_ip6) /
1065              IPv6ExtHdrHopByHop(
1066                  options=[HBHOptUnknown(otype=0xff, optlen=0)] * 1000) /
1067              IPv6ExtHdrFragment(m=1) /
1068              UDP(sport=1234, dport=5678) /
1069              Raw())
1070         self.pg_enable_capture()
1071         self.src_if.add_stream([p])
1072         self.pg_start()
1073         self.src_if.assert_nothing_captured()
1074         self.dst_if.assert_nothing_captured()
1075
1076     def test_reversed(self):
1077         """ reverse order reassembly """
1078
1079         fragments = list(self.fragments_400)
1080         fragments.reverse()
1081
1082         self.pg_enable_capture()
1083         self.src_if.add_stream(fragments)
1084         self.pg_start()
1085
1086         packets = self.dst_if.get_capture(len(self.pkt_infos))
1087         self.verify_capture(packets)
1088         self.src_if.assert_nothing_captured()
1089
1090         # run it all again to verify correctness
1091         self.pg_enable_capture()
1092         self.src_if.add_stream(fragments)
1093         self.pg_start()
1094
1095         packets = self.dst_if.get_capture(len(self.pkt_infos))
1096         self.verify_capture(packets)
1097         self.src_if.assert_nothing_captured()
1098
1099     def test_random(self):
1100         """ random order reassembly """
1101
1102         fragments = list(self.fragments_400)
1103         shuffle(fragments)
1104
1105         self.pg_enable_capture()
1106         self.src_if.add_stream(fragments)
1107         self.pg_start()
1108
1109         packets = self.dst_if.get_capture(len(self.pkt_infos))
1110         self.verify_capture(packets)
1111         self.src_if.assert_nothing_captured()
1112
1113         # run it all again to verify correctness
1114         self.pg_enable_capture()
1115         self.src_if.add_stream(fragments)
1116         self.pg_start()
1117
1118         packets = self.dst_if.get_capture(len(self.pkt_infos))
1119         self.verify_capture(packets)
1120         self.src_if.assert_nothing_captured()
1121
1122     def test_duplicates(self):
1123         """ duplicate fragments """
1124
1125         fragments = [
1126             x for (_, frags, _) in self.pkt_infos
1127             for x in frags
1128             for _ in range(0, min(2, len(frags)))
1129         ]
1130
1131         self.pg_enable_capture()
1132         self.src_if.add_stream(fragments)
1133         self.pg_start()
1134
1135         packets = self.dst_if.get_capture(len(self.pkt_infos))
1136         self.verify_capture(packets)
1137         self.src_if.assert_nothing_captured()
1138
1139     def test_long_fragment_chain(self):
1140         """ long fragment chain """
1141
1142         error_cnt_str = \
1143             "/err/ip6-full-reassembly-feature/fragment chain too long (drop)"
1144
1145         error_cnt = self.statistics.get_err_counter(error_cnt_str)
1146
1147         self.vapi.ip_reassembly_set(timeout_ms=100, max_reassemblies=1000,
1148                                     max_reassembly_length=3,
1149                                     expire_walk_interval_ms=50, is_ip6=1)
1150
1151         p = (Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac) /
1152              IPv6(src=self.src_if.remote_ip6,
1153                   dst=self.dst_if.remote_ip6) /
1154              UDP(sport=1234, dport=5678) /
1155              Raw(b"X" * 1000))
1156         frags = fragment_rfc8200(p, 1, 300) + fragment_rfc8200(p, 2, 500)
1157
1158         self.pg_enable_capture()
1159         self.src_if.add_stream(frags)
1160         self.pg_start()
1161
1162         self.dst_if.get_capture(1)
1163         self.assert_error_counter_equal(error_cnt_str, error_cnt + 1)
1164
1165     def test_overlap1(self):
1166         """ overlapping fragments case #1 """
1167
1168         fragments = []
1169         for _, frags_400, frags_300 in self.pkt_infos:
1170             if len(frags_300) == 1:
1171                 fragments.extend(frags_400)
1172             else:
1173                 for i, j in zip(frags_300, frags_400):
1174                     fragments.extend(i)
1175                     fragments.extend(j)
1176
1177         dropped_packet_indexes = set(
1178             index for (index, _, frags) in self.pkt_infos if len(frags) > 1
1179         )
1180
1181         self.pg_enable_capture()
1182         self.src_if.add_stream(fragments)
1183         self.pg_start()
1184
1185         packets = self.dst_if.get_capture(
1186             len(self.pkt_infos) - len(dropped_packet_indexes))
1187         self.verify_capture(packets, dropped_packet_indexes)
1188         self.src_if.assert_nothing_captured()
1189
1190     def test_overlap2(self):
1191         """ overlapping fragments case #2 """
1192
1193         fragments = []
1194         for _, frags_400, frags_300 in self.pkt_infos:
1195             if len(frags_400) == 1:
1196                 fragments.extend(frags_400)
1197             else:
1198                 # care must be taken here so that there are no fragments
1199                 # received by vpp after reassembly is finished, otherwise
1200                 # new reassemblies will be started and packet generator will
1201                 # freak out when it detects unfreed buffers
1202                 zipped = zip(frags_400, frags_300)
1203                 for i, j in zipped:
1204                     fragments.extend(i)
1205                     fragments.extend(j)
1206                 fragments.pop()
1207
1208         dropped_packet_indexes = set(
1209             index for (index, _, frags) in self.pkt_infos if len(frags) > 1
1210         )
1211
1212         self.pg_enable_capture()
1213         self.src_if.add_stream(fragments)
1214         self.pg_start()
1215
1216         packets = self.dst_if.get_capture(
1217             len(self.pkt_infos) - len(dropped_packet_indexes))
1218         self.verify_capture(packets, dropped_packet_indexes)
1219         self.src_if.assert_nothing_captured()
1220
1221     def test_timeout_inline(self):
1222         """ timeout (inline) """
1223
1224         dropped_packet_indexes = set(
1225             index for (index, frags, _) in self.pkt_infos if len(frags) > 1
1226         )
1227
1228         self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000,
1229                                     max_reassembly_length=3,
1230                                     expire_walk_interval_ms=10000, is_ip6=1)
1231
1232         self.pg_enable_capture()
1233         self.src_if.add_stream(self.fragments_400)
1234         self.pg_start()
1235
1236         packets = self.dst_if.get_capture(
1237             len(self.pkt_infos) - len(dropped_packet_indexes))
1238         self.verify_capture(packets, dropped_packet_indexes)
1239         pkts = self.src_if.get_capture(
1240             expected_count=len(dropped_packet_indexes))
1241         for icmp in pkts:
1242             self.assertIn(ICMPv6TimeExceeded, icmp)
1243             self.assertIn(IPv6ExtHdrFragment, icmp)
1244             self.assertIn(icmp[IPv6ExtHdrFragment].id, dropped_packet_indexes)
1245             dropped_packet_indexes.remove(icmp[IPv6ExtHdrFragment].id)
1246
1247     def test_timeout_cleanup(self):
1248         """ timeout (cleanup) """
1249
1250         # whole packets + fragmented packets sans last fragment
1251         fragments = [
1252             x for (_, frags_400, _) in self.pkt_infos
1253             for x in frags_400[:-1 if len(frags_400) > 1 else None]
1254         ]
1255
1256         # last fragments for fragmented packets
1257         fragments2 = [frags_400[-1]
1258                       for (_, frags_400, _) in self.pkt_infos
1259                       if len(frags_400) > 1]
1260
1261         dropped_packet_indexes = set(
1262             index for (index, frags_400, _) in self.pkt_infos
1263             if len(frags_400) > 1)
1264
1265         self.vapi.ip_reassembly_set(timeout_ms=100, max_reassemblies=1000,
1266                                     max_reassembly_length=1000,
1267                                     expire_walk_interval_ms=50)
1268
1269         self.vapi.ip_reassembly_set(timeout_ms=100, max_reassemblies=1000,
1270                                     max_reassembly_length=1000,
1271                                     expire_walk_interval_ms=50, is_ip6=1)
1272
1273         self.pg_enable_capture()
1274         self.src_if.add_stream(fragments)
1275         self.pg_start()
1276
1277         self.sleep(.25, "wait before sending rest of fragments")
1278
1279         self.src_if.add_stream(fragments2)
1280         self.pg_start()
1281
1282         packets = self.dst_if.get_capture(
1283             len(self.pkt_infos) - len(dropped_packet_indexes))
1284         self.verify_capture(packets, dropped_packet_indexes)
1285         pkts = self.src_if.get_capture(
1286             expected_count=len(dropped_packet_indexes))
1287         for icmp in pkts:
1288             self.assertIn(ICMPv6TimeExceeded, icmp)
1289             self.assertIn(IPv6ExtHdrFragment, icmp)
1290             self.assertIn(icmp[IPv6ExtHdrFragment].id, dropped_packet_indexes)
1291             dropped_packet_indexes.remove(icmp[IPv6ExtHdrFragment].id)
1292
1293     def test_disabled(self):
1294         """ reassembly disabled """
1295
1296         dropped_packet_indexes = set(
1297             index for (index, frags_400, _) in self.pkt_infos
1298             if len(frags_400) > 1)
1299
1300         self.vapi.ip_reassembly_set(timeout_ms=1000, max_reassemblies=0,
1301                                     max_reassembly_length=3,
1302                                     expire_walk_interval_ms=10000, is_ip6=1)
1303
1304         self.pg_enable_capture()
1305         self.src_if.add_stream(self.fragments_400)
1306         self.pg_start()
1307
1308         packets = self.dst_if.get_capture(
1309             len(self.pkt_infos) - len(dropped_packet_indexes))
1310         self.verify_capture(packets, dropped_packet_indexes)
1311         self.src_if.assert_nothing_captured()
1312
1313     def test_missing_upper(self):
1314         """ missing upper layer """
1315         p = (Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac) /
1316              IPv6(src=self.src_if.remote_ip6,
1317                   dst=self.src_if.local_ip6) /
1318              UDP(sport=1234, dport=5678) /
1319              Raw())
1320         self.extend_packet(p, 1000, self.padding)
1321         fragments = fragment_rfc8200(p, 1, 500)
1322         bad_fragment = p.__class__(scapy.compat.raw(fragments[1]))
1323         bad_fragment[IPv6ExtHdrFragment].nh = 59
1324         bad_fragment[IPv6ExtHdrFragment].offset = 0
1325         self.pg_enable_capture()
1326         self.src_if.add_stream([bad_fragment])
1327         self.pg_start()
1328         pkts = self.src_if.get_capture(expected_count=1)
1329         icmp = pkts[0]
1330         self.assertIn(ICMPv6ParamProblem, icmp)
1331         self.assert_equal(icmp[ICMPv6ParamProblem].code, 3, "ICMP code")
1332
1333     def test_invalid_frag_size(self):
1334         """ fragment size not a multiple of 8 """
1335         p = (Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac) /
1336              IPv6(src=self.src_if.remote_ip6,
1337                   dst=self.src_if.local_ip6) /
1338              UDP(sport=1234, dport=5678) /
1339              Raw())
1340         self.extend_packet(p, 1000, self.padding)
1341         fragments = fragment_rfc8200(p, 1, 500)
1342         bad_fragment = fragments[0]
1343         self.extend_packet(bad_fragment, len(bad_fragment) + 5)
1344         self.pg_enable_capture()
1345         self.src_if.add_stream([bad_fragment])
1346         self.pg_start()
1347         pkts = self.src_if.get_capture(expected_count=1)
1348         icmp = pkts[0]
1349         self.assertIn(ICMPv6ParamProblem, icmp)
1350         self.assert_equal(icmp[ICMPv6ParamProblem].code, 0, "ICMP code")
1351
1352     def test_invalid_packet_size(self):
1353         """ total packet size > 65535 """
1354         p = (Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac) /
1355              IPv6(src=self.src_if.remote_ip6,
1356                   dst=self.src_if.local_ip6) /
1357              UDP(sport=1234, dport=5678) /
1358              Raw())
1359         self.extend_packet(p, 1000, self.padding)
1360         fragments = fragment_rfc8200(p, 1, 500)
1361         bad_fragment = fragments[1]
1362         bad_fragment[IPv6ExtHdrFragment].offset = 65500
1363         self.pg_enable_capture()
1364         self.src_if.add_stream([bad_fragment])
1365         self.pg_start()
1366         pkts = self.src_if.get_capture(expected_count=1)
1367         icmp = pkts[0]
1368         self.assertIn(ICMPv6ParamProblem, icmp)
1369         self.assert_equal(icmp[ICMPv6ParamProblem].code, 0, "ICMP code")
1370
1371
1372 class TestIPv6MWReassembly(VppTestCase):
1373     """ IPv6 Reassembly (multiple workers) """
1374     worker_config = "workers %d" % worker_count
1375
1376     @classmethod
1377     def setUpClass(cls):
1378         super(TestIPv6MWReassembly, cls).setUpClass()
1379
1380         cls.create_pg_interfaces(range(worker_count+1))
1381         cls.src_if = cls.pg0
1382         cls.send_ifs = cls.pg_interfaces[:-1]
1383         cls.dst_if = cls.pg_interfaces[-1]
1384
1385         # setup all interfaces
1386         for i in cls.pg_interfaces:
1387             i.admin_up()
1388             i.config_ip6()
1389             i.resolve_ndp()
1390
1391         # packets sizes reduced here because we are generating packets without
1392         # Ethernet headers, which are added later (diff fragments go via
1393         # different interfaces)
1394         cls.packet_sizes = [64-len(Ether()), 512-len(Ether()),
1395                             1518-len(Ether()), 9018-len(Ether())]
1396         cls.padding = " abcdefghijklmn"
1397         cls.create_stream(cls.packet_sizes)
1398         cls.create_fragments()
1399
1400     @classmethod
1401     def tearDownClass(cls):
1402         super(TestIPv6MWReassembly, cls).tearDownClass()
1403
1404     def setUp(self):
1405         """ Test setup - force timeout on existing reassemblies """
1406         super(TestIPv6MWReassembly, self).setUp()
1407         for intf in self.send_ifs:
1408             self.vapi.ip_reassembly_enable_disable(
1409                 sw_if_index=intf.sw_if_index, enable_ip6=True)
1410         self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000,
1411                                     max_reassembly_length=1000,
1412                                     expire_walk_interval_ms=10, is_ip6=1)
1413         self.sleep(.25)
1414         self.vapi.ip_reassembly_set(timeout_ms=1000000, max_reassemblies=1000,
1415                                     max_reassembly_length=1000,
1416                                     expire_walk_interval_ms=1000, is_ip6=1)
1417
1418     def tearDown(self):
1419         super(TestIPv6MWReassembly, self).tearDown()
1420
1421     def show_commands_at_teardown(self):
1422         self.logger.debug(self.vapi.ppcli("show ip6-full-reassembly details"))
1423         self.logger.debug(self.vapi.ppcli("show buffers"))
1424
1425     @classmethod
1426     def create_stream(cls, packet_sizes, packet_count=test_packet_count):
1427         """Create input packet stream
1428
1429         :param list packet_sizes: Required packet sizes.
1430         """
1431         for i in range(0, packet_count):
1432             info = cls.create_packet_info(cls.src_if, cls.src_if)
1433             payload = cls.info_to_payload(info)
1434             p = (IPv6(src=cls.src_if.remote_ip6,
1435                       dst=cls.dst_if.remote_ip6) /
1436                  UDP(sport=1234, dport=5678) /
1437                  Raw(payload))
1438             size = packet_sizes[(i // 2) % len(packet_sizes)]
1439             cls.extend_packet(p, size, cls.padding)
1440             info.data = p
1441
1442     @classmethod
1443     def create_fragments(cls):
1444         infos = cls._packet_infos
1445         cls.pkt_infos = []
1446         for index, info in infos.items():
1447             p = info.data
1448             # cls.logger.debug(ppp("Packet:",
1449             #                      p.__class__(scapy.compat.raw(p))))
1450             fragments_400 = fragment_rfc8200(p, index, 400)
1451             cls.pkt_infos.append((index, fragments_400))
1452         cls.fragments_400 = [
1453             x for (_, frags) in cls.pkt_infos for x in frags]
1454         cls.logger.debug("Fragmented %s packets into %s 400-byte fragments, " %
1455                          (len(infos), len(cls.fragments_400)))
1456
1457     def verify_capture(self, capture, dropped_packet_indexes=[]):
1458         """Verify captured packet strea .
1459
1460         :param list capture: Captured packet stream.
1461         """
1462         info = None
1463         seen = set()
1464         for packet in capture:
1465             try:
1466                 self.logger.debug(ppp("Got packet:", packet))
1467                 ip = packet[IPv6]
1468                 udp = packet[UDP]
1469                 payload_info = self.payload_to_info(packet[Raw])
1470                 packet_index = payload_info.index
1471                 self.assertTrue(
1472                     packet_index not in dropped_packet_indexes,
1473                     ppp("Packet received, but should be dropped:", packet))
1474                 if packet_index in seen:
1475                     raise Exception(ppp("Duplicate packet received", packet))
1476                 seen.add(packet_index)
1477                 self.assertEqual(payload_info.dst, self.src_if.sw_if_index)
1478                 info = self._packet_infos[packet_index]
1479                 self.assertTrue(info is not None)
1480                 self.assertEqual(packet_index, info.index)
1481                 saved_packet = info.data
1482                 self.assertEqual(ip.src, saved_packet[IPv6].src)
1483                 self.assertEqual(ip.dst, saved_packet[IPv6].dst)
1484                 self.assertEqual(udp.payload, saved_packet[UDP].payload)
1485             except Exception:
1486                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
1487                 raise
1488         for index in self._packet_infos:
1489             self.assertTrue(index in seen or index in dropped_packet_indexes,
1490                             "Packet with packet_index %d not received" % index)
1491
1492     def send_packets(self, packets):
1493         for counter in range(worker_count):
1494             if 0 == len(packets[counter]):
1495                 continue
1496             send_if = self.send_ifs[counter]
1497             send_if.add_stream(
1498                 (Ether(dst=send_if.local_mac, src=send_if.remote_mac) / x
1499                  for x in packets[counter]),
1500                 worker=counter)
1501         self.pg_start()
1502
1503     def test_worker_conflict(self):
1504         """ 1st and FO=0 fragments on different workers """
1505
1506         # in first wave we send fragments which don't start at offset 0
1507         # then we send fragments with offset 0 on a different thread
1508         # then the rest of packets on a random thread
1509         first_packets = [[] for n in range(worker_count)]
1510         second_packets = [[] for n in range(worker_count)]
1511         rest_of_packets = [[] for n in range(worker_count)]
1512         for (_, p) in self.pkt_infos:
1513             wi = randrange(worker_count)
1514             second_packets[wi].append(p[0])
1515             if len(p) <= 1:
1516                 continue
1517             wi2 = wi
1518             while wi2 == wi:
1519                 wi2 = randrange(worker_count)
1520             first_packets[wi2].append(p[1])
1521             wi3 = randrange(worker_count)
1522             rest_of_packets[wi3].extend(p[2:])
1523
1524         self.pg_enable_capture()
1525         self.send_packets(first_packets)
1526         self.send_packets(second_packets)
1527         self.send_packets(rest_of_packets)
1528
1529         packets = self.dst_if.get_capture(len(self.pkt_infos))
1530         self.verify_capture(packets)
1531         for send_if in self.send_ifs:
1532             send_if.assert_nothing_captured()
1533
1534         self.logger.debug(self.vapi.ppcli("show trace"))
1535         self.logger.debug(self.vapi.ppcli("show ip6-full-reassembly details"))
1536         self.logger.debug(self.vapi.ppcli("show buffers"))
1537         self.vapi.cli("clear trace")
1538
1539         self.pg_enable_capture()
1540         self.send_packets(first_packets)
1541         self.send_packets(second_packets)
1542         self.send_packets(rest_of_packets)
1543
1544         packets = self.dst_if.get_capture(len(self.pkt_infos))
1545         self.verify_capture(packets)
1546         for send_if in self.send_ifs:
1547             send_if.assert_nothing_captured()
1548
1549
1550 class TestIPv6SVReassembly(VppTestCase):
1551     """ IPv6 Shallow Virtual Reassembly """
1552
1553     @classmethod
1554     def setUpClass(cls):
1555         super(TestIPv6SVReassembly, cls).setUpClass()
1556
1557         cls.create_pg_interfaces([0, 1])
1558         cls.src_if = cls.pg0
1559         cls.dst_if = cls.pg1
1560
1561         # setup all interfaces
1562         for i in cls.pg_interfaces:
1563             i.admin_up()
1564             i.config_ip6()
1565             i.resolve_ndp()
1566
1567     def setUp(self):
1568         """ Test setup - force timeout on existing reassemblies """
1569         super(TestIPv6SVReassembly, self).setUp()
1570         self.vapi.ip_reassembly_enable_disable(
1571             sw_if_index=self.src_if.sw_if_index, enable_ip6=True,
1572             type=VppEnum.vl_api_ip_reass_type_t.IP_REASS_TYPE_SHALLOW_VIRTUAL)
1573         self.vapi.ip_reassembly_set(
1574             timeout_ms=0, max_reassemblies=1000,
1575             max_reassembly_length=1000,
1576             type=VppEnum.vl_api_ip_reass_type_t.IP_REASS_TYPE_SHALLOW_VIRTUAL,
1577             expire_walk_interval_ms=10, is_ip6=1)
1578         self.sleep(.25)
1579         self.vapi.ip_reassembly_set(
1580             timeout_ms=1000000, max_reassemblies=1000,
1581             max_reassembly_length=1000,
1582             type=VppEnum.vl_api_ip_reass_type_t.IP_REASS_TYPE_SHALLOW_VIRTUAL,
1583             expire_walk_interval_ms=10000, is_ip6=1)
1584
1585     def tearDown(self):
1586         super(TestIPv6SVReassembly, self).tearDown()
1587         self.logger.debug(self.vapi.ppcli("show ip6-sv-reassembly details"))
1588         self.logger.debug(self.vapi.ppcli("show buffers"))
1589
1590     def test_basic(self):
1591         """ basic reassembly """
1592         payload_len = 1000
1593         payload = ""
1594         counter = 0
1595         while len(payload) < payload_len:
1596             payload += "%u " % counter
1597             counter += 1
1598
1599         p = (Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac) /
1600              IPv6(src=self.src_if.remote_ip6, dst=self.dst_if.remote_ip6) /
1601              UDP(sport=1234, dport=5678) /
1602              Raw(payload))
1603         fragments = fragment_rfc8200(p, 1, payload_len/4)
1604
1605         # send fragment #2 - should be cached inside reassembly
1606         self.pg_enable_capture()
1607         self.src_if.add_stream(fragments[1])
1608         self.pg_start()
1609         self.logger.debug(self.vapi.ppcli("show ip6-sv-reassembly details"))
1610         self.logger.debug(self.vapi.ppcli("show buffers"))
1611         self.logger.debug(self.vapi.ppcli("show trace"))
1612         self.dst_if.assert_nothing_captured()
1613
1614         # send fragment #1 - reassembly is finished now and both fragments
1615         # forwarded
1616         self.pg_enable_capture()
1617         self.src_if.add_stream(fragments[0])
1618         self.pg_start()
1619         self.logger.debug(self.vapi.ppcli("show ip6-sv-reassembly details"))
1620         self.logger.debug(self.vapi.ppcli("show buffers"))
1621         self.logger.debug(self.vapi.ppcli("show trace"))
1622         c = self.dst_if.get_capture(2)
1623         for sent, recvd in zip([fragments[1], fragments[0]], c):
1624             self.assertEqual(sent[IPv6].src, recvd[IPv6].src)
1625             self.assertEqual(sent[IPv6].dst, recvd[IPv6].dst)
1626             self.assertEqual(sent[Raw].payload, recvd[Raw].payload)
1627
1628         # send rest of fragments - should be immediately forwarded
1629         self.pg_enable_capture()
1630         self.src_if.add_stream(fragments[2:])
1631         self.pg_start()
1632         c = self.dst_if.get_capture(len(fragments[2:]))
1633         for sent, recvd in zip(fragments[2:], c):
1634             self.assertEqual(sent[IPv6].src, recvd[IPv6].src)
1635             self.assertEqual(sent[IPv6].dst, recvd[IPv6].dst)
1636             self.assertEqual(sent[Raw].payload, recvd[Raw].payload)
1637
1638     def test_timeout(self):
1639         """ reassembly timeout """
1640         payload_len = 1000
1641         payload = ""
1642         counter = 0
1643         while len(payload) < payload_len:
1644             payload += "%u " % counter
1645             counter += 1
1646
1647         p = (Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac) /
1648              IPv6(src=self.src_if.remote_ip6, dst=self.dst_if.remote_ip6) /
1649              UDP(sport=1234, dport=5678) /
1650              Raw(payload))
1651         fragments = fragment_rfc8200(p, 1, payload_len/4)
1652
1653         self.vapi.ip_reassembly_set(
1654             timeout_ms=100, max_reassemblies=1000,
1655             max_reassembly_length=1000,
1656             expire_walk_interval_ms=50,
1657             is_ip6=1,
1658             type=VppEnum.vl_api_ip_reass_type_t.IP_REASS_TYPE_SHALLOW_VIRTUAL)
1659
1660         # send fragments #2 and #1 - should be forwarded
1661         self.pg_enable_capture()
1662         self.src_if.add_stream(fragments[0:2])
1663         self.pg_start()
1664         self.logger.debug(self.vapi.ppcli("show ip4-sv-reassembly details"))
1665         self.logger.debug(self.vapi.ppcli("show buffers"))
1666         self.logger.debug(self.vapi.ppcli("show trace"))
1667         c = self.dst_if.get_capture(2)
1668         for sent, recvd in zip([fragments[1], fragments[0]], c):
1669             self.assertEqual(sent[IPv6].src, recvd[IPv6].src)
1670             self.assertEqual(sent[IPv6].dst, recvd[IPv6].dst)
1671             self.assertEqual(sent[Raw].payload, recvd[Raw].payload)
1672
1673         # wait for cleanup
1674         self.sleep(.25, "wait before sending rest of fragments")
1675
1676         # send rest of fragments - shouldn't be forwarded
1677         self.pg_enable_capture()
1678         self.src_if.add_stream(fragments[2:])
1679         self.pg_start()
1680         self.dst_if.assert_nothing_captured()
1681
1682     def test_lru(self):
1683         """ reassembly reuses LRU element """
1684
1685         self.vapi.ip_reassembly_set(
1686             timeout_ms=1000000, max_reassemblies=1,
1687             max_reassembly_length=1000,
1688             type=VppEnum.vl_api_ip_reass_type_t.IP_REASS_TYPE_SHALLOW_VIRTUAL,
1689             is_ip6=1, expire_walk_interval_ms=10000)
1690
1691         payload_len = 1000
1692         payload = ""
1693         counter = 0
1694         while len(payload) < payload_len:
1695             payload += "%u " % counter
1696             counter += 1
1697
1698         packet_count = 10
1699
1700         fragments = [f
1701                      for i in range(packet_count)
1702                      for p in (Ether(dst=self.src_if.local_mac,
1703                                      src=self.src_if.remote_mac) /
1704                                IPv6(src=self.src_if.remote_ip6,
1705                                     dst=self.dst_if.remote_ip6) /
1706                                UDP(sport=1234, dport=5678) /
1707                                Raw(payload))
1708                      for f in fragment_rfc8200(p, i, payload_len/4)]
1709
1710         self.pg_enable_capture()
1711         self.src_if.add_stream(fragments)
1712         self.pg_start()
1713         c = self.dst_if.get_capture(len(fragments))
1714         for sent, recvd in zip(fragments, c):
1715             self.assertEqual(sent[IPv6].src, recvd[IPv6].src)
1716             self.assertEqual(sent[IPv6].dst, recvd[IPv6].dst)
1717             self.assertEqual(sent[Raw].payload, recvd[Raw].payload)
1718
1719
1720 class TestIPv4ReassemblyLocalNode(VppTestCase):
1721     """ IPv4 Reassembly for packets coming to ip4-local node """
1722
1723     @classmethod
1724     def setUpClass(cls):
1725         super(TestIPv4ReassemblyLocalNode, cls).setUpClass()
1726
1727         cls.create_pg_interfaces([0])
1728         cls.src_dst_if = cls.pg0
1729
1730         # setup all interfaces
1731         for i in cls.pg_interfaces:
1732             i.admin_up()
1733             i.config_ip4()
1734             i.resolve_arp()
1735
1736         cls.padding = " abcdefghijklmn"
1737         cls.create_stream()
1738         cls.create_fragments()
1739
1740     @classmethod
1741     def tearDownClass(cls):
1742         super(TestIPv4ReassemblyLocalNode, cls).tearDownClass()
1743
1744     def setUp(self):
1745         """ Test setup - force timeout on existing reassemblies """
1746         super(TestIPv4ReassemblyLocalNode, self).setUp()
1747         self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000,
1748                                     max_reassembly_length=1000,
1749                                     expire_walk_interval_ms=10)
1750         self.sleep(.25)
1751         self.vapi.ip_reassembly_set(timeout_ms=1000000, max_reassemblies=1000,
1752                                     max_reassembly_length=1000,
1753                                     expire_walk_interval_ms=10000)
1754
1755     def tearDown(self):
1756         super(TestIPv4ReassemblyLocalNode, self).tearDown()
1757
1758     def show_commands_at_teardown(self):
1759         self.logger.debug(self.vapi.ppcli("show ip4-full-reassembly details"))
1760         self.logger.debug(self.vapi.ppcli("show buffers"))
1761
1762     @classmethod
1763     def create_stream(cls, packet_count=test_packet_count):
1764         """Create input packet stream for defined interface.
1765
1766         :param list packet_sizes: Required packet sizes.
1767         """
1768         for i in range(0, packet_count):
1769             info = cls.create_packet_info(cls.src_dst_if, cls.src_dst_if)
1770             payload = cls.info_to_payload(info)
1771             p = (Ether(dst=cls.src_dst_if.local_mac,
1772                        src=cls.src_dst_if.remote_mac) /
1773                  IP(id=info.index, src=cls.src_dst_if.remote_ip4,
1774                     dst=cls.src_dst_if.local_ip4) /
1775                  ICMP(type='echo-request', id=1234) /
1776                  Raw(payload))
1777             cls.extend_packet(p, 1518, cls.padding)
1778             info.data = p
1779
1780     @classmethod
1781     def create_fragments(cls):
1782         infos = cls._packet_infos
1783         cls.pkt_infos = []
1784         for index, info in infos.items():
1785             p = info.data
1786             # cls.logger.debug(ppp("Packet:",
1787             #                      p.__class__(scapy.compat.raw(p))))
1788             fragments_300 = fragment_rfc791(p, 300)
1789             cls.pkt_infos.append((index, fragments_300))
1790         cls.fragments_300 = [x for (_, frags) in cls.pkt_infos for x in frags]
1791         cls.logger.debug("Fragmented %s packets into %s 300-byte fragments" %
1792                          (len(infos), len(cls.fragments_300)))
1793
1794     def verify_capture(self, capture):
1795         """Verify captured packet stream.
1796
1797         :param list capture: Captured packet stream.
1798         """
1799         info = None
1800         seen = set()
1801         for packet in capture:
1802             try:
1803                 self.logger.debug(ppp("Got packet:", packet))
1804                 ip = packet[IP]
1805                 icmp = packet[ICMP]
1806                 payload_info = self.payload_to_info(packet[Raw])
1807                 packet_index = payload_info.index
1808                 if packet_index in seen:
1809                     raise Exception(ppp("Duplicate packet received", packet))
1810                 seen.add(packet_index)
1811                 self.assertEqual(payload_info.dst, self.src_dst_if.sw_if_index)
1812                 info = self._packet_infos[packet_index]
1813                 self.assertIsNotNone(info)
1814                 self.assertEqual(packet_index, info.index)
1815                 saved_packet = info.data
1816                 self.assertEqual(ip.src, saved_packet[IP].dst)
1817                 self.assertEqual(ip.dst, saved_packet[IP].src)
1818                 self.assertEqual(icmp.type, 0)  # echo reply
1819                 self.assertEqual(icmp.id, saved_packet[ICMP].id)
1820                 self.assertEqual(icmp.payload, saved_packet[ICMP].payload)
1821             except Exception:
1822                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
1823                 raise
1824         for index in self._packet_infos:
1825             self.assertIn(index, seen,
1826                           "Packet with packet_index %d not received" % index)
1827
1828     def test_reassembly(self):
1829         """ basic reassembly """
1830
1831         self.pg_enable_capture()
1832         self.src_dst_if.add_stream(self.fragments_300)
1833         self.pg_start()
1834
1835         packets = self.src_dst_if.get_capture(len(self.pkt_infos))
1836         self.verify_capture(packets)
1837
1838         # run it all again to verify correctness
1839         self.pg_enable_capture()
1840         self.src_dst_if.add_stream(self.fragments_300)
1841         self.pg_start()
1842
1843         packets = self.src_dst_if.get_capture(len(self.pkt_infos))
1844         self.verify_capture(packets)
1845
1846
1847 class TestFIFReassembly(VppTestCase):
1848     """ Fragments in fragments reassembly """
1849
1850     @classmethod
1851     def setUpClass(cls):
1852         super(TestFIFReassembly, cls).setUpClass()
1853
1854         cls.create_pg_interfaces([0, 1])
1855         cls.src_if = cls.pg0
1856         cls.dst_if = cls.pg1
1857         for i in cls.pg_interfaces:
1858             i.admin_up()
1859             i.config_ip4()
1860             i.resolve_arp()
1861             i.config_ip6()
1862             i.resolve_ndp()
1863
1864         cls.packet_sizes = [64, 512, 1518, 9018]
1865         cls.padding = " abcdefghijklmn"
1866
1867     @classmethod
1868     def tearDownClass(cls):
1869         super(TestFIFReassembly, cls).tearDownClass()
1870
1871     def setUp(self):
1872         """ Test setup - force timeout on existing reassemblies """
1873         super(TestFIFReassembly, self).setUp()
1874         self.vapi.ip_reassembly_enable_disable(
1875             sw_if_index=self.src_if.sw_if_index, enable_ip4=True,
1876             enable_ip6=True)
1877         self.vapi.ip_reassembly_enable_disable(
1878             sw_if_index=self.dst_if.sw_if_index, enable_ip4=True,
1879             enable_ip6=True)
1880         self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000,
1881                                     max_reassembly_length=1000,
1882                                     expire_walk_interval_ms=10)
1883         self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000,
1884                                     max_reassembly_length=1000,
1885                                     expire_walk_interval_ms=10, is_ip6=1)
1886         self.sleep(.25)
1887         self.vapi.ip_reassembly_set(timeout_ms=1000000, max_reassemblies=1000,
1888                                     max_reassembly_length=1000,
1889                                     expire_walk_interval_ms=10000)
1890         self.vapi.ip_reassembly_set(timeout_ms=1000000, max_reassemblies=1000,
1891                                     max_reassembly_length=1000,
1892                                     expire_walk_interval_ms=10000, is_ip6=1)
1893
1894     def tearDown(self):
1895         super(TestFIFReassembly, self).tearDown()
1896
1897     def show_commands_at_teardown(self):
1898         self.logger.debug(self.vapi.ppcli("show ip4-full-reassembly details"))
1899         self.logger.debug(self.vapi.ppcli("show ip6-full-reassembly details"))
1900         self.logger.debug(self.vapi.ppcli("show buffers"))
1901
1902     def verify_capture(self, capture, ip_class, dropped_packet_indexes=[]):
1903         """Verify captured packet stream.
1904
1905         :param list capture: Captured packet stream.
1906         """
1907         info = None
1908         seen = set()
1909         for packet in capture:
1910             try:
1911                 self.logger.debug(ppp("Got packet:", packet))
1912                 ip = packet[ip_class]
1913                 udp = packet[UDP]
1914                 payload_info = self.payload_to_info(packet[Raw])
1915                 packet_index = payload_info.index
1916                 self.assertTrue(
1917                     packet_index not in dropped_packet_indexes,
1918                     ppp("Packet received, but should be dropped:", packet))
1919                 if packet_index in seen:
1920                     raise Exception(ppp("Duplicate packet received", packet))
1921                 seen.add(packet_index)
1922                 self.assertEqual(payload_info.dst, self.dst_if.sw_if_index)
1923                 info = self._packet_infos[packet_index]
1924                 self.assertTrue(info is not None)
1925                 self.assertEqual(packet_index, info.index)
1926                 saved_packet = info.data
1927                 self.assertEqual(ip.src, saved_packet[ip_class].src)
1928                 self.assertEqual(ip.dst, saved_packet[ip_class].dst)
1929                 self.assertEqual(udp.payload, saved_packet[UDP].payload)
1930             except Exception:
1931                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
1932                 raise
1933         for index in self._packet_infos:
1934             self.assertTrue(index in seen or index in dropped_packet_indexes,
1935                             "Packet with packet_index %d not received" % index)
1936
1937     def test_fif4(self):
1938         """ Fragments in fragments (4o4) """
1939
1940         # TODO this should be ideally in setUpClass, but then we hit a bug
1941         # with VppIpRoute incorrectly reporting it's present when it's not
1942         # so we need to manually remove the vpp config, thus we cannot have
1943         # it shared for multiple test cases
1944         self.tun_ip4 = "1.1.1.2"
1945
1946         self.gre4 = VppGreInterface(self, self.src_if.local_ip4, self.tun_ip4)
1947         self.gre4.add_vpp_config()
1948         self.gre4.admin_up()
1949         self.gre4.config_ip4()
1950
1951         self.vapi.ip_reassembly_enable_disable(
1952             sw_if_index=self.gre4.sw_if_index, enable_ip4=True)
1953
1954         self.route4 = VppIpRoute(self, self.tun_ip4, 32,
1955                                  [VppRoutePath(self.src_if.remote_ip4,
1956                                                self.src_if.sw_if_index)])
1957         self.route4.add_vpp_config()
1958
1959         self.reset_packet_infos()
1960         for i in range(test_packet_count):
1961             info = self.create_packet_info(self.src_if, self.dst_if)
1962             payload = self.info_to_payload(info)
1963             # Ethernet header here is only for size calculation, thus it
1964             # doesn't matter how it's initialized. This is to ensure that
1965             # reassembled packet is not > 9000 bytes, so that it's not dropped
1966             p = (Ether() /
1967                  IP(id=i, src=self.src_if.remote_ip4,
1968                     dst=self.dst_if.remote_ip4) /
1969                  UDP(sport=1234, dport=5678) /
1970                  Raw(payload))
1971             size = self.packet_sizes[(i // 2) % len(self.packet_sizes)]
1972             self.extend_packet(p, size, self.padding)
1973             info.data = p[IP]  # use only IP part, without ethernet header
1974
1975         fragments = [x for _, p in self._packet_infos.items()
1976                      for x in fragment_rfc791(p.data, 400)]
1977
1978         encapped_fragments = \
1979             [Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac) /
1980              IP(src=self.tun_ip4, dst=self.src_if.local_ip4) /
1981                 GRE() /
1982                 p
1983                 for p in fragments]
1984
1985         fragmented_encapped_fragments = \
1986             [x for p in encapped_fragments
1987              for x in fragment_rfc791(p, 200)]
1988
1989         self.src_if.add_stream(fragmented_encapped_fragments)
1990
1991         self.pg_enable_capture(self.pg_interfaces)
1992         self.pg_start()
1993
1994         self.src_if.assert_nothing_captured()
1995         packets = self.dst_if.get_capture(len(self._packet_infos))
1996         self.verify_capture(packets, IP)
1997
1998         # TODO remove gre vpp config by hand until VppIpRoute gets fixed
1999         # so that it's query_vpp_config() works as it should
2000         self.gre4.remove_vpp_config()
2001         self.logger.debug(self.vapi.ppcli("show interface"))
2002
2003     def test_fif6(self):
2004         """ Fragments in fragments (6o6) """
2005         # TODO this should be ideally in setUpClass, but then we hit a bug
2006         # with VppIpRoute incorrectly reporting it's present when it's not
2007         # so we need to manually remove the vpp config, thus we cannot have
2008         # it shared for multiple test cases
2009         self.tun_ip6 = "1002::1"
2010
2011         self.gre6 = VppGreInterface(self, self.src_if.local_ip6, self.tun_ip6)
2012         self.gre6.add_vpp_config()
2013         self.gre6.admin_up()
2014         self.gre6.config_ip6()
2015
2016         self.vapi.ip_reassembly_enable_disable(
2017             sw_if_index=self.gre6.sw_if_index, enable_ip6=True)
2018
2019         self.route6 = VppIpRoute(self, self.tun_ip6, 128,
2020                                  [VppRoutePath(
2021                                      self.src_if.remote_ip6,
2022                                      self.src_if.sw_if_index)])
2023         self.route6.add_vpp_config()
2024
2025         self.reset_packet_infos()
2026         for i in range(test_packet_count):
2027             info = self.create_packet_info(self.src_if, self.dst_if)
2028             payload = self.info_to_payload(info)
2029             # Ethernet header here is only for size calculation, thus it
2030             # doesn't matter how it's initialized. This is to ensure that
2031             # reassembled packet is not > 9000 bytes, so that it's not dropped
2032             p = (Ether() /
2033                  IPv6(src=self.src_if.remote_ip6, dst=self.dst_if.remote_ip6) /
2034                  UDP(sport=1234, dport=5678) /
2035                  Raw(payload))
2036             size = self.packet_sizes[(i // 2) % len(self.packet_sizes)]
2037             self.extend_packet(p, size, self.padding)
2038             info.data = p[IPv6]  # use only IPv6 part, without ethernet header
2039
2040         fragments = [x for _, i in self._packet_infos.items()
2041                      for x in fragment_rfc8200(
2042                          i.data, i.index, 400)]
2043
2044         encapped_fragments = \
2045             [Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac) /
2046              IPv6(src=self.tun_ip6, dst=self.src_if.local_ip6) /
2047                 GRE() /
2048                 p
2049                 for p in fragments]
2050
2051         fragmented_encapped_fragments = \
2052             [x for p in encapped_fragments for x in (
2053                 fragment_rfc8200(
2054                     p,
2055                     2 * len(self._packet_infos) + p[IPv6ExtHdrFragment].id,
2056                     200)
2057                 if IPv6ExtHdrFragment in p else [p]
2058             )
2059             ]
2060
2061         self.src_if.add_stream(fragmented_encapped_fragments)
2062
2063         self.pg_enable_capture(self.pg_interfaces)
2064         self.pg_start()
2065
2066         self.src_if.assert_nothing_captured()
2067         packets = self.dst_if.get_capture(len(self._packet_infos))
2068         self.verify_capture(packets, IPv6)
2069
2070         # TODO remove gre vpp config by hand until VppIpRoute gets fixed
2071         # so that it's query_vpp_config() works as it should
2072         self.gre6.remove_vpp_config()
2073
2074
2075 if __name__ == '__main__':
2076     unittest.main(testRunner=VppTestRunner)