ip: rate-limit the sending of ICMP error messages
[vpp.git] / test / test_reassembly.py
index 124fec0..4c7a7cd 100644 (file)
@@ -8,9 +8,11 @@ from framework import VppTestCase, VppTestRunner
 import scapy.compat
 from scapy.packet import Raw
 from scapy.layers.l2 import Ether, GRE
-from scapy.layers.inet import IP, UDP, ICMP
+from scapy.layers.inet import IP, UDP, ICMP, icmptypes
 from scapy.layers.inet6 import HBHOptUnknown, ICMPv6ParamProblem,\
-    ICMPv6TimeExceeded, IPv6, IPv6ExtHdrFragment, IPv6ExtHdrHopByHop
+    ICMPv6TimeExceeded, IPv6, IPv6ExtHdrFragment,\
+    IPv6ExtHdrHopByHop, IPv6ExtHdrDestOpt, PadN, ICMPv6EchoRequest,\
+    ICMPv6EchoReply
 from framework import VppTestCase, VppTestRunner
 from util import ppp, ppc, fragment_rfc791, fragment_rfc8200
 from vpp_gre_interface import VppGreInterface
@@ -27,7 +29,7 @@ class TestIPv4Reassembly(VppTestCase):
 
     @classmethod
     def setUpClass(cls):
-        super(TestIPv4Reassembly, cls).setUpClass()
+        super().setUpClass()
 
         cls.create_pg_interfaces([0, 1])
         cls.src_if = cls.pg0
@@ -47,23 +49,25 @@ class TestIPv4Reassembly(VppTestCase):
 
     @classmethod
     def tearDownClass(cls):
-        super(TestIPv4Reassembly, cls).tearDownClass()
+        super().tearDownClass()
 
     def setUp(self):
         """ Test setup - force timeout on existing reassemblies """
-        super(TestIPv4Reassembly, self).setUp()
+        super().setUp()
         self.vapi.ip_reassembly_enable_disable(
             sw_if_index=self.src_if.sw_if_index, enable_ip4=True)
         self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000,
                                     max_reassembly_length=1000,
                                     expire_walk_interval_ms=10)
-        self.sleep(.25)
+        self.virtual_sleep(.25)
         self.vapi.ip_reassembly_set(timeout_ms=1000000, max_reassemblies=1000,
                                     max_reassembly_length=1000,
                                     expire_walk_interval_ms=10000)
 
     def tearDown(self):
-        super(TestIPv4Reassembly, self).tearDown()
+        self.vapi.ip_reassembly_enable_disable(
+            sw_if_index=self.src_if.sw_if_index, enable_ip4=False)
+        super().tearDown()
 
     def show_commands_at_teardown(self):
         self.logger.debug(self.vapi.ppcli("show ip4-full-reassembly details"))
@@ -480,7 +484,7 @@ Ethernet-Payload.IPv4-Packet.IPv4-Header.Fragment-Offset; Test-case: 5737'''
         self.src_if.add_stream(fragments)
         self.pg_start()
 
-        self.sleep(.25, "wait before sending rest of fragments")
+        self.virtual_sleep(.25, "wait before sending rest of fragments")
 
         self.src_if.add_stream(fragments2)
         self.pg_start()
@@ -510,13 +514,31 @@ Ethernet-Payload.IPv4-Packet.IPv4-Header.Fragment-Offset; Test-case: 5737'''
         self.verify_capture(packets, dropped_packet_indexes)
         self.src_if.assert_nothing_captured()
 
+    def test_local_enable_disable(self):
+        """ local reassembly enabled/disable """
+        self.vapi.ip_reassembly_enable_disable(
+            sw_if_index=self.src_if.sw_if_index, enable_ip4=False)
+        self.vapi.ip_local_reass_enable_disable(enable_ip4=True)
+        p = (Ether(src=self.src_if.remote_mac, dst=self.src_if.local_mac) /
+             IP(src=self.src_if.remote_ip4, dst=self.src_if.local_ip4) /
+             ICMP(id=1234, type='echo-request') /
+             Raw('x' * 1000))
+        frags = fragment_rfc791(p, 400)
+        r = self.send_and_expect(self.src_if, frags, self.src_if, n_rx=1)[0]
+        self.assertEqual(1234, r[ICMP].id)
+        self.assertEqual(icmptypes[r[ICMP].type], 'echo-reply')
+        self.vapi.ip_local_reass_enable_disable()
+
+        self.send_and_assert_no_replies(self.src_if, frags)
+        self.vapi.ip_local_reass_enable_disable(enable_ip4=True)
+
 
 class TestIPv4SVReassembly(VppTestCase):
     """ IPv4 Shallow Virtual Reassembly """
 
     @classmethod
     def setUpClass(cls):
-        super(TestIPv4SVReassembly, cls).setUpClass()
+        super().setUpClass()
 
         cls.create_pg_interfaces([0, 1])
         cls.src_if = cls.pg0
@@ -530,7 +552,7 @@ class TestIPv4SVReassembly(VppTestCase):
 
     def setUp(self):
         """ Test setup - force timeout on existing reassemblies """
-        super(TestIPv4SVReassembly, self).setUp()
+        super().setUp()
         self.vapi.ip_reassembly_enable_disable(
             sw_if_index=self.src_if.sw_if_index, enable_ip4=True,
             type=VppEnum.vl_api_ip_reass_type_t.IP_REASS_TYPE_SHALLOW_VIRTUAL)
@@ -539,7 +561,7 @@ class TestIPv4SVReassembly(VppTestCase):
             max_reassembly_length=1000,
             type=VppEnum.vl_api_ip_reass_type_t.IP_REASS_TYPE_SHALLOW_VIRTUAL,
             expire_walk_interval_ms=10)
-        self.sleep(.25)
+        self.virtual_sleep(.25)
         self.vapi.ip_reassembly_set(
             timeout_ms=1000000, max_reassemblies=1000,
             max_reassembly_length=1000,
@@ -547,7 +569,7 @@ class TestIPv4SVReassembly(VppTestCase):
             expire_walk_interval_ms=10000)
 
     def tearDown(self):
-        super(TestIPv4SVReassembly, self).tearDown()
+        super().tearDown()
         self.logger.debug(self.vapi.ppcli("show ip4-sv-reassembly details"))
         self.logger.debug(self.vapi.ppcli("show buffers"))
 
@@ -672,7 +694,7 @@ class TestIPv4SVReassembly(VppTestCase):
             self.assertEqual(sent[Raw].payload, recvd[Raw].payload)
 
         # wait for cleanup
-        self.sleep(.25, "wait before sending rest of fragments")
+        self.virtual_sleep(.25, "wait before sending rest of fragments")
 
         # send rest of fragments - shouldn't be forwarded
         self.pg_enable_capture()
@@ -790,7 +812,7 @@ class TestIPv4MWReassembly(VppTestCase):
 
     @classmethod
     def setUpClass(cls):
-        super(TestIPv4MWReassembly, cls).setUpClass()
+        super().setUpClass()
 
         cls.create_pg_interfaces(range(cls.vpp_worker_count+1))
         cls.src_if = cls.pg0
@@ -814,24 +836,27 @@ class TestIPv4MWReassembly(VppTestCase):
 
     @classmethod
     def tearDownClass(cls):
-        super(TestIPv4MWReassembly, cls).tearDownClass()
+        super().tearDownClass()
 
     def setUp(self):
         """ Test setup - force timeout on existing reassemblies """
-        super(TestIPv4MWReassembly, self).setUp()
+        super().setUp()
         for intf in self.send_ifs:
             self.vapi.ip_reassembly_enable_disable(
                 sw_if_index=intf.sw_if_index, enable_ip4=True)
         self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000,
                                     max_reassembly_length=1000,
                                     expire_walk_interval_ms=10)
-        self.sleep(.25)
+        self.virtual_sleep(.25)
         self.vapi.ip_reassembly_set(timeout_ms=1000000, max_reassemblies=1000,
                                     max_reassembly_length=1000,
                                     expire_walk_interval_ms=10000)
 
     def tearDown(self):
-        super(TestIPv4MWReassembly, self).tearDown()
+        for intf in self.send_ifs:
+            self.vapi.ip_reassembly_enable_disable(
+                sw_if_index=intf.sw_if_index, enable_ip4=False)
+        super().tearDown()
 
     def show_commands_at_teardown(self):
         self.logger.debug(self.vapi.ppcli("show ip4-full-reassembly details"))
@@ -967,7 +992,7 @@ class TestIPv6Reassembly(VppTestCase):
 
     @classmethod
     def setUpClass(cls):
-        super(TestIPv6Reassembly, cls).setUpClass()
+        super().setUpClass()
 
         cls.create_pg_interfaces([0, 1])
         cls.src_if = cls.pg0
@@ -987,17 +1012,17 @@ class TestIPv6Reassembly(VppTestCase):
 
     @classmethod
     def tearDownClass(cls):
-        super(TestIPv6Reassembly, cls).tearDownClass()
+        super().tearDownClass()
 
     def setUp(self):
         """ Test setup - force timeout on existing reassemblies """
-        super(TestIPv6Reassembly, self).setUp()
+        super().setUp()
         self.vapi.ip_reassembly_enable_disable(
             sw_if_index=self.src_if.sw_if_index, enable_ip6=True)
         self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000,
                                     max_reassembly_length=1000,
                                     expire_walk_interval_ms=10, is_ip6=1)
-        self.sleep(.25)
+        self.virtual_sleep(.25)
         self.vapi.ip_reassembly_set(timeout_ms=1000000, max_reassemblies=1000,
                                     max_reassembly_length=1000,
                                     expire_walk_interval_ms=10000, is_ip6=1)
@@ -1005,7 +1030,9 @@ class TestIPv6Reassembly(VppTestCase):
         self.logger.debug(self.vapi.ppcli("show buffers"))
 
     def tearDown(self):
-        super(TestIPv6Reassembly, self).tearDown()
+        self.vapi.ip_reassembly_enable_disable(
+            sw_if_index=self.src_if.sw_if_index, enable_ip6=False)
+        super().tearDown()
 
     def show_commands_at_teardown(self):
         self.logger.debug(self.vapi.ppcli("show ip6-full-reassembly details"))
@@ -1299,8 +1326,7 @@ class TestIPv6Reassembly(VppTestCase):
         packets = self.dst_if.get_capture(
             len(self.pkt_infos) - len(dropped_packet_indexes))
         self.verify_capture(packets, dropped_packet_indexes)
-        pkts = self.src_if.get_capture(
-            expected_count=len(dropped_packet_indexes))
+        pkts = self.src_if._get_capture(1)
         for icmp in pkts:
             self.assertIn(ICMPv6TimeExceeded, icmp)
             self.assertIn(IPv6ExtHdrFragment, icmp)
@@ -1337,7 +1363,7 @@ class TestIPv6Reassembly(VppTestCase):
         self.src_if.add_stream(fragments)
         self.pg_start()
 
-        self.sleep(.25, "wait before sending rest of fragments")
+        self.virtual_sleep(.25, "wait before sending rest of fragments")
 
         self.src_if.add_stream(fragments2)
         self.pg_start()
@@ -1345,8 +1371,7 @@ class TestIPv6Reassembly(VppTestCase):
         packets = self.dst_if.get_capture(
             len(self.pkt_infos) - len(dropped_packet_indexes))
         self.verify_capture(packets, dropped_packet_indexes)
-        pkts = self.src_if.get_capture(
-            expected_count=len(dropped_packet_indexes))
+        pkts = self.src_if._get_capture(1)
         for icmp in pkts:
             self.assertIn(ICMPv6TimeExceeded, icmp)
             self.assertIn(IPv6ExtHdrFragment, icmp)
@@ -1375,24 +1400,31 @@ class TestIPv6Reassembly(VppTestCase):
 
     def test_missing_upper(self):
         """ missing upper layer """
+        optdata = '\x00' * 100
         p = (Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac) /
              IPv6(src=self.src_if.remote_ip6,
                   dst=self.src_if.local_ip6) /
-             UDP(sport=1234, dport=5678) /
-             Raw())
-        self.extend_packet(p, 1000, self.padding)
-        fragments = fragment_rfc8200(p, 1, 500)
-        bad_fragment = p.__class__(scapy.compat.raw(fragments[1]))
-        bad_fragment[IPv6ExtHdrFragment].nh = 59
-        bad_fragment[IPv6ExtHdrFragment].offset = 0
+             IPv6ExtHdrFragment(m=1) /
+             IPv6ExtHdrDestOpt(nh=17, options=PadN(optdata='\101' * 255) /
+             PadN(optdata='\102'*255)))
+
         self.pg_enable_capture()
-        self.src_if.add_stream([bad_fragment])
+        self.src_if.add_stream([p])
         self.pg_start()
         pkts = self.src_if.get_capture(expected_count=1)
         icmp = pkts[0]
         self.assertIn(ICMPv6ParamProblem, icmp)
         self.assert_equal(icmp[ICMPv6ParamProblem].code, 3, "ICMP code")
 
+    def test_truncated_fragment(self):
+        """ truncated fragment """
+        pkt = (Ether(src=self.pg0.local_mac, dst=self.pg0.remote_mac) /
+               IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6,
+                    nh=44, plen=2) /
+               IPv6ExtHdrFragment(nh=6))
+
+        self.send_and_assert_no_replies(self.pg0, [pkt], self.pg0)
+
     def test_invalid_frag_size(self):
         """ fragment size not a multiple of 8 """
         p = (Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac) /
@@ -1431,6 +1463,88 @@ class TestIPv6Reassembly(VppTestCase):
         self.assertIn(ICMPv6ParamProblem, icmp)
         self.assert_equal(icmp[ICMPv6ParamProblem].code, 0, "ICMP code")
 
+    def test_atomic_fragment(self):
+        """ IPv6 atomic fragment """
+        pkt = (Ether(src=self.pg0.local_mac, dst=self.pg0.remote_mac) /
+               IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6,
+                    nh=44, plen=65535) /
+               IPv6ExtHdrFragment(offset=8191, m=1, res1=0xFF, res2=0xFF,
+                                  nh=255, id=0xffff)/('X'*1452))
+
+        rx = self.send_and_expect(self.pg0, [pkt], self.pg0)
+        self.assertIn(ICMPv6ParamProblem, rx[0])
+
+    def test_truncated_fragment(self):
+        """ IPv6 truncated fragment header """
+        pkt = (Ether(src=self.pg0.local_mac, dst=self.pg0.remote_mac) /
+               IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6,
+                    nh=44, plen=2) /
+               IPv6ExtHdrFragment(nh=6))
+
+        self.send_and_assert_no_replies(self.pg0, [pkt])
+
+        pkt = (Ether(src=self.pg0.local_mac, dst=self.pg0.remote_mac) /
+               IPv6(src=self.pg0.remote_ip6, dst=self.pg0.remote_ip6) /
+               ICMPv6EchoRequest())
+        rx = self.send_and_expect(self.pg0, [pkt], self.pg0)
+
+    def test_one_fragment(self):
+        """ whole packet in one fragment processed independently """
+        pkt = (Ether(src=self.pg0.local_mac, dst=self.pg0.remote_mac) /
+               IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6) /
+               ICMPv6EchoRequest()/Raw('X' * 1600))
+        frags = fragment_rfc8200(pkt, 1, 400)
+
+        # send a fragment with known id
+        self.send_and_assert_no_replies(self.pg0, [frags[0]])
+
+        # send an atomic fragment with same id - should be reassembled
+        pkt = (Ether(src=self.pg0.local_mac, dst=self.pg0.remote_mac) /
+               IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6) /
+               IPv6ExtHdrFragment(id=1)/ICMPv6EchoRequest())
+        rx = self.send_and_expect(self.pg0, [pkt], self.pg0)
+        self.assertNotIn(IPv6ExtHdrFragment, rx)
+
+        # now finish the original reassembly, this should still be possible
+        rx = self.send_and_expect(self.pg0, frags[1:], self.pg0, n_rx=1)
+        self.assertNotIn(IPv6ExtHdrFragment, rx)
+
+    def test_bunch_of_fragments(self):
+        """ valid fragments followed by rogue fragments and atomic fragment"""
+        pkt = (Ether(src=self.pg0.local_mac, dst=self.pg0.remote_mac) /
+               IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6) /
+               ICMPv6EchoRequest()/Raw('X' * 1600))
+        frags = fragment_rfc8200(pkt, 1, 400)
+        self.send_and_expect(self.pg0, frags, self.pg0, n_rx=1)
+
+        inc_frag = (Ether(src=self.pg0.local_mac, dst=self.pg0.remote_mac) /
+                    IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6) /
+                    IPv6ExtHdrFragment(id=1, nh=58, offset=608)/Raw('X'*308))
+
+        self.send_and_assert_no_replies(self.pg0, inc_frag*604)
+
+        pkt = (Ether(src=self.pg0.local_mac, dst=self.pg0.remote_mac) /
+               IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6) /
+               IPv6ExtHdrFragment(id=1)/ICMPv6EchoRequest())
+        rx = self.send_and_expect(self.pg0, [pkt], self.pg0)
+        self.assertNotIn(IPv6ExtHdrFragment, rx)
+
+    def test_local_enable_disable(self):
+        """ local reassembly enabled/disable """
+        self.vapi.ip_reassembly_enable_disable(
+            sw_if_index=self.src_if.sw_if_index, enable_ip6=False)
+        self.vapi.ip_local_reass_enable_disable(enable_ip6=True)
+        pkt = (Ether(src=self.src_if.local_mac, dst=self.src_if.remote_mac) /
+               IPv6(src=self.src_if.remote_ip6, dst=self.src_if.local_ip6) /
+               ICMPv6EchoRequest(id=1234)/Raw('X' * 1600))
+        frags = fragment_rfc8200(pkt, 1, 400)
+        r = self.send_and_expect(self.src_if, frags, self.src_if, n_rx=1)[0]
+        self.assertEqual(1234, r[ICMPv6EchoReply].id)
+        self.vapi.ip_local_reass_enable_disable()
+
+        self.send_and_assert_no_replies(self.src_if, frags)
+        self.vapi.ip_local_reass_enable_disable(enable_ip6=True)
+
 
 class TestIPv6MWReassembly(VppTestCase):
     """ IPv6 Reassembly (multiple workers) """
@@ -1438,7 +1552,7 @@ class TestIPv6MWReassembly(VppTestCase):
 
     @classmethod
     def setUpClass(cls):
-        super(TestIPv6MWReassembly, cls).setUpClass()
+        super().setUpClass()
 
         cls.create_pg_interfaces(range(cls.vpp_worker_count+1))
         cls.src_if = cls.pg0
@@ -1462,24 +1576,27 @@ class TestIPv6MWReassembly(VppTestCase):
 
     @classmethod
     def tearDownClass(cls):
-        super(TestIPv6MWReassembly, cls).tearDownClass()
+        super().tearDownClass()
 
     def setUp(self):
         """ Test setup - force timeout on existing reassemblies """
-        super(TestIPv6MWReassembly, self).setUp()
+        super().setUp()
         for intf in self.send_ifs:
             self.vapi.ip_reassembly_enable_disable(
                 sw_if_index=intf.sw_if_index, enable_ip6=True)
         self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000,
                                     max_reassembly_length=1000,
                                     expire_walk_interval_ms=10, is_ip6=1)
-        self.sleep(.25)
+        self.virtual_sleep(.25)
         self.vapi.ip_reassembly_set(timeout_ms=1000000, max_reassemblies=1000,
                                     max_reassembly_length=1000,
                                     expire_walk_interval_ms=1000, is_ip6=1)
 
     def tearDown(self):
-        super(TestIPv6MWReassembly, self).tearDown()
+        for intf in self.send_ifs:
+            self.vapi.ip_reassembly_enable_disable(
+                sw_if_index=intf.sw_if_index, enable_ip6=False)
+        super().tearDown()
 
     def show_commands_at_teardown(self):
         self.logger.debug(self.vapi.ppcli("show ip6-full-reassembly details"))
@@ -1615,7 +1732,7 @@ class TestIPv6SVReassembly(VppTestCase):
 
     @classmethod
     def setUpClass(cls):
-        super(TestIPv6SVReassembly, cls).setUpClass()
+        super().setUpClass()
 
         cls.create_pg_interfaces([0, 1])
         cls.src_if = cls.pg0
@@ -1629,7 +1746,7 @@ class TestIPv6SVReassembly(VppTestCase):
 
     def setUp(self):
         """ Test setup - force timeout on existing reassemblies """
-        super(TestIPv6SVReassembly, self).setUp()
+        super().setUp()
         self.vapi.ip_reassembly_enable_disable(
             sw_if_index=self.src_if.sw_if_index, enable_ip6=True,
             type=VppEnum.vl_api_ip_reass_type_t.IP_REASS_TYPE_SHALLOW_VIRTUAL)
@@ -1638,7 +1755,7 @@ class TestIPv6SVReassembly(VppTestCase):
             max_reassembly_length=1000,
             type=VppEnum.vl_api_ip_reass_type_t.IP_REASS_TYPE_SHALLOW_VIRTUAL,
             expire_walk_interval_ms=10, is_ip6=1)
-        self.sleep(.25)
+        self.virtual_sleep(.25)
         self.vapi.ip_reassembly_set(
             timeout_ms=1000000, max_reassemblies=1000,
             max_reassembly_length=1000,
@@ -1646,7 +1763,7 @@ class TestIPv6SVReassembly(VppTestCase):
             expire_walk_interval_ms=10000, is_ip6=1)
 
     def tearDown(self):
-        super(TestIPv6SVReassembly, self).tearDown()
+        super().tearDown()
         self.logger.debug(self.vapi.ppcli("show ip6-sv-reassembly details"))
         self.logger.debug(self.vapi.ppcli("show buffers"))
 
@@ -1769,7 +1886,7 @@ class TestIPv6SVReassembly(VppTestCase):
             self.assertEqual(sent[Raw].payload, recvd[Raw].payload)
 
         # wait for cleanup
-        self.sleep(.25, "wait before sending rest of fragments")
+        self.virtual_sleep(.25, "wait before sending rest of fragments")
 
         # send rest of fragments - shouldn't be forwarded
         self.pg_enable_capture()
@@ -1814,13 +1931,60 @@ class TestIPv6SVReassembly(VppTestCase):
             self.assertEqual(sent[IPv6].dst, recvd[IPv6].dst)
             self.assertEqual(sent[Raw].payload, recvd[Raw].payload)
 
+    def test_one_fragment(self):
+        """ whole packet in one fragment processed independently """
+        pkt = (Ether(src=self.src_if.local_mac, dst=self.src_if.remote_mac) /
+               IPv6(src=self.src_if.remote_ip6, dst=self.dst_if.remote_ip6) /
+               ICMPv6EchoRequest()/Raw('X' * 1600))
+        frags = fragment_rfc8200(pkt, 1, 400)
+
+        # send a fragment with known id
+        self.send_and_expect(self.src_if, [frags[0]], self.dst_if)
+
+        # send an atomic fragment with same id - should be reassembled
+        pkt = (Ether(src=self.src_if.local_mac, dst=self.src_if.remote_mac) /
+               IPv6(src=self.src_if.remote_ip6, dst=self.dst_if.remote_ip6) /
+               IPv6ExtHdrFragment(id=1)/ICMPv6EchoRequest())
+        rx = self.send_and_expect(self.src_if, [pkt], self.dst_if)
+
+        # now forward packets matching original reassembly, should still work
+        rx = self.send_and_expect(self.src_if, frags[1:], self.dst_if)
+
+    def test_bunch_of_fragments(self):
+        """ valid fragments followed by rogue fragments and atomic fragment"""
+        pkt = (Ether(src=self.src_if.local_mac, dst=self.src_if.remote_mac) /
+               IPv6(src=self.src_if.remote_ip6, dst=self.dst_if.remote_ip6) /
+               ICMPv6EchoRequest()/Raw('X' * 1600))
+        frags = fragment_rfc8200(pkt, 1, 400)
+        rx = self.send_and_expect(self.src_if, frags, self.dst_if)
+
+        rogue = (Ether(src=self.src_if.local_mac, dst=self.src_if.remote_mac) /
+                 IPv6(src=self.src_if.remote_ip6, dst=self.dst_if.remote_ip6) /
+                 IPv6ExtHdrFragment(id=1, nh=58, offset=608)/Raw('X'*308))
+
+        self.send_and_expect(self.src_if, rogue*604, self.dst_if)
+
+        pkt = (Ether(src=self.src_if.local_mac, dst=self.src_if.remote_mac) /
+               IPv6(src=self.src_if.remote_ip6, dst=self.dst_if.remote_ip6) /
+               IPv6ExtHdrFragment(id=1)/ICMPv6EchoRequest())
+        rx = self.send_and_expect(self.src_if, [pkt], self.dst_if)
+
+    def test_truncated_fragment(self):
+        """ truncated fragment """
+        pkt = (Ether(src=self.pg0.local_mac, dst=self.pg0.remote_mac) /
+               IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6,
+                    nh=44, plen=2) /
+               IPv6ExtHdrFragment(nh=6))
+
+        self.send_and_assert_no_replies(self.pg0, [pkt], self.pg0)
+
 
 class TestIPv4ReassemblyLocalNode(VppTestCase):
     """ IPv4 Reassembly for packets coming to ip4-local node """
 
     @classmethod
     def setUpClass(cls):
-        super(TestIPv4ReassemblyLocalNode, cls).setUpClass()
+        super().setUpClass()
 
         cls.create_pg_interfaces([0])
         cls.src_dst_if = cls.pg0
@@ -1837,21 +2001,21 @@ class TestIPv4ReassemblyLocalNode(VppTestCase):
 
     @classmethod
     def tearDownClass(cls):
-        super(TestIPv4ReassemblyLocalNode, cls).tearDownClass()
+        super().tearDownClass()
 
     def setUp(self):
         """ Test setup - force timeout on existing reassemblies """
-        super(TestIPv4ReassemblyLocalNode, self).setUp()
+        super().setUp()
         self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000,
                                     max_reassembly_length=1000,
                                     expire_walk_interval_ms=10)
-        self.sleep(.25)
+        self.virtual_sleep(.25)
         self.vapi.ip_reassembly_set(timeout_ms=1000000, max_reassemblies=1000,
                                     max_reassembly_length=1000,
                                     expire_walk_interval_ms=10000)
 
     def tearDown(self):
-        super(TestIPv4ReassemblyLocalNode, self).tearDown()
+        super().tearDown()
 
     def show_commands_at_teardown(self):
         self.logger.debug(self.vapi.ppcli("show ip4-full-reassembly details"))
@@ -1947,7 +2111,7 @@ class TestFIFReassembly(VppTestCase):
 
     @classmethod
     def setUpClass(cls):
-        super(TestFIFReassembly, cls).setUpClass()
+        super().setUpClass()
 
         cls.create_pg_interfaces([0, 1])
         cls.src_if = cls.pg0
@@ -1964,11 +2128,11 @@ class TestFIFReassembly(VppTestCase):
 
     @classmethod
     def tearDownClass(cls):
-        super(TestFIFReassembly, cls).tearDownClass()
+        super().tearDownClass()
 
     def setUp(self):
         """ Test setup - force timeout on existing reassemblies """
-        super(TestFIFReassembly, self).setUp()
+        super().setUp()
         self.vapi.ip_reassembly_enable_disable(
             sw_if_index=self.src_if.sw_if_index, enable_ip4=True,
             enable_ip6=True)
@@ -1981,7 +2145,7 @@ class TestFIFReassembly(VppTestCase):
         self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000,
                                     max_reassembly_length=1000,
                                     expire_walk_interval_ms=10, is_ip6=1)
-        self.sleep(.25)
+        self.virtual_sleep(.25)
         self.vapi.ip_reassembly_set(timeout_ms=1000000, max_reassemblies=1000,
                                     max_reassembly_length=1000,
                                     expire_walk_interval_ms=10000)
@@ -1990,7 +2154,7 @@ class TestFIFReassembly(VppTestCase):
                                     expire_walk_interval_ms=10000, is_ip6=1)
 
     def tearDown(self):
-        super(TestFIFReassembly, self).tearDown()
+        super().tearDown()
 
     def show_commands_at_teardown(self):
         self.logger.debug(self.vapi.ppcli("show ip4-full-reassembly details"))