test/test_reassembly.py

   1 #!/usr/bin/env python
   2 import unittest
   3 from random import shuffle
   4
   5 from framework import VppTestCase, VppTestRunner
   6
   7 from scapy.packet import Raw
   8 from scapy.layers.l2 import Ether, GRE
   9 from scapy.layers.inet import IP, UDP, ICMP
  10 from util import ppp, fragment_rfc791, fragment_rfc8200
  11 from scapy.layers.inet6 import IPv6, IPv6ExtHdrFragment, ICMPv6ParamProblem,\
  12     ICMPv6TimeExceeded
  13 from vpp_gre_interface import VppGreInterface, VppGre6Interface
  14 from vpp_ip import DpoProto
  15 from vpp_ip_route import VppIpRoute, VppRoutePath
  16
  17 test_packet_count = 257
  18
  19
  20 class TestIPv4Reassembly(VppTestCase):
  21     """ IPv4 Reassembly """
  22
  23     @classmethod
  24     def setUpClass(cls):
  25         super(TestIPv4Reassembly, cls).setUpClass()
  26
  27         cls.create_pg_interfaces([0, 1])
  28         cls.src_if = cls.pg0
  29         cls.dst_if = cls.pg1
  30
  31         # setup all interfaces
  32         for i in cls.pg_interfaces:
  33             i.admin_up()
  34             i.config_ip4()
  35             i.resolve_arp()
  36
  37         # packet sizes
  38         cls.packet_sizes = [64, 512, 1518, 9018]
  39         cls.padding = " abcdefghijklmn"
  40         cls.create_stream(cls.packet_sizes)
  41         cls.create_fragments()
  42
  43     def setUp(self):
  44         """ Test setup - force timeout on existing reassemblies """
  45         super(TestIPv4Reassembly, self).setUp()
  46         self.vapi.ip_reassembly_enable_disable(
  47             sw_if_index=self.src_if.sw_if_index, enable_ip4=True)
  48         self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000,
  49                                     expire_walk_interval_ms=10)
  50         self.sleep(.25)
  51         self.vapi.ip_reassembly_set(timeout_ms=1000000, max_reassemblies=1000,
  52                                     expire_walk_interval_ms=10000)
  53
  54     def tearDown(self):
  55         super(TestIPv4Reassembly, self).tearDown()
  56         self.logger.debug(self.vapi.ppcli("show ip4-reassembly details"))
  57
  58     @classmethod
  59     def create_stream(cls, packet_sizes, packet_count=test_packet_count):
  60         """Create input packet stream for defined interface.
  61
  62         :param list packet_sizes: Required packet sizes.
  63         """
  64         for i in range(0, packet_count):
  65             info = cls.create_packet_info(cls.src_if, cls.src_if)
  66             payload = cls.info_to_payload(info)
  67             p = (Ether(dst=cls.src_if.local_mac, src=cls.src_if.remote_mac) /
  68                  IP(id=info.index, src=cls.src_if.remote_ip4,
  69                     dst=cls.dst_if.remote_ip4) /
  70                  UDP(sport=1234, dport=5678) /
  71                  Raw(payload))
  72             size = packet_sizes[(i // 2) % len(packet_sizes)]
  73             cls.extend_packet(p, size, cls.padding)
  74             info.data = p
  75
  76     @classmethod
  77     def create_fragments(cls):
  78         infos = cls._packet_infos
  79         cls.pkt_infos = []
  80         for index, info in infos.iteritems():
  81             p = info.data
  82             # cls.logger.debug(ppp("Packet:", p.__class__(str(p))))
  83             fragments_400 = fragment_rfc791(p, 400)
  84             fragments_300 = fragment_rfc791(p, 300)
  85             fragments_200 = [
  86                 x for f in fragments_400 for x in fragment_rfc791(f, 200)]
  87             cls.pkt_infos.append(
  88                 (index, fragments_400, fragments_300, fragments_200))
  89         cls.fragments_400 = [
  90             x for (_, frags, _, _) in cls.pkt_infos for x in frags]
  91         cls.fragments_300 = [
  92             x for (_, _, frags, _) in cls.pkt_infos for x in frags]
  93         cls.fragments_200 = [
  94             x for (_, _, _, frags) in cls.pkt_infos for x in frags]
  95         cls.logger.debug("Fragmented %s packets into %s 400-byte fragments, "
  96                          "%s 300-byte fragments and %s 200-byte fragments" %
  97                          (len(infos), len(cls.fragments_400),
  98                              len(cls.fragments_300), len(cls.fragments_200)))
  99
 100     def verify_capture(self, capture, dropped_packet_indexes=[]):
 101         """Verify captured packet stream.
 102
 103         :param list capture: Captured packet stream.
 104         """
 105         info = None
 106         seen = set()
 107         for packet in capture:
 108             try:
 109                 self.logger.debug(ppp("Got packet:", packet))
 110                 ip = packet[IP]
 111                 udp = packet[UDP]
 112                 payload_info = self.payload_to_info(str(packet[Raw]))
 113                 packet_index = payload_info.index
 114                 self.assertTrue(
 115                     packet_index not in dropped_packet_indexes,
 116                     ppp("Packet received, but should be dropped:", packet))
 117                 if packet_index in seen:
 118                     raise Exception(ppp("Duplicate packet received", packet))
 119                 seen.add(packet_index)
 120                 self.assertEqual(payload_info.dst, self.src_if.sw_if_index)
 121                 info = self._packet_infos[packet_index]
 122                 self.assertTrue(info is not None)
 123                 self.assertEqual(packet_index, info.index)
 124                 saved_packet = info.data
 125                 self.assertEqual(ip.src, saved_packet[IP].src)
 126                 self.assertEqual(ip.dst, saved_packet[IP].dst)
 127                 self.assertEqual(udp.payload, saved_packet[UDP].payload)
 128             except Exception:
 129                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
 130                 raise
 131         for index in self._packet_infos:
 132             self.assertTrue(index in seen or index in dropped_packet_indexes,
 133                             "Packet with packet_index %d not received" % index)
 134
 135     def test_reassembly(self):
 136         """ basic reassembly """
 137
 138         self.pg_enable_capture()
 139         self.src_if.add_stream(self.fragments_200)
 140         self.pg_start()
 141
 142         packets = self.dst_if.get_capture(len(self.pkt_infos))
 143         self.verify_capture(packets)
 144         self.src_if.assert_nothing_captured()
 145
 146         # run it all again to verify correctness
 147         self.pg_enable_capture()
 148         self.src_if.add_stream(self.fragments_200)
 149         self.pg_start()
 150
 151         packets = self.dst_if.get_capture(len(self.pkt_infos))
 152         self.verify_capture(packets)
 153         self.src_if.assert_nothing_captured()
 154
 155     def test_reversed(self):
 156         """ reverse order reassembly """
 157
 158         fragments = list(self.fragments_200)
 159         fragments.reverse()
 160
 161         self.pg_enable_capture()
 162         self.src_if.add_stream(fragments)
 163         self.pg_start()
 164
 165         packets = self.dst_if.get_capture(len(self.packet_infos))
 166         self.verify_capture(packets)
 167         self.src_if.assert_nothing_captured()
 168
 169         # run it all again to verify correctness
 170         self.pg_enable_capture()
 171         self.src_if.add_stream(fragments)
 172         self.pg_start()
 173
 174         packets = self.dst_if.get_capture(len(self.packet_infos))
 175         self.verify_capture(packets)
 176         self.src_if.assert_nothing_captured()
 177
 178     def test_random(self):
 179         """ random order reassembly """
 180
 181         fragments = list(self.fragments_200)
 182         shuffle(fragments)
 183
 184         self.pg_enable_capture()
 185         self.src_if.add_stream(fragments)
 186         self.pg_start()
 187
 188         packets = self.dst_if.get_capture(len(self.packet_infos))
 189         self.verify_capture(packets)
 190         self.src_if.assert_nothing_captured()
 191
 192         # run it all again to verify correctness
 193         self.pg_enable_capture()
 194         self.src_if.add_stream(fragments)
 195         self.pg_start()
 196
 197         packets = self.dst_if.get_capture(len(self.packet_infos))
 198         self.verify_capture(packets)
 199         self.src_if.assert_nothing_captured()
 200
 201     def test_duplicates(self):
 202         """ duplicate fragments """
 203
 204         fragments = [
 205             x for (_, frags, _, _) in self.pkt_infos
 206             for x in frags
 207             for _ in range(0, min(2, len(frags)))
 208         ]
 209
 210         self.pg_enable_capture()
 211         self.src_if.add_stream(fragments)
 212         self.pg_start()
 213
 214         packets = self.dst_if.get_capture(len(self.pkt_infos))
 215         self.verify_capture(packets)
 216         self.src_if.assert_nothing_captured()
 217
 218     def test_overlap1(self):
 219         """ overlapping fragments case #1 """
 220
 221         fragments = []
 222         for _, _, frags_300, frags_200 in self.pkt_infos:
 223             if len(frags_300) == 1:
 224                 fragments.extend(frags_300)
 225             else:
 226                 for i, j in zip(frags_200, frags_300):
 227                     fragments.extend(i)
 228                     fragments.extend(j)
 229
 230         self.pg_enable_capture()
 231         self.src_if.add_stream(fragments)
 232         self.pg_start()
 233
 234         packets = self.dst_if.get_capture(len(self.pkt_infos))
 235         self.verify_capture(packets)
 236         self.src_if.assert_nothing_captured()
 237
 238         # run it all to verify correctness
 239         self.pg_enable_capture()
 240         self.src_if.add_stream(fragments)
 241         self.pg_start()
 242
 243         packets = self.dst_if.get_capture(len(self.pkt_infos))
 244         self.verify_capture(packets)
 245         self.src_if.assert_nothing_captured()
 246
 247     def test_overlap2(self):
 248         """ overlapping fragments case #2 """
 249
 250         fragments = []
 251         for _, _, frags_300, frags_200 in self.pkt_infos:
 252             if len(frags_300) == 1:
 253                 fragments.extend(frags_300)
 254             else:
 255                 # care must be taken here so that there are no fragments
 256                 # received by vpp after reassembly is finished, otherwise
 257                 # new reassemblies will be started and packet generator will
 258                 # freak out when it detects unfreed buffers
 259                 zipped = zip(frags_300, frags_200)
 260                 for i, j in zipped[:-1]:
 261                     fragments.extend(i)
 262                     fragments.extend(j)
 263                 fragments.append(zipped[-1][0])
 264
 265         self.pg_enable_capture()
 266         self.src_if.add_stream(fragments)
 267         self.pg_start()
 268
 269         packets = self.dst_if.get_capture(len(self.pkt_infos))
 270         self.verify_capture(packets)
 271         self.src_if.assert_nothing_captured()
 272
 273         # run it all to verify correctness
 274         self.pg_enable_capture()
 275         self.src_if.add_stream(fragments)
 276         self.pg_start()
 277
 278         packets = self.dst_if.get_capture(len(self.pkt_infos))
 279         self.verify_capture(packets)
 280         self.src_if.assert_nothing_captured()
 281
 282     def test_timeout_inline(self):
 283         """ timeout (inline) """
 284
 285         dropped_packet_indexes = set(
 286             index for (index, frags, _, _) in self.pkt_infos if len(frags) > 1
 287         )
 288
 289         self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000,
 290                                     expire_walk_interval_ms=10000)
 291
 292         self.pg_enable_capture()
 293         self.src_if.add_stream(self.fragments_400)
 294         self.pg_start()
 295
 296         packets = self.dst_if.get_capture(
 297             len(self.pkt_infos) - len(dropped_packet_indexes))
 298         self.verify_capture(packets, dropped_packet_indexes)
 299         self.src_if.assert_nothing_captured()
 300
 301     def test_timeout_cleanup(self):
 302         """ timeout (cleanup) """
 303
 304         # whole packets + fragmented packets sans last fragment
 305         fragments = [
 306             x for (_, frags_400, _, _) in self.pkt_infos
 307             for x in frags_400[:-1 if len(frags_400) > 1 else None]
 308         ]
 309
 310         # last fragments for fragmented packets
 311         fragments2 = [frags_400[-1]
 312                       for (_, frags_400, _, _) in self.pkt_infos
 313                       if len(frags_400) > 1]
 314
 315         dropped_packet_indexes = set(
 316             index for (index, frags_400, _, _) in self.pkt_infos
 317             if len(frags_400) > 1)
 318
 319         self.vapi.ip_reassembly_set(timeout_ms=100, max_reassemblies=1000,
 320                                     expire_walk_interval_ms=50)
 321
 322         self.pg_enable_capture()
 323         self.src_if.add_stream(fragments)
 324         self.pg_start()
 325
 326         self.sleep(.25, "wait before sending rest of fragments")
 327
 328         self.src_if.add_stream(fragments2)
 329         self.pg_start()
 330
 331         packets = self.dst_if.get_capture(
 332             len(self.pkt_infos) - len(dropped_packet_indexes))
 333         self.verify_capture(packets, dropped_packet_indexes)
 334         self.src_if.assert_nothing_captured()
 335
 336     def test_disabled(self):
 337         """ reassembly disabled """
 338
 339         dropped_packet_indexes = set(
 340             index for (index, frags_400, _, _) in self.pkt_infos
 341             if len(frags_400) > 1)
 342
 343         self.vapi.ip_reassembly_set(timeout_ms=1000, max_reassemblies=0,
 344                                     expire_walk_interval_ms=10000)
 345
 346         self.pg_enable_capture()
 347         self.src_if.add_stream(self.fragments_400)
 348         self.pg_start()
 349
 350         packets = self.dst_if.get_capture(
 351             len(self.pkt_infos) - len(dropped_packet_indexes))
 352         self.verify_capture(packets, dropped_packet_indexes)
 353         self.src_if.assert_nothing_captured()
 354
 355
 356 class TestIPv6Reassembly(VppTestCase):
 357     """ IPv6 Reassembly """
 358
 359     @classmethod
 360     def setUpClass(cls):
 361         super(TestIPv6Reassembly, cls).setUpClass()
 362
 363         cls.create_pg_interfaces([0, 1])
 364         cls.src_if = cls.pg0
 365         cls.dst_if = cls.pg1
 366
 367         # setup all interfaces
 368         for i in cls.pg_interfaces:
 369             i.admin_up()
 370             i.config_ip6()
 371             i.resolve_ndp()
 372
 373         # packet sizes
 374         cls.packet_sizes = [64, 512, 1518, 9018]
 375         cls.padding = " abcdefghijklmn"
 376         cls.create_stream(cls.packet_sizes)
 377         cls.create_fragments()
 378
 379     def setUp(self):
 380         """ Test setup - force timeout on existing reassemblies """
 381         super(TestIPv6Reassembly, self).setUp()
 382         self.vapi.ip_reassembly_enable_disable(
 383             sw_if_index=self.src_if.sw_if_index, enable_ip6=True)
 384         self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000,
 385                                     expire_walk_interval_ms=10, is_ip6=1)
 386         self.sleep(.25)
 387         self.vapi.ip_reassembly_set(timeout_ms=1000000, max_reassemblies=1000,
 388                                     expire_walk_interval_ms=10000, is_ip6=1)
 389         self.logger.debug(self.vapi.ppcli("show ip6-reassembly details"))
 390
 391     def tearDown(self):
 392         super(TestIPv6Reassembly, self).tearDown()
 393         self.logger.debug(self.vapi.ppcli("show ip6-reassembly details"))
 394
 395     @classmethod
 396     def create_stream(cls, packet_sizes, packet_count=test_packet_count):
 397         """Create input packet stream for defined interface.
 398
 399         :param list packet_sizes: Required packet sizes.
 400         """
 401         for i in range(0, packet_count):
 402             info = cls.create_packet_info(cls.src_if, cls.src_if)
 403             payload = cls.info_to_payload(info)
 404             p = (Ether(dst=cls.src_if.local_mac, src=cls.src_if.remote_mac) /
 405                  IPv6(src=cls.src_if.remote_ip6,
 406                       dst=cls.dst_if.remote_ip6) /
 407                  UDP(sport=1234, dport=5678) /
 408                  Raw(payload))
 409             size = packet_sizes[(i // 2) % len(packet_sizes)]
 410             cls.extend_packet(p, size, cls.padding)
 411             info.data = p
 412
 413     @classmethod
 414     def create_fragments(cls):
 415         infos = cls._packet_infos
 416         cls.pkt_infos = []
 417         for index, info in infos.iteritems():
 418             p = info.data
 419             # cls.logger.debug(ppp("Packet:", p.__class__(str(p))))
 420             fragments_400 = fragment_rfc8200(p, info.index, 400)
 421             fragments_300 = fragment_rfc8200(p, info.index, 300)
 422             cls.pkt_infos.append((index, fragments_400, fragments_300))
 423         cls.fragments_400 = [
 424             x for _, frags, _ in cls.pkt_infos for x in frags]
 425         cls.fragments_300 = [
 426             x for _, _, frags in cls.pkt_infos for x in frags]
 427         cls.logger.debug("Fragmented %s packets into %s 400-byte fragments, "
 428                          "and %s 300-byte fragments" %
 429                          (len(infos), len(cls.fragments_400),
 430                              len(cls.fragments_300)))
 431
 432     def verify_capture(self, capture, dropped_packet_indexes=[]):
 433         """Verify captured packet strea .
 434
 435         :param list capture: Captured packet stream.
 436         """
 437         info = None
 438         seen = set()
 439         for packet in capture:
 440             try:
 441                 self.logger.debug(ppp("Got packet:", packet))
 442                 ip = packet[IPv6]
 443                 udp = packet[UDP]
 444                 payload_info = self.payload_to_info(str(packet[Raw]))
 445                 packet_index = payload_info.index
 446                 self.assertTrue(
 447                     packet_index not in dropped_packet_indexes,
 448                     ppp("Packet received, but should be dropped:", packet))
 449                 if packet_index in seen:
 450                     raise Exception(ppp("Duplicate packet received", packet))
 451                 seen.add(packet_index)
 452                 self.assertEqual(payload_info.dst, self.src_if.sw_if_index)
 453                 info = self._packet_infos[packet_index]
 454                 self.assertTrue(info is not None)
 455                 self.assertEqual(packet_index, info.index)
 456                 saved_packet = info.data
 457                 self.assertEqual(ip.src, saved_packet[IPv6].src)
 458                 self.assertEqual(ip.dst, saved_packet[IPv6].dst)
 459                 self.assertEqual(udp.payload, saved_packet[UDP].payload)
 460             except Exception:
 461                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
 462                 raise
 463         for index in self._packet_infos:
 464             self.assertTrue(index in seen or index in dropped_packet_indexes,
 465                             "Packet with packet_index %d not received" % index)
 466
 467     def test_reassembly(self):
 468         """ basic reassembly """
 469
 470         self.pg_enable_capture()
 471         self.src_if.add_stream(self.fragments_400)
 472         self.pg_start()
 473
 474         packets = self.dst_if.get_capture(len(self.pkt_infos))
 475         self.verify_capture(packets)
 476         self.src_if.assert_nothing_captured()
 477
 478         # run it all again to verify correctness
 479         self.pg_enable_capture()
 480         self.src_if.add_stream(self.fragments_400)
 481         self.pg_start()
 482
 483         packets = self.dst_if.get_capture(len(self.pkt_infos))
 484         self.verify_capture(packets)
 485         self.src_if.assert_nothing_captured()
 486
 487     def test_reversed(self):
 488         """ reverse order reassembly """
 489
 490         fragments = list(self.fragments_400)
 491         fragments.reverse()
 492
 493         self.pg_enable_capture()
 494         self.src_if.add_stream(fragments)
 495         self.pg_start()
 496
 497         packets = self.dst_if.get_capture(len(self.pkt_infos))
 498         self.verify_capture(packets)
 499         self.src_if.assert_nothing_captured()
 500
 501         # run it all again to verify correctness
 502         self.pg_enable_capture()
 503         self.src_if.add_stream(fragments)
 504         self.pg_start()
 505
 506         packets = self.dst_if.get_capture(len(self.pkt_infos))
 507         self.verify_capture(packets)
 508         self.src_if.assert_nothing_captured()
 509
 510     def test_random(self):
 511         """ random order reassembly """
 512
 513         fragments = list(self.fragments_400)
 514         shuffle(fragments)
 515
 516         self.pg_enable_capture()
 517         self.src_if.add_stream(fragments)
 518         self.pg_start()
 519
 520         packets = self.dst_if.get_capture(len(self.pkt_infos))
 521         self.verify_capture(packets)
 522         self.src_if.assert_nothing_captured()
 523
 524         # run it all again to verify correctness
 525         self.pg_enable_capture()
 526         self.src_if.add_stream(fragments)
 527         self.pg_start()
 528
 529         packets = self.dst_if.get_capture(len(self.pkt_infos))
 530         self.verify_capture(packets)
 531         self.src_if.assert_nothing_captured()
 532
 533     def test_duplicates(self):
 534         """ duplicate fragments """
 535
 536         fragments = [
 537             x for (_, frags, _) in self.pkt_infos
 538             for x in frags
 539             for _ in range(0, min(2, len(frags)))
 540         ]
 541
 542         self.pg_enable_capture()
 543         self.src_if.add_stream(fragments)
 544         self.pg_start()
 545
 546         packets = self.dst_if.get_capture(len(self.pkt_infos))
 547         self.verify_capture(packets)
 548         self.src_if.assert_nothing_captured()
 549
 550     def test_overlap1(self):
 551         """ overlapping fragments case #1 """
 552
 553         fragments = []
 554         for _, frags_400, frags_300 in self.pkt_infos:
 555             if len(frags_300) == 1:
 556                 fragments.extend(frags_400)
 557             else:
 558                 for i, j in zip(frags_300, frags_400):
 559                     fragments.extend(i)
 560                     fragments.extend(j)
 561
 562         dropped_packet_indexes = set(
 563             index for (index, _, frags) in self.pkt_infos if len(frags) > 1
 564         )
 565
 566         self.pg_enable_capture()
 567         self.src_if.add_stream(fragments)
 568         self.pg_start()
 569
 570         packets = self.dst_if.get_capture(
 571             len(self.pkt_infos) - len(dropped_packet_indexes))
 572         self.verify_capture(packets, dropped_packet_indexes)
 573         self.src_if.assert_nothing_captured()
 574
 575     def test_overlap2(self):
 576         """ overlapping fragments case #2 """
 577
 578         fragments = []
 579         for _, frags_400, frags_300 in self.pkt_infos:
 580             if len(frags_400) == 1:
 581                 fragments.extend(frags_400)
 582             else:
 583                 # care must be taken here so that there are no fragments
 584                 # received by vpp after reassembly is finished, otherwise
 585                 # new reassemblies will be started and packet generator will
 586                 # freak out when it detects unfreed buffers
 587                 zipped = zip(frags_400, frags_300)
 588                 for i, j in zipped[:-1]:
 589                     fragments.extend(i)
 590                     fragments.extend(j)
 591                 fragments.append(zipped[-1][0])
 592
 593         dropped_packet_indexes = set(
 594             index for (index, _, frags) in self.pkt_infos if len(frags) > 1
 595         )
 596
 597         self.pg_enable_capture()
 598         self.src_if.add_stream(fragments)
 599         self.pg_start()
 600
 601         packets = self.dst_if.get_capture(
 602             len(self.pkt_infos) - len(dropped_packet_indexes))
 603         self.verify_capture(packets, dropped_packet_indexes)
 604         self.src_if.assert_nothing_captured()
 605
 606     def test_timeout_inline(self):
 607         """ timeout (inline) """
 608
 609         dropped_packet_indexes = set(
 610             index for (index, frags, _) in self.pkt_infos if len(frags) > 1
 611         )
 612
 613         self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000,
 614                                     expire_walk_interval_ms=10000, is_ip6=1)
 615
 616         self.pg_enable_capture()
 617         self.src_if.add_stream(self.fragments_400)
 618         self.pg_start()
 619
 620         packets = self.dst_if.get_capture(
 621             len(self.pkt_infos) - len(dropped_packet_indexes))
 622         self.verify_capture(packets, dropped_packet_indexes)
 623         pkts = self.src_if.get_capture(
 624             expected_count=len(dropped_packet_indexes))
 625         for icmp in pkts:
 626             self.assertIn(ICMPv6TimeExceeded, icmp)
 627             self.assertIn(IPv6ExtHdrFragment, icmp)
 628             self.assertIn(icmp[IPv6ExtHdrFragment].id, dropped_packet_indexes)
 629             dropped_packet_indexes.remove(icmp[IPv6ExtHdrFragment].id)
 630
 631     def test_timeout_cleanup(self):
 632         """ timeout (cleanup) """
 633
 634         # whole packets + fragmented packets sans last fragment
 635         fragments = [
 636             x for (_, frags_400, _) in self.pkt_infos
 637             for x in frags_400[:-1 if len(frags_400) > 1 else None]
 638         ]
 639
 640         # last fragments for fragmented packets
 641         fragments2 = [frags_400[-1]
 642                       for (_, frags_400, _) in self.pkt_infos
 643                       if len(frags_400) > 1]
 644
 645         dropped_packet_indexes = set(
 646             index for (index, frags_400, _) in self.pkt_infos
 647             if len(frags_400) > 1)
 648
 649         self.vapi.ip_reassembly_set(timeout_ms=100, max_reassemblies=1000,
 650                                     expire_walk_interval_ms=50)
 651
 652         self.vapi.ip_reassembly_set(timeout_ms=100, max_reassemblies=1000,
 653                                     expire_walk_interval_ms=50, is_ip6=1)
 654
 655         self.pg_enable_capture()
 656         self.src_if.add_stream(fragments)
 657         self.pg_start()
 658
 659         self.sleep(.25, "wait before sending rest of fragments")
 660
 661         self.src_if.add_stream(fragments2)
 662         self.pg_start()
 663
 664         packets = self.dst_if.get_capture(
 665             len(self.pkt_infos) - len(dropped_packet_indexes))
 666         self.verify_capture(packets, dropped_packet_indexes)
 667         pkts = self.src_if.get_capture(
 668             expected_count=len(dropped_packet_indexes))
 669         for icmp in pkts:
 670             self.assertIn(ICMPv6TimeExceeded, icmp)
 671             self.assertIn(IPv6ExtHdrFragment, icmp)
 672             self.assertIn(icmp[IPv6ExtHdrFragment].id, dropped_packet_indexes)
 673             dropped_packet_indexes.remove(icmp[IPv6ExtHdrFragment].id)
 674
 675     def test_disabled(self):
 676         """ reassembly disabled """
 677
 678         dropped_packet_indexes = set(
 679             index for (index, frags_400, _) in self.pkt_infos
 680             if len(frags_400) > 1)
 681
 682         self.vapi.ip_reassembly_set(timeout_ms=1000, max_reassemblies=0,
 683                                     expire_walk_interval_ms=10000, is_ip6=1)
 684
 685         self.pg_enable_capture()
 686         self.src_if.add_stream(self.fragments_400)
 687         self.pg_start()
 688
 689         packets = self.dst_if.get_capture(
 690             len(self.pkt_infos) - len(dropped_packet_indexes))
 691         self.verify_capture(packets, dropped_packet_indexes)
 692         self.src_if.assert_nothing_captured()
 693
 694     def test_missing_upper(self):
 695         """ missing upper layer """
 696         p = (Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac) /
 697              IPv6(src=self.src_if.remote_ip6,
 698                   dst=self.src_if.local_ip6) /
 699              UDP(sport=1234, dport=5678) /
 700              Raw())
 701         self.extend_packet(p, 1000, self.padding)
 702         fragments = fragment_rfc8200(p, 1, 500)
 703         bad_fragment = p.__class__(str(fragments[1]))
 704         bad_fragment[IPv6ExtHdrFragment].nh = 59
 705         bad_fragment[IPv6ExtHdrFragment].offset = 0
 706         self.pg_enable_capture()
 707         self.src_if.add_stream([bad_fragment])
 708         self.pg_start()
 709         pkts = self.src_if.get_capture(expected_count=1)
 710         icmp = pkts[0]
 711         self.assertIn(ICMPv6ParamProblem, icmp)
 712         self.assert_equal(icmp[ICMPv6ParamProblem].code, 3, "ICMP code")
 713
 714     def test_invalid_frag_size(self):
 715         """ fragment size not a multiple of 8 """
 716         p = (Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac) /
 717              IPv6(src=self.src_if.remote_ip6,
 718                   dst=self.src_if.local_ip6) /
 719              UDP(sport=1234, dport=5678) /
 720              Raw())
 721         self.extend_packet(p, 1000, self.padding)
 722         fragments = fragment_rfc8200(p, 1, 500)
 723         bad_fragment = fragments[0]
 724         self.extend_packet(bad_fragment, len(bad_fragment) + 5)
 725         self.pg_enable_capture()
 726         self.src_if.add_stream([bad_fragment])
 727         self.pg_start()
 728         pkts = self.src_if.get_capture(expected_count=1)
 729         icmp = pkts[0]
 730         self.assertIn(ICMPv6ParamProblem, icmp)
 731         self.assert_equal(icmp[ICMPv6ParamProblem].code, 0, "ICMP code")
 732
 733     def test_invalid_packet_size(self):
 734         """ total packet size > 65535 """
 735         p = (Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac) /
 736              IPv6(src=self.src_if.remote_ip6,
 737                   dst=self.src_if.local_ip6) /
 738              UDP(sport=1234, dport=5678) /
 739              Raw())
 740         self.extend_packet(p, 1000, self.padding)
 741         fragments = fragment_rfc8200(p, 1, 500)
 742         bad_fragment = fragments[1]
 743         bad_fragment[IPv6ExtHdrFragment].offset = 65500
 744         self.pg_enable_capture()
 745         self.src_if.add_stream([bad_fragment])
 746         self.pg_start()
 747         pkts = self.src_if.get_capture(expected_count=1)
 748         icmp = pkts[0]
 749         self.assertIn(ICMPv6ParamProblem, icmp)
 750         self.assert_equal(icmp[ICMPv6ParamProblem].code, 0, "ICMP code")
 751
 752
 753 class TestIPv4ReassemblyLocalNode(VppTestCase):
 754     """ IPv4 Reassembly for packets coming to ip4-local node """
 755
 756     @classmethod
 757     def setUpClass(cls):
 758         super(TestIPv4ReassemblyLocalNode, cls).setUpClass()
 759
 760         cls.create_pg_interfaces([0])
 761         cls.src_dst_if = cls.pg0
 762
 763         # setup all interfaces
 764         for i in cls.pg_interfaces:
 765             i.admin_up()
 766             i.config_ip4()
 767             i.resolve_arp()
 768
 769         cls.padding = " abcdefghijklmn"
 770         cls.create_stream()
 771         cls.create_fragments()
 772
 773     def setUp(self):
 774         """ Test setup - force timeout on existing reassemblies """
 775         super(TestIPv4ReassemblyLocalNode, self).setUp()
 776         self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000,
 777                                     expire_walk_interval_ms=10)
 778         self.sleep(.25)
 779         self.vapi.ip_reassembly_set(timeout_ms=1000000, max_reassemblies=1000,
 780                                     expire_walk_interval_ms=10000)
 781
 782     def tearDown(self):
 783         super(TestIPv4ReassemblyLocalNode, self).tearDown()
 784         self.logger.debug(self.vapi.ppcli("show ip4-reassembly details"))
 785
 786     @classmethod
 787     def create_stream(cls, packet_count=test_packet_count):
 788         """Create input packet stream for defined interface.
 789
 790         :param list packet_sizes: Required packet sizes.
 791         """
 792         for i in range(0, packet_count):
 793             info = cls.create_packet_info(cls.src_dst_if, cls.src_dst_if)
 794             payload = cls.info_to_payload(info)
 795             p = (Ether(dst=cls.src_dst_if.local_mac,
 796                        src=cls.src_dst_if.remote_mac) /
 797                  IP(id=info.index, src=cls.src_dst_if.remote_ip4,
 798                     dst=cls.src_dst_if.local_ip4) /
 799                  ICMP(type='echo-request', id=1234) /
 800                  Raw(payload))
 801             cls.extend_packet(p, 1518, cls.padding)
 802             info.data = p
 803
 804     @classmethod
 805     def create_fragments(cls):
 806         infos = cls._packet_infos
 807         cls.pkt_infos = []
 808         for index, info in infos.iteritems():
 809             p = info.data
 810             # cls.logger.debug(ppp("Packet:", p.__class__(str(p))))
 811             fragments_300 = fragment_rfc791(p, 300)
 812             cls.pkt_infos.append((index, fragments_300))
 813         cls.fragments_300 = [x for (_, frags) in cls.pkt_infos for x in frags]
 814         cls.logger.debug("Fragmented %s packets into %s 300-byte fragments" %
 815                          (len(infos), len(cls.fragments_300)))
 816
 817     def verify_capture(self, capture):
 818         """Verify captured packet stream.
 819
 820         :param list capture: Captured packet stream.
 821         """
 822         info = None
 823         seen = set()
 824         for packet in capture:
 825             try:
 826                 self.logger.debug(ppp("Got packet:", packet))
 827                 ip = packet[IP]
 828                 icmp = packet[ICMP]
 829                 payload_info = self.payload_to_info(str(packet[Raw]))
 830                 packet_index = payload_info.index
 831                 if packet_index in seen:
 832                     raise Exception(ppp("Duplicate packet received", packet))
 833                 seen.add(packet_index)
 834                 self.assertEqual(payload_info.dst, self.src_dst_if.sw_if_index)
 835                 info = self._packet_infos[packet_index]
 836                 self.assertTrue(info is not None)
 837                 self.assertEqual(packet_index, info.index)
 838                 saved_packet = info.data
 839                 self.assertEqual(ip.src, saved_packet[IP].dst)
 840                 self.assertEqual(ip.dst, saved_packet[IP].src)
 841                 self.assertEqual(icmp.type, 0)  # echo reply
 842                 self.assertEqual(icmp.id, saved_packet[ICMP].id)
 843                 self.assertEqual(icmp.payload, saved_packet[ICMP].payload)
 844             except Exception:
 845                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
 846                 raise
 847         for index in self._packet_infos:
 848             self.assertTrue(index in seen or index in dropped_packet_indexes,
 849                             "Packet with packet_index %d not received" % index)
 850
 851     def test_reassembly(self):
 852         """ basic reassembly """
 853
 854         self.pg_enable_capture()
 855         self.src_dst_if.add_stream(self.fragments_300)
 856         self.pg_start()
 857
 858         packets = self.src_dst_if.get_capture(len(self.pkt_infos))
 859         self.verify_capture(packets)
 860
 861         # run it all again to verify correctness
 862         self.pg_enable_capture()
 863         self.src_dst_if.add_stream(self.fragments_300)
 864         self.pg_start()
 865
 866         packets = self.src_dst_if.get_capture(len(self.pkt_infos))
 867         self.verify_capture(packets)
 868
 869
 870 class TestFIFReassembly(VppTestCase):
 871     """ Fragments in fragments reassembly """
 872
 873     @classmethod
 874     def setUpClass(cls):
 875         super(TestFIFReassembly, cls).setUpClass()
 876
 877         cls.create_pg_interfaces([0, 1])
 878         cls.src_if = cls.pg0
 879         cls.dst_if = cls.pg1
 880         for i in cls.pg_interfaces:
 881             i.admin_up()
 882             i.config_ip4()
 883             i.resolve_arp()
 884             i.config_ip6()
 885             i.resolve_ndp()
 886
 887         cls.packet_sizes = [64, 512, 1518, 9018]
 888         cls.padding = " abcdefghijklmn"
 889
 890     def setUp(self):
 891         """ Test setup - force timeout on existing reassemblies """
 892         super(TestFIFReassembly, self).setUp()
 893         self.vapi.ip_reassembly_enable_disable(
 894             sw_if_index=self.src_if.sw_if_index, enable_ip4=True,
 895             enable_ip6=True)
 896         self.vapi.ip_reassembly_enable_disable(
 897             sw_if_index=self.dst_if.sw_if_index, enable_ip4=True,
 898             enable_ip6=True)
 899         self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000,
 900                                     expire_walk_interval_ms=10)
 901         self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000,
 902                                     expire_walk_interval_ms=10, is_ip6=1)
 903         self.sleep(.25)
 904         self.vapi.ip_reassembly_set(timeout_ms=1000000, max_reassemblies=1000,
 905                                     expire_walk_interval_ms=10000)
 906         self.vapi.ip_reassembly_set(timeout_ms=1000000, max_reassemblies=1000,
 907                                     expire_walk_interval_ms=10000, is_ip6=1)
 908
 909     def tearDown(self):
 910         self.logger.debug(self.vapi.ppcli("show ip4-reassembly details"))
 911         self.logger.debug(self.vapi.ppcli("show ip6-reassembly details"))
 912         super(TestFIFReassembly, self).tearDown()
 913
 914     def verify_capture(self, capture, ip_class, dropped_packet_indexes=[]):
 915         """Verify captured packet stream.
 916
 917         :param list capture: Captured packet stream.
 918         """
 919         info = None
 920         seen = set()
 921         for packet in capture:
 922             try:
 923                 self.logger.debug(ppp("Got packet:", packet))
 924                 ip = packet[ip_class]
 925                 udp = packet[UDP]
 926                 payload_info = self.payload_to_info(str(packet[Raw]))
 927                 packet_index = payload_info.index
 928                 self.assertTrue(
 929                     packet_index not in dropped_packet_indexes,
 930                     ppp("Packet received, but should be dropped:", packet))
 931                 if packet_index in seen:
 932                     raise Exception(ppp("Duplicate packet received", packet))
 933                 seen.add(packet_index)
 934                 self.assertEqual(payload_info.dst, self.dst_if.sw_if_index)
 935                 info = self._packet_infos[packet_index]
 936                 self.assertTrue(info is not None)
 937                 self.assertEqual(packet_index, info.index)
 938                 saved_packet = info.data
 939                 self.assertEqual(ip.src, saved_packet[ip_class].src)
 940                 self.assertEqual(ip.dst, saved_packet[ip_class].dst)
 941                 self.assertEqual(udp.payload, saved_packet[UDP].payload)
 942             except Exception:
 943                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
 944                 raise
 945         for index in self._packet_infos:
 946             self.assertTrue(index in seen or index in dropped_packet_indexes,
 947                             "Packet with packet_index %d not received" % index)
 948
 949     def test_fif4(self):
 950         """ Fragments in fragments (4o4) """
 951
 952         # TODO this should be ideally in setUpClass, but then we hit a bug
 953         # with VppIpRoute incorrectly reporting it's present when it's not
 954         # so we need to manually remove the vpp config, thus we cannot have
 955         # it shared for multiple test cases
 956         self.tun_ip4 = "1.1.1.2"
 957
 958         self.gre4 = VppGreInterface(self, self.src_if.local_ip4, self.tun_ip4)
 959         self.gre4.add_vpp_config()
 960         self.gre4.admin_up()
 961         self.gre4.config_ip4()
 962
 963         self.vapi.ip_reassembly_enable_disable(
 964             sw_if_index=self.gre4.sw_if_index, enable_ip4=True)
 965
 966         self.route4 = VppIpRoute(self, self.tun_ip4, 32,
 967                                  [VppRoutePath(self.src_if.remote_ip4,
 968                                                self.src_if.sw_if_index)])
 969         self.route4.add_vpp_config()
 970
 971         self.reset_packet_infos()
 972         for i in range(test_packet_count):
 973             info = self.create_packet_info(self.src_if, self.dst_if)
 974             payload = self.info_to_payload(info)
 975             # Ethernet header here is only for size calculation, thus it
 976             # doesn't matter how it's initialized. This is to ensure that
 977             # reassembled packet is not > 9000 bytes, so that it's not dropped
 978             p = (Ether() /
 979                  IP(id=i, src=self.src_if.remote_ip4,
 980                     dst=self.dst_if.remote_ip4) /
 981                  UDP(sport=1234, dport=5678) /
 982                  Raw(payload))
 983             size = self.packet_sizes[(i // 2) % len(self.packet_sizes)]
 984             self.extend_packet(p, size, self.padding)
 985             info.data = p[IP]  # use only IP part, without ethernet header
 986
 987         fragments = [x for _, p in self._packet_infos.iteritems()
 988                      for x in fragment_rfc791(p.data, 400)]
 989
 990         encapped_fragments = \
 991             [Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac) /
 992              IP(src=self.tun_ip4, dst=self.src_if.local_ip4) /
 993                 GRE() /
 994                 p
 995                 for p in fragments]
 996
 997         fragmented_encapped_fragments = \
 998             [x for p in encapped_fragments
 999              for x in fragment_rfc791(p, 200)]
1000
1001         self.src_if.add_stream(fragmented_encapped_fragments)
1002
1003         self.pg_enable_capture(self.pg_interfaces)
1004         self.pg_start()
1005
1006         self.src_if.assert_nothing_captured()
1007         packets = self.dst_if.get_capture(len(self._packet_infos))
1008         self.verify_capture(packets, IP)
1009
1010         # TODO remove gre vpp config by hand until VppIpRoute gets fixed
1011         # so that it's query_vpp_config() works as it should
1012         self.gre4.remove_vpp_config()
1013         self.logger.debug(self.vapi.ppcli("show interface"))
1014
1015     def test_fif6(self):
1016         """ Fragments in fragments (6o6) """
1017         # TODO this should be ideally in setUpClass, but then we hit a bug
1018         # with VppIpRoute incorrectly reporting it's present when it's not
1019         # so we need to manually remove the vpp config, thus we cannot have
1020         # it shared for multiple test cases
1021         self.tun_ip6 = "1002::1"
1022
1023         self.gre6 = VppGre6Interface(self, self.src_if.local_ip6, self.tun_ip6)
1024         self.gre6.add_vpp_config()
1025         self.gre6.admin_up()
1026         self.gre6.config_ip6()
1027
1028         self.vapi.ip_reassembly_enable_disable(
1029             sw_if_index=self.gre6.sw_if_index, enable_ip6=True)
1030
1031         self.route6 = VppIpRoute(self, self.tun_ip6, 128,
1032                                  [VppRoutePath(self.src_if.remote_ip6,
1033                                                self.src_if.sw_if_index,
1034                                                proto=DpoProto.DPO_PROTO_IP6)],
1035                                  is_ip6=1)
1036         self.route6.add_vpp_config()
1037
1038         self.reset_packet_infos()
1039         for i in range(test_packet_count):
1040             info = self.create_packet_info(self.src_if, self.dst_if)
1041             payload = self.info_to_payload(info)
1042             # Ethernet header here is only for size calculation, thus it
1043             # doesn't matter how it's initialized. This is to ensure that
1044             # reassembled packet is not > 9000 bytes, so that it's not dropped
1045             p = (Ether() /
1046                  IPv6(src=self.src_if.remote_ip6, dst=self.dst_if.remote_ip6) /
1047                  UDP(sport=1234, dport=5678) /
1048                  Raw(payload))
1049             size = self.packet_sizes[(i // 2) % len(self.packet_sizes)]
1050             self.extend_packet(p, size, self.padding)
1051             info.data = p[IPv6]  # use only IPv6 part, without ethernet header
1052
1053         fragments = [x for _, i in self._packet_infos.iteritems()
1054                      for x in fragment_rfc8200(
1055                          i.data, i.index, 400)]
1056
1057         encapped_fragments = \
1058             [Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac) /
1059              IPv6(src=self.tun_ip6, dst=self.src_if.local_ip6) /
1060                 GRE() /
1061                 p
1062                 for p in fragments]
1063
1064         fragmented_encapped_fragments = \
1065             [x for p in encapped_fragments for x in (
1066                 fragment_rfc8200(
1067                     p,
1068                     2 * len(self._packet_infos) + p[IPv6ExtHdrFragment].id,
1069                     200)
1070                 if IPv6ExtHdrFragment in p else [p]
1071             )
1072             ]
1073
1074         self.src_if.add_stream(fragmented_encapped_fragments)
1075
1076         self.pg_enable_capture(self.pg_interfaces)
1077         self.pg_start()
1078
1079         self.src_if.assert_nothing_captured()
1080         packets = self.dst_if.get_capture(len(self._packet_infos))
1081         self.verify_capture(packets, IPv6)
1082
1083         # TODO remove gre vpp config by hand until VppIpRoute gets fixed
1084         # so that it's query_vpp_config() works as it should
1085         self.gre6.remove_vpp_config()
1086
1087
1088 if __name__ == '__main__':
1089     unittest.main(testRunner=VppTestRunner)