#!/usr/bin/env python3
-import six
import unittest
from random import shuffle, choice, randrange
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
# 35 is enough to have >257 400-byte fragments
test_packet_count = 35
-# number of workers used for multi-worker test cases
-worker_count = 3
-
class TestIPv4Reassembly(VppTestCase):
""" IPv4 Reassembly """
@classmethod
def setUpClass(cls):
- super(TestIPv4Reassembly, cls).setUpClass()
+ super().setUpClass()
cls.create_pg_interfaces([0, 1])
cls.src_if = cls.pg0
@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"))
def create_fragments(cls):
infos = cls._packet_infos
cls.pkt_infos = []
- for index, info in six.iteritems(infos):
+ for index, info in infos.items():
p = info.data
# cls.logger.debug(ppp("Packet:",
# p.__class__(scapy.compat.raw(p))))
self.verify_capture(packets)
self.src_if.assert_nothing_captured()
+ def test_verify_clear_trace_mid_reassembly(self):
+ """ verify clear trace works mid-reassembly """
+
+ self.pg_enable_capture()
+ self.src_if.add_stream(self.fragments_200[0:-1])
+ self.pg_start()
+
+ self.logger.debug(self.vapi.cli("show trace"))
+ self.vapi.cli("clear trace")
+
+ self.src_if.add_stream(self.fragments_200[-1])
+ self.pg_start()
+ packets = self.dst_if.get_capture(len(self.pkt_infos))
+ self.verify_capture(packets)
+
def test_reversed(self):
""" reverse order reassembly """
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()
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
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)
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,
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"))
self.assertEqual(sent[IP].dst, recvd[IP].dst)
self.assertEqual(sent[Raw].payload, recvd[Raw].payload)
+ def test_verify_clear_trace_mid_reassembly(self):
+ """ verify clear trace works mid-reassembly """
+ payload_len = 1000
+ payload = ""
+ counter = 0
+ while len(payload) < payload_len:
+ payload += "%u " % counter
+ counter += 1
+
+ p = (Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac) /
+ IP(id=1, src=self.src_if.remote_ip4,
+ dst=self.dst_if.remote_ip4) /
+ UDP(sport=1234, dport=5678) /
+ Raw(payload))
+ fragments = fragment_rfc791(p, payload_len/4)
+
+ self.pg_enable_capture()
+ self.src_if.add_stream(fragments[1])
+ self.pg_start()
+
+ self.logger.debug(self.vapi.cli("show trace"))
+ self.vapi.cli("clear trace")
+
+ self.pg_enable_capture()
+ self.src_if.add_stream(fragments[0])
+ self.pg_start()
+ self.dst_if.get_capture(2)
+
+ self.logger.debug(self.vapi.cli("show trace"))
+ self.vapi.cli("clear trace")
+
+ self.pg_enable_capture()
+ self.src_if.add_stream(fragments[2:])
+ self.pg_start()
+ self.dst_if.get_capture(len(fragments[2:]))
+
def test_timeout(self):
""" reassembly timeout """
payload_len = 1000
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()
class TestIPv4MWReassembly(VppTestCase):
""" IPv4 Reassembly (multiple workers) """
- worker_config = "workers %d" % worker_count
+ vpp_worker_count = 3
@classmethod
def setUpClass(cls):
- super(TestIPv4MWReassembly, cls).setUpClass()
+ super().setUpClass()
- cls.create_pg_interfaces(range(worker_count+1))
+ cls.create_pg_interfaces(range(cls.vpp_worker_count+1))
cls.src_if = cls.pg0
cls.send_ifs = cls.pg_interfaces[:-1]
cls.dst_if = cls.pg_interfaces[-1]
@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"))
def create_fragments(cls):
infos = cls._packet_infos
cls.pkt_infos = []
- for index, info in six.iteritems(infos):
+ for index, info in infos.items():
p = info.data
# cls.logger.debug(ppp("Packet:",
# p.__class__(scapy.compat.raw(p))))
"Packet with packet_index %d not received" % index)
def send_packets(self, packets):
- for counter in range(worker_count):
+ for counter in range(self.vpp_worker_count):
if 0 == len(packets[counter]):
continue
send_if = self.send_ifs[counter]
# in first wave we send fragments which don't start at offset 0
# then we send fragments with offset 0 on a different thread
# then the rest of packets on a random thread
- first_packets = [[] for n in range(worker_count)]
- second_packets = [[] for n in range(worker_count)]
- rest_of_packets = [[] for n in range(worker_count)]
+ first_packets = [[] for n in range(self.vpp_worker_count)]
+ second_packets = [[] for n in range(self.vpp_worker_count)]
+ rest_of_packets = [[] for n in range(self.vpp_worker_count)]
for (_, p) in self.pkt_infos:
- wi = randrange(worker_count)
+ wi = randrange(self.vpp_worker_count)
second_packets[wi].append(p[0])
if len(p) <= 1:
continue
wi2 = wi
while wi2 == wi:
- wi2 = randrange(worker_count)
+ wi2 = randrange(self.vpp_worker_count)
first_packets[wi2].append(p[1])
- wi3 = randrange(worker_count)
+ wi3 = randrange(self.vpp_worker_count)
rest_of_packets[wi3].extend(p[2:])
self.pg_enable_capture()
@classmethod
def setUpClass(cls):
- super(TestIPv6Reassembly, cls).setUpClass()
+ super().setUpClass()
cls.create_pg_interfaces([0, 1])
cls.src_if = cls.pg0
@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)
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"))
def create_fragments(cls):
infos = cls._packet_infos
cls.pkt_infos = []
- for index, info in six.iteritems(infos):
+ for index, info in infos.items():
p = info.data
# cls.logger.debug(ppp("Packet:",
# p.__class__(scapy.compat.raw(p))))
self.src_if.assert_nothing_captured()
self.dst_if.assert_nothing_captured()
+ def test_verify_clear_trace_mid_reassembly(self):
+ """ verify clear trace works mid-reassembly """
+
+ self.pg_enable_capture()
+ self.src_if.add_stream(self.fragments_400[0:-1])
+ self.pg_start()
+
+ self.logger.debug(self.vapi.cli("show trace"))
+ self.vapi.cli("clear trace")
+
+ self.src_if.add_stream(self.fragments_400[-1])
+ self.pg_start()
+ packets = self.dst_if.get_capture(len(self.pkt_infos))
+ self.verify_capture(packets)
+
def test_reversed(self):
""" reverse order reassembly """
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()
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) /
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) """
- worker_config = "workers %d" % worker_count
+ vpp_worker_count = 3
@classmethod
def setUpClass(cls):
- super(TestIPv6MWReassembly, cls).setUpClass()
+ super().setUpClass()
- cls.create_pg_interfaces(range(worker_count+1))
+ cls.create_pg_interfaces(range(cls.vpp_worker_count+1))
cls.src_if = cls.pg0
cls.send_ifs = cls.pg_interfaces[:-1]
cls.dst_if = cls.pg_interfaces[-1]
@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"))
def create_fragments(cls):
infos = cls._packet_infos
cls.pkt_infos = []
- for index, info in six.iteritems(infos):
+ for index, info in infos.items():
p = info.data
# cls.logger.debug(ppp("Packet:",
# p.__class__(scapy.compat.raw(p))))
"Packet with packet_index %d not received" % index)
def send_packets(self, packets):
- for counter in range(worker_count):
+ for counter in range(self.vpp_worker_count):
if 0 == len(packets[counter]):
continue
send_if = self.send_ifs[counter]
# in first wave we send fragments which don't start at offset 0
# then we send fragments with offset 0 on a different thread
# then the rest of packets on a random thread
- first_packets = [[] for n in range(worker_count)]
- second_packets = [[] for n in range(worker_count)]
- rest_of_packets = [[] for n in range(worker_count)]
+ first_packets = [[] for n in range(self.vpp_worker_count)]
+ second_packets = [[] for n in range(self.vpp_worker_count)]
+ rest_of_packets = [[] for n in range(self.vpp_worker_count)]
for (_, p) in self.pkt_infos:
- wi = randrange(worker_count)
+ wi = randrange(self.vpp_worker_count)
second_packets[wi].append(p[0])
if len(p) <= 1:
continue
wi2 = wi
while wi2 == wi:
- wi2 = randrange(worker_count)
+ wi2 = randrange(self.vpp_worker_count)
first_packets[wi2].append(p[1])
- wi3 = randrange(worker_count)
+ wi3 = randrange(self.vpp_worker_count)
rest_of_packets[wi3].extend(p[2:])
self.pg_enable_capture()
@classmethod
def setUpClass(cls):
- super(TestIPv6SVReassembly, cls).setUpClass()
+ super().setUpClass()
cls.create_pg_interfaces([0, 1])
cls.src_if = cls.pg0
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)
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,
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"))
self.assertEqual(sent[IPv6].dst, recvd[IPv6].dst)
self.assertEqual(sent[Raw].payload, recvd[Raw].payload)
+ def test_verify_clear_trace_mid_reassembly(self):
+ """ verify clear trace works mid-reassembly """
+ payload_len = 1000
+ payload = ""
+ counter = 0
+ while len(payload) < payload_len:
+ payload += "%u " % counter
+ counter += 1
+
+ p = (Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac) /
+ IPv6(src=self.src_if.remote_ip6, dst=self.dst_if.remote_ip6) /
+ UDP(sport=1234, dport=5678) /
+ Raw(payload))
+ fragments = fragment_rfc8200(p, 1, payload_len/4)
+
+ self.pg_enable_capture()
+ self.src_if.add_stream(fragments[1])
+ self.pg_start()
+
+ self.logger.debug(self.vapi.cli("show trace"))
+ self.vapi.cli("clear trace")
+
+ self.pg_enable_capture()
+ self.src_if.add_stream(fragments[0])
+ self.pg_start()
+ self.dst_if.get_capture(2)
+
+ self.logger.debug(self.vapi.cli("show trace"))
+ self.vapi.cli("clear trace")
+
+ self.pg_enable_capture()
+ self.src_if.add_stream(fragments[2:])
+ self.pg_start()
+ self.dst_if.get_capture(len(fragments[2:]))
+
def test_timeout(self):
""" reassembly timeout """
payload_len = 1000
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()
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
@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"))
def create_fragments(cls):
infos = cls._packet_infos
cls.pkt_infos = []
- for index, info in six.iteritems(infos):
+ for index, info in infos.items():
p = info.data
# cls.logger.debug(ppp("Packet:",
# p.__class__(scapy.compat.raw(p))))
@classmethod
def setUpClass(cls):
- super(TestFIFReassembly, cls).setUpClass()
+ super().setUpClass()
cls.create_pg_interfaces([0, 1])
cls.src_if = cls.pg0
@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)
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)
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"))
self.extend_packet(p, size, self.padding)
info.data = p[IP] # use only IP part, without ethernet header
- fragments = [x for _, p in six.iteritems(self._packet_infos)
+ fragments = [x for _, p in self._packet_infos.items()
for x in fragment_rfc791(p.data, 400)]
encapped_fragments = \
self.extend_packet(p, size, self.padding)
info.data = p[IPv6] # use only IPv6 part, without ethernet header
- fragments = [x for _, i in six.iteritems(self._packet_infos)
+ fragments = [x for _, i in self._packet_infos.items()
for x in fragment_rfc8200(
i.data, i.index, 400)]
Code Review / vpp.git / blobdiff