From 4c53313cd7e9b866412ad3e04b2d91ac098c1398 Mon Sep 17 00:00:00 2001 From: Klement Sekera Date: Thu, 22 Feb 2018 11:41:12 +0100 Subject: reassembly: feature/concurrency This change makes ip reassembly an interface feature, while adding concurrency support. Due to this, punt is no longer needed to test reassembly. Change-Id: I467669514ec33283ce935be0f1dd08f07684f0c7 Signed-off-by: Klement Sekera --- test/framework.py | 4 + test/sys_req/set_system_parameters.sh | 68 --- test/sys_req/system_parameters | 18 - test/test_reassembly.py | 972 ++++++++++++++++++++++++++++++++++ test/test_reassembly.py.disabled | 966 --------------------------------- test/vpp_papi_provider.py | 9 + test/vpp_punt_socket.py | 75 --- 7 files changed, 985 insertions(+), 1127 deletions(-) delete mode 100755 test/sys_req/set_system_parameters.sh delete mode 100644 test/sys_req/system_parameters create mode 100644 test/test_reassembly.py delete mode 100644 test/test_reassembly.py.disabled delete mode 100644 test/vpp_punt_socket.py (limited to 'test') diff --git a/test/framework.py b/test/framework.py index 423feaf7545..2167399b122 100644 --- a/test/framework.py +++ b/test/framework.py @@ -377,6 +377,10 @@ class VppTestCase(unittest.TestCase): try: cls.vapi.connect() except: + try: + cls.vapi.disconnect() + except: + pass if cls.debug_gdbserver: print(colorize("You're running VPP inside gdbserver but " "VPP-API connection failed, did you forget " diff --git a/test/sys_req/set_system_parameters.sh b/test/sys_req/set_system_parameters.sh deleted file mode 100755 index aa23634b435..00000000000 --- a/test/sys_req/set_system_parameters.sh +++ /dev/null @@ -1,68 +0,0 @@ -#!/bin/bash - -file="$1" - -usage(){ - echo "Usage: $0 " -} - -if [ "$file" == "" ] -then - echo "Invalid parameters specified." - usage - exit 1 -fi - -if [ ! -f $file ] -then - echo "File '$file' does not exist." - usage - exit 1 -fi - -if test "$DOCKER_TEST" = "True" -then - echo "==============================================================================" - echo "DOCKER_TEST is set to '$DOCKER_TEST'." - echo "Skipping verification of some system parameters." - echo "Make sure these are set properly, otherwise tests might fail." - echo "Required values/criteria are in '`readlink -e $file`'." - echo "==============================================================================" - exit 0 -fi - -cat $file | grep -v -e '^#.*$' | grep -v -e '^ *$' | while read line -do - value_file=`echo $line | awk '{print $1}'` - operator=`echo $line | awk '{print $2}'` - value=`echo $line | awk '{print $3}'` - set_value=`echo $line | awk '{print $4}'` - if [[ "$value_file" == "" || "$operator" == "" || "$value" == "" || "$set_value" == "" ]] - then - echo "Syntax error in requirements file." - exit 1 - fi - current_value=`cat $value_file` - if test "$current_value" $operator "$value" - then - if test "$V" = "2" - then - echo "Requirement '$value_file $operator $value' satisfied." - fi - else - echo "Requirement '$value_file $operator $value' not satisfied." - echo "Writing '$set_value' to '$value_file'." - echo "$set_value" | tee "$value_file" > /dev/null - if ! test "`cat $value_file`" = "$set_value" - then - echo "Repeating the write using sudo..." - echo "$set_value" | sudo -n tee "$value_file" > /dev/null - if ! test "`cat $value_file`" = "$set_value" - then - echo "Couldn't set the required value. Is that value allowed? Is sudo working?" - exit 1 - fi - fi - echo "Succesfully wrote '$set_value' to '$value_file'." - fi -done diff --git a/test/sys_req/system_parameters b/test/sys_req/system_parameters deleted file mode 100644 index 6373774dd94..00000000000 --- a/test/sys_req/system_parameters +++ /dev/null @@ -1,18 +0,0 @@ -# test framework system requirements -# format of this file is -# -# -# path - path to value e.g. in /proc which needs to be checked -# operator - test operator (e.g. -gt) -# comparison-value - value, against which the value read from is compared -# set-value - value, to which the path is set if the test fails -# -# the comparison is done using `test' command - - -# test_reassembly.py -# needed by test_reassembly which uses udp punt via unix domain sockets -# to ensure that all data which vpp might produce in a burst fits into -# the socket send buffer -/proc/sys/net/core/wmem_max -ge 4636252 4636252 -/proc/sys/net/core/wmem_default -ge 4636252 4636252 diff --git a/test/test_reassembly.py b/test/test_reassembly.py new file mode 100644 index 00000000000..531caa46257 --- /dev/null +++ b/test/test_reassembly.py @@ -0,0 +1,972 @@ +#!/usr/bin/env python +import unittest +from random import shuffle + +from framework import VppTestCase, VppTestRunner + +from scapy.packet import Raw +from scapy.layers.l2 import Ether, GRE +from scapy.layers.inet import IP, UDP +from util import ppp, fragment_rfc791, fragment_rfc8200 +from scapy.layers.inet6 import IPv6, IPv6ExtHdrFragment, ICMPv6ParamProblem,\ + ICMPv6TimeExceeded +from vpp_gre_interface import VppGreInterface, VppGre6Interface +from vpp_ip_route import VppIpRoute, VppRoutePath, DpoProto + +test_packet_count = 257 + + +class TestIPv4Reassembly(VppTestCase): + """ IPv4 Reassembly """ + + @classmethod + def setUpClass(cls): + super(TestIPv4Reassembly, cls).setUpClass() + + cls.create_pg_interfaces([0, 1]) + cls.src_if = cls.pg0 + cls.dst_if = cls.pg1 + + # setup all interfaces + for i in cls.pg_interfaces: + i.admin_up() + i.config_ip4() + i.resolve_arp() + + # packet sizes + cls.packet_sizes = [64, 512, 1518, 9018] + cls.padding = " abcdefghijklmn" + cls.create_stream(cls.packet_sizes) + cls.create_fragments() + + def setUp(self): + """ Test setup - force timeout on existing reassemblies """ + super(TestIPv4Reassembly, self).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, + expire_walk_interval_ms=10) + self.sleep(.25) + self.vapi.ip_reassembly_set(timeout_ms=1000000, max_reassemblies=1000, + expire_walk_interval_ms=10000) + + def tearDown(self): + super(TestIPv4Reassembly, self).tearDown() + self.logger.debug(self.vapi.ppcli("show ip4-reassembly details")) + + @classmethod + def create_stream(cls, packet_sizes, packet_count=test_packet_count): + """Create input packet stream for defined interface. + + :param list packet_sizes: Required packet sizes. + """ + for i in range(0, packet_count): + info = cls.create_packet_info(cls.src_if, cls.src_if) + payload = cls.info_to_payload(info) + p = (Ether(dst=cls.src_if.local_mac, src=cls.src_if.remote_mac) / + IP(id=info.index, src=cls.src_if.remote_ip4, + dst=cls.dst_if.remote_ip4) / + UDP(sport=1234, dport=5678) / + Raw(payload)) + size = packet_sizes[(i // 2) % len(packet_sizes)] + cls.extend_packet(p, size, cls.padding) + info.data = p + + @classmethod + def create_fragments(cls): + infos = cls._packet_infos + cls.pkt_infos = [] + for index, info in infos.iteritems(): + p = info.data + # cls.logger.debug(ppp("Packet:", p.__class__(str(p)))) + fragments_400 = fragment_rfc791(p, 400) + fragments_300 = fragment_rfc791(p, 300) + fragments_200 = [ + x for f in fragments_400 for x in fragment_rfc791(f, 200)] + cls.pkt_infos.append( + (index, fragments_400, fragments_300, fragments_200)) + cls.fragments_400 = [ + x for (_, frags, _, _) in cls.pkt_infos for x in frags] + cls.fragments_300 = [ + x for (_, _, frags, _) in cls.pkt_infos for x in frags] + cls.fragments_200 = [ + x for (_, _, _, frags) in cls.pkt_infos for x in frags] + cls.logger.debug("Fragmented %s packets into %s 400-byte fragments, " + "%s 300-byte fragments and %s 200-byte fragments" % + (len(infos), len(cls.fragments_400), + len(cls.fragments_300), len(cls.fragments_200))) + + def verify_capture(self, capture, dropped_packet_indexes=[]): + """Verify captured packet stream. + + :param list capture: Captured packet stream. + """ + info = None + seen = set() + for packet in capture: + try: + self.logger.debug(ppp("Got packet:", packet)) + ip = packet[IP] + udp = packet[UDP] + payload_info = self.payload_to_info(str(packet[Raw])) + packet_index = payload_info.index + self.assertTrue( + packet_index not in dropped_packet_indexes, + ppp("Packet received, but should be dropped:", packet)) + if packet_index in seen: + raise Exception(ppp("Duplicate packet received", packet)) + seen.add(packet_index) + self.assertEqual(payload_info.dst, self.src_if.sw_if_index) + info = self._packet_infos[packet_index] + self.assertTrue(info is not None) + self.assertEqual(packet_index, info.index) + saved_packet = info.data + self.assertEqual(ip.src, saved_packet[IP].src) + self.assertEqual(ip.dst, saved_packet[IP].dst) + self.assertEqual(udp.payload, saved_packet[UDP].payload) + except Exception: + self.logger.error(ppp("Unexpected or invalid packet:", packet)) + raise + for index in self._packet_infos: + self.assertTrue(index in seen or index in dropped_packet_indexes, + "Packet with packet_index %d not received" % index) + + def test_reassembly(self): + """ basic reassembly """ + + self.pg_enable_capture() + self.src_if.add_stream(self.fragments_200) + self.pg_start() + + packets = self.dst_if.get_capture(len(self.pkt_infos)) + self.verify_capture(packets) + self.src_if.assert_nothing_captured() + + # run it all again to verify correctness + self.pg_enable_capture() + self.src_if.add_stream(self.fragments_200) + self.pg_start() + + packets = self.dst_if.get_capture(len(self.pkt_infos)) + self.verify_capture(packets) + self.src_if.assert_nothing_captured() + + def test_reversed(self): + """ reverse order reassembly """ + + fragments = list(self.fragments_200) + fragments.reverse() + + self.pg_enable_capture() + self.src_if.add_stream(fragments) + self.pg_start() + + packets = self.dst_if.get_capture(len(self.packet_infos)) + self.verify_capture(packets) + self.src_if.assert_nothing_captured() + + # run it all again to verify correctness + self.pg_enable_capture() + self.src_if.add_stream(fragments) + self.pg_start() + + packets = self.dst_if.get_capture(len(self.packet_infos)) + self.verify_capture(packets) + self.src_if.assert_nothing_captured() + + def test_random(self): + """ random order reassembly """ + + fragments = list(self.fragments_200) + shuffle(fragments) + + self.pg_enable_capture() + self.src_if.add_stream(fragments) + self.pg_start() + + packets = self.dst_if.get_capture(len(self.packet_infos)) + self.verify_capture(packets) + self.src_if.assert_nothing_captured() + + # run it all again to verify correctness + self.pg_enable_capture() + self.src_if.add_stream(fragments) + self.pg_start() + + packets = self.dst_if.get_capture(len(self.packet_infos)) + self.verify_capture(packets) + self.src_if.assert_nothing_captured() + + def test_duplicates(self): + """ duplicate fragments """ + + fragments = [ + x for (_, frags, _, _) in self.pkt_infos + for x in frags + for _ in range(0, min(2, len(frags))) + ] + + self.pg_enable_capture() + self.src_if.add_stream(fragments) + self.pg_start() + + packets = self.dst_if.get_capture(len(self.pkt_infos)) + self.verify_capture(packets) + self.src_if.assert_nothing_captured() + + def test_overlap1(self): + """ overlapping fragments case #1 """ + + fragments = [] + for _, _, frags_300, frags_200 in self.pkt_infos: + if len(frags_300) == 1: + fragments.extend(frags_300) + else: + for i, j in zip(frags_200, frags_300): + fragments.extend(i) + fragments.extend(j) + + self.pg_enable_capture() + self.src_if.add_stream(fragments) + self.pg_start() + + packets = self.dst_if.get_capture(len(self.pkt_infos)) + self.verify_capture(packets) + self.src_if.assert_nothing_captured() + + # run it all to verify correctness + self.pg_enable_capture() + self.src_if.add_stream(fragments) + self.pg_start() + + packets = self.dst_if.get_capture(len(self.pkt_infos)) + self.verify_capture(packets) + self.src_if.assert_nothing_captured() + + def test_overlap2(self): + """ overlapping fragments case #2 """ + + fragments = [] + for _, _, frags_300, frags_200 in self.pkt_infos: + if len(frags_300) == 1: + fragments.extend(frags_300) + else: + # care must be taken here so that there are no fragments + # received by vpp after reassembly is finished, otherwise + # new reassemblies will be started and packet generator will + # freak out when it detects unfreed buffers + zipped = zip(frags_300, frags_200) + for i, j in zipped[:-1]: + fragments.extend(i) + fragments.extend(j) + fragments.append(zipped[-1][0]) + + self.pg_enable_capture() + self.src_if.add_stream(fragments) + self.pg_start() + + packets = self.dst_if.get_capture(len(self.pkt_infos)) + self.verify_capture(packets) + self.src_if.assert_nothing_captured() + + # run it all to verify correctness + self.pg_enable_capture() + self.src_if.add_stream(fragments) + self.pg_start() + + packets = self.dst_if.get_capture(len(self.pkt_infos)) + self.verify_capture(packets) + self.src_if.assert_nothing_captured() + + def test_timeout_inline(self): + """ timeout (inline) """ + + dropped_packet_indexes = set( + index for (index, frags, _, _) in self.pkt_infos if len(frags) > 1 + ) + + self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000, + expire_walk_interval_ms=10000) + + self.pg_enable_capture() + self.src_if.add_stream(self.fragments_400) + self.pg_start() + + packets = self.dst_if.get_capture( + len(self.pkt_infos) - len(dropped_packet_indexes)) + self.verify_capture(packets, dropped_packet_indexes) + self.src_if.assert_nothing_captured() + + def test_timeout_cleanup(self): + """ timeout (cleanup) """ + + # whole packets + fragmented packets sans last fragment + fragments = [ + x for (_, frags_400, _, _) in self.pkt_infos + for x in frags_400[:-1 if len(frags_400) > 1 else None] + ] + + # last fragments for fragmented packets + fragments2 = [frags_400[-1] + for (_, frags_400, _, _) in self.pkt_infos + if len(frags_400) > 1] + + dropped_packet_indexes = set( + index for (index, frags_400, _, _) in self.pkt_infos + if len(frags_400) > 1) + + self.vapi.ip_reassembly_set(timeout_ms=100, max_reassemblies=1000, + expire_walk_interval_ms=50) + + self.pg_enable_capture() + self.src_if.add_stream(fragments) + self.pg_start() + + self.sleep(.25, "wait before sending rest of fragments") + + self.src_if.add_stream(fragments2) + self.pg_start() + + packets = self.dst_if.get_capture( + len(self.pkt_infos) - len(dropped_packet_indexes)) + self.verify_capture(packets, dropped_packet_indexes) + self.src_if.assert_nothing_captured() + + def test_disabled(self): + """ reassembly disabled """ + + dropped_packet_indexes = set( + index for (index, frags_400, _, _) in self.pkt_infos + if len(frags_400) > 1) + + self.vapi.ip_reassembly_set(timeout_ms=1000, max_reassemblies=0, + expire_walk_interval_ms=10000) + + self.pg_enable_capture() + self.src_if.add_stream(self.fragments_400) + self.pg_start() + + packets = self.dst_if.get_capture( + len(self.pkt_infos) - len(dropped_packet_indexes)) + self.verify_capture(packets, dropped_packet_indexes) + self.src_if.assert_nothing_captured() + + +class TestIPv6Reassembly(VppTestCase): + """ IPv6 Reassembly """ + + @classmethod + def setUpClass(cls): + super(TestIPv6Reassembly, cls).setUpClass() + + cls.create_pg_interfaces([0, 1]) + cls.src_if = cls.pg0 + cls.dst_if = cls.pg1 + + # setup all interfaces + for i in cls.pg_interfaces: + i.admin_up() + i.config_ip6() + i.resolve_ndp() + + # packet sizes + cls.packet_sizes = [64, 512, 1518, 9018] + cls.padding = " abcdefghijklmn" + cls.create_stream(cls.packet_sizes) + cls.create_fragments() + + def setUp(self): + """ Test setup - force timeout on existing reassemblies """ + super(TestIPv6Reassembly, self).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, + expire_walk_interval_ms=10, is_ip6=1) + self.sleep(.25) + self.vapi.ip_reassembly_set(timeout_ms=1000000, max_reassemblies=1000, + expire_walk_interval_ms=10000, is_ip6=1) + self.logger.debug(self.vapi.ppcli("show ip6-reassembly details")) + + def tearDown(self): + super(TestIPv6Reassembly, self).tearDown() + self.logger.debug(self.vapi.ppcli("show ip6-reassembly details")) + + @classmethod + def create_stream(cls, packet_sizes, packet_count=test_packet_count): + """Create input packet stream for defined interface. + + :param list packet_sizes: Required packet sizes. + """ + for i in range(0, packet_count): + info = cls.create_packet_info(cls.src_if, cls.src_if) + payload = cls.info_to_payload(info) + p = (Ether(dst=cls.src_if.local_mac, src=cls.src_if.remote_mac) / + IPv6(src=cls.src_if.remote_ip6, + dst=cls.dst_if.remote_ip6) / + UDP(sport=1234, dport=5678) / + Raw(payload)) + size = packet_sizes[(i // 2) % len(packet_sizes)] + cls.extend_packet(p, size, cls.padding) + info.data = p + + @classmethod + def create_fragments(cls): + infos = cls._packet_infos + cls.pkt_infos = [] + for index, info in infos.iteritems(): + p = info.data + # cls.logger.debug(ppp("Packet:", p.__class__(str(p)))) + fragments_400 = fragment_rfc8200(p, info.index, 400) + fragments_300 = fragment_rfc8200(p, info.index, 300) + cls.pkt_infos.append((index, fragments_400, fragments_300)) + cls.fragments_400 = [ + x for _, frags, _ in cls.pkt_infos for x in frags] + cls.fragments_300 = [ + x for _, _, frags in cls.pkt_infos for x in frags] + cls.logger.debug("Fragmented %s packets into %s 400-byte fragments, " + "and %s 300-byte fragments" % + (len(infos), len(cls.fragments_400), + len(cls.fragments_300))) + + def verify_capture(self, capture, dropped_packet_indexes=[]): + """Verify captured packet strea . + + :param list capture: Captured packet stream. + """ + info = None + seen = set() + for packet in capture: + try: + self.logger.debug(ppp("Got packet:", packet)) + ip = packet[IPv6] + udp = packet[UDP] + payload_info = self.payload_to_info(str(packet[Raw])) + packet_index = payload_info.index + self.assertTrue( + packet_index not in dropped_packet_indexes, + ppp("Packet received, but should be dropped:", packet)) + if packet_index in seen: + raise Exception(ppp("Duplicate packet received", packet)) + seen.add(packet_index) + self.assertEqual(payload_info.dst, self.src_if.sw_if_index) + info = self._packet_infos[packet_index] + self.assertTrue(info is not None) + self.assertEqual(packet_index, info.index) + saved_packet = info.data + self.assertEqual(ip.src, saved_packet[IPv6].src) + self.assertEqual(ip.dst, saved_packet[IPv6].dst) + self.assertEqual(udp.payload, saved_packet[UDP].payload) + except Exception: + self.logger.error(ppp("Unexpected or invalid packet:", packet)) + raise + for index in self._packet_infos: + self.assertTrue(index in seen or index in dropped_packet_indexes, + "Packet with packet_index %d not received" % index) + + def test_reassembly(self): + """ basic reassembly """ + + self.pg_enable_capture() + self.src_if.add_stream(self.fragments_400) + self.pg_start() + + packets = self.dst_if.get_capture(len(self.pkt_infos)) + self.verify_capture(packets) + self.src_if.assert_nothing_captured() + + # run it all again to verify correctness + self.pg_enable_capture() + self.src_if.add_stream(self.fragments_400) + self.pg_start() + + packets = self.dst_if.get_capture(len(self.pkt_infos)) + self.verify_capture(packets) + self.src_if.assert_nothing_captured() + + def test_reversed(self): + """ reverse order reassembly """ + + fragments = list(self.fragments_400) + fragments.reverse() + + self.pg_enable_capture() + self.src_if.add_stream(fragments) + self.pg_start() + + packets = self.dst_if.get_capture(len(self.pkt_infos)) + self.verify_capture(packets) + self.src_if.assert_nothing_captured() + + # run it all again to verify correctness + self.pg_enable_capture() + self.src_if.add_stream(fragments) + self.pg_start() + + packets = self.dst_if.get_capture(len(self.pkt_infos)) + self.verify_capture(packets) + self.src_if.assert_nothing_captured() + + def test_random(self): + """ random order reassembly """ + + fragments = list(self.fragments_400) + shuffle(fragments) + + self.pg_enable_capture() + self.src_if.add_stream(fragments) + self.pg_start() + + packets = self.dst_if.get_capture(len(self.pkt_infos)) + self.verify_capture(packets) + self.src_if.assert_nothing_captured() + + # run it all again to verify correctness + self.pg_enable_capture() + self.src_if.add_stream(fragments) + self.pg_start() + + packets = self.dst_if.get_capture(len(self.pkt_infos)) + self.verify_capture(packets) + self.src_if.assert_nothing_captured() + + def test_duplicates(self): + """ duplicate fragments """ + + fragments = [ + x for (_, frags, _) in self.pkt_infos + for x in frags + for _ in range(0, min(2, len(frags))) + ] + + self.pg_enable_capture() + self.src_if.add_stream(fragments) + self.pg_start() + + packets = self.dst_if.get_capture(len(self.pkt_infos)) + self.verify_capture(packets) + self.src_if.assert_nothing_captured() + + def test_overlap1(self): + """ overlapping fragments case #1 """ + + fragments = [] + for _, frags_400, frags_300 in self.pkt_infos: + if len(frags_300) == 1: + fragments.extend(frags_400) + else: + for i, j in zip(frags_300, frags_400): + fragments.extend(i) + fragments.extend(j) + + dropped_packet_indexes = set( + index for (index, _, frags) in self.pkt_infos if len(frags) > 1 + ) + + self.pg_enable_capture() + self.src_if.add_stream(fragments) + self.pg_start() + + packets = self.dst_if.get_capture( + len(self.pkt_infos) - len(dropped_packet_indexes)) + self.verify_capture(packets, dropped_packet_indexes) + self.src_if.assert_nothing_captured() + + def test_overlap2(self): + """ overlapping fragments case #2 """ + + fragments = [] + for _, frags_400, frags_300 in self.pkt_infos: + if len(frags_400) == 1: + fragments.extend(frags_400) + else: + # care must be taken here so that there are no fragments + # received by vpp after reassembly is finished, otherwise + # new reassemblies will be started and packet generator will + # freak out when it detects unfreed buffers + zipped = zip(frags_400, frags_300) + for i, j in zipped[:-1]: + fragments.extend(i) + fragments.extend(j) + fragments.append(zipped[-1][0]) + + dropped_packet_indexes = set( + index for (index, _, frags) in self.pkt_infos if len(frags) > 1 + ) + + self.pg_enable_capture() + self.src_if.add_stream(fragments) + self.pg_start() + + packets = self.dst_if.get_capture( + len(self.pkt_infos) - len(dropped_packet_indexes)) + self.verify_capture(packets, dropped_packet_indexes) + self.src_if.assert_nothing_captured() + + def test_timeout_inline(self): + """ timeout (inline) """ + + dropped_packet_indexes = set( + index for (index, frags, _) in self.pkt_infos if len(frags) > 1 + ) + + self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000, + expire_walk_interval_ms=10000, is_ip6=1) + + self.pg_enable_capture() + self.src_if.add_stream(self.fragments_400) + self.pg_start() + + 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)) + for icmp in pkts: + self.assertIn(ICMPv6TimeExceeded, icmp) + self.assertIn(IPv6ExtHdrFragment, icmp) + self.assertIn(icmp[IPv6ExtHdrFragment].id, dropped_packet_indexes) + dropped_packet_indexes.remove(icmp[IPv6ExtHdrFragment].id) + + def test_timeout_cleanup(self): + """ timeout (cleanup) """ + + # whole packets + fragmented packets sans last fragment + fragments = [ + x for (_, frags_400, _) in self.pkt_infos + for x in frags_400[:-1 if len(frags_400) > 1 else None] + ] + + # last fragments for fragmented packets + fragments2 = [frags_400[-1] + for (_, frags_400, _) in self.pkt_infos + if len(frags_400) > 1] + + dropped_packet_indexes = set( + index for (index, frags_400, _) in self.pkt_infos + if len(frags_400) > 1) + + self.vapi.ip_reassembly_set(timeout_ms=100, max_reassemblies=1000, + expire_walk_interval_ms=50) + + self.vapi.ip_reassembly_set(timeout_ms=100, max_reassemblies=1000, + expire_walk_interval_ms=50, is_ip6=1) + + self.pg_enable_capture() + self.src_if.add_stream(fragments) + self.pg_start() + + self.sleep(.25, "wait before sending rest of fragments") + + self.src_if.add_stream(fragments2) + self.pg_start() + + 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)) + for icmp in pkts: + self.assertIn(ICMPv6TimeExceeded, icmp) + self.assertIn(IPv6ExtHdrFragment, icmp) + self.assertIn(icmp[IPv6ExtHdrFragment].id, dropped_packet_indexes) + dropped_packet_indexes.remove(icmp[IPv6ExtHdrFragment].id) + + def test_disabled(self): + """ reassembly disabled """ + + dropped_packet_indexes = set( + index for (index, frags_400, _) in self.pkt_infos + if len(frags_400) > 1) + + self.vapi.ip_reassembly_set(timeout_ms=1000, max_reassemblies=0, + expire_walk_interval_ms=10000, is_ip6=1) + + self.pg_enable_capture() + self.src_if.add_stream(self.fragments_400) + self.pg_start() + + packets = self.dst_if.get_capture( + len(self.pkt_infos) - len(dropped_packet_indexes)) + self.verify_capture(packets, dropped_packet_indexes) + self.src_if.assert_nothing_captured() + + def test_missing_upper(self): + """ missing upper layer """ + 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__(str(fragments[1])) + bad_fragment[IPv6ExtHdrFragment].nh = 59 + bad_fragment[IPv6ExtHdrFragment].offset = 0 + self.pg_enable_capture() + self.src_if.add_stream([bad_fragment]) + 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_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) / + 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 = fragments[0] + self.extend_packet(bad_fragment, len(bad_fragment) + 5) + self.pg_enable_capture() + self.src_if.add_stream([bad_fragment]) + 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, 0, "ICMP code") + + def test_invalid_packet_size(self): + """ total packet size > 65535 """ + 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 = fragments[1] + bad_fragment[IPv6ExtHdrFragment].offset = 65500 + self.pg_enable_capture() + self.src_if.add_stream([bad_fragment]) + 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, 0, "ICMP code") + + +@unittest.skip("removing GRE tunnels broken, need fix") +class TestFIFReassembly(VppTestCase): + """ Fragments in fragments reassembly """ + + @classmethod + def setUpClass(cls): + super(TestFIFReassembly, cls).setUpClass() + + cls.create_pg_interfaces([0, 1]) + cls.src_if = cls.pg0 + cls.dst_if = cls.pg1 + for i in cls.pg_interfaces: + i.admin_up() + i.config_ip4() + i.resolve_arp() + i.config_ip6() + i.resolve_ndp() + + cls.packet_sizes = [64, 512, 1518, 9018] + cls.padding = " abcdefghijklmn" + + def setUp(self): + """ Test setup - force timeout on existing reassemblies """ + super(TestFIFReassembly, self).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_enable_disable( + sw_if_index=self.dst_if.sw_if_index, enable_ip4=True, + enable_ip6=True) + self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000, + expire_walk_interval_ms=10) + self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000, + expire_walk_interval_ms=10, is_ip6=1) + self.sleep(.25) + self.vapi.ip_reassembly_set(timeout_ms=1000000, max_reassemblies=1000, + expire_walk_interval_ms=10000) + self.vapi.ip_reassembly_set(timeout_ms=1000000, max_reassemblies=1000, + expire_walk_interval_ms=10000, is_ip6=1) + + def tearDown(self): + self.logger.debug(self.vapi.ppcli("show ip4-reassembly details")) + self.logger.debug(self.vapi.ppcli("show ip6-reassembly details")) + super(TestFIFReassembly, self).tearDown() + + def verify_capture(self, capture, ip_class, dropped_packet_indexes=[]): + """Verify captured packet stream. + + :param list capture: Captured packet stream. + """ + info = None + seen = set() + for packet in capture: + try: + self.logger.debug(ppp("Got packet:", packet)) + ip = packet[ip_class] + udp = packet[UDP] + payload_info = self.payload_to_info(str(packet[Raw])) + packet_index = payload_info.index + self.assertTrue( + packet_index not in dropped_packet_indexes, + ppp("Packet received, but should be dropped:", packet)) + if packet_index in seen: + raise Exception(ppp("Duplicate packet received", packet)) + seen.add(packet_index) + self.assertEqual(payload_info.dst, self.dst_if.sw_if_index) + info = self._packet_infos[packet_index] + self.assertTrue(info is not None) + self.assertEqual(packet_index, info.index) + saved_packet = info.data + self.assertEqual(ip.src, saved_packet[ip_class].src) + self.assertEqual(ip.dst, saved_packet[ip_class].dst) + self.assertEqual(udp.payload, saved_packet[UDP].payload) + except Exception: + self.logger.error(ppp("Unexpected or invalid packet:", packet)) + raise + for index in self._packet_infos: + self.assertTrue(index in seen or index in dropped_packet_indexes, + "Packet with packet_index %d not received" % index) + + def test_fif4(self): + """ Fragments in fragments (4o4) """ + + # TODO this should be ideally in setUpClass, but then we hit a bug + # with VppIpRoute incorrectly reporting it's present when it's not + # so we need to manually remove the vpp config, thus we cannot have + # it shared for multiple test cases + self.tun_ip4 = "1.1.1.2" + + self.gre4 = VppGreInterface(self, self.src_if.local_ip4, self.tun_ip4) + self.gre4.add_vpp_config() + self.gre4.admin_up() + self.gre4.config_ip4() + + self.vapi.ip_reassembly_enable_disable( + sw_if_index=self.gre4.sw_if_index, enable_ip4=True) + + self.route4 = VppIpRoute(self, self.tun_ip4, 32, + [VppRoutePath(self.src_if.remote_ip4, + self.src_if.sw_if_index)]) + self.route4.add_vpp_config() + + self.reset_packet_infos() + for i in range(test_packet_count): + info = self.create_packet_info(self.src_if, self.dst_if) + payload = self.info_to_payload(info) + # Ethernet header here is only for size calculation, thus it + # doesn't matter how it's initialized. This is to ensure that + # reassembled packet is not > 9000 bytes, so that it's not dropped + p = (Ether() / + IP(id=i, src=self.src_if.remote_ip4, + dst=self.dst_if.remote_ip4) / + UDP(sport=1234, dport=5678) / + Raw(payload)) + size = self.packet_sizes[(i // 2) % len(self.packet_sizes)] + self.extend_packet(p, size, self.padding) + info.data = p[IP] # use only IP part, without ethernet header + + fragments = [x for _, p in self._packet_infos.iteritems() + for x in fragment_rfc791(p.data, 400)] + + encapped_fragments = \ + [Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac) / + IP(src=self.tun_ip4, dst=self.src_if.local_ip4) / + GRE() / + p + for p in fragments] + + fragmented_encapped_fragments = \ + [x for p in encapped_fragments + for x in fragment_rfc791(p, 200)] + + self.src_if.add_stream(fragmented_encapped_fragments) + + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + + self.src_if.assert_nothing_captured() + packets = self.dst_if.get_capture(len(self._packet_infos)) + self.verify_capture(packets, IP) + + # TODO remove gre vpp config by hand until VppIpRoute gets fixed + # so that it's query_vpp_config() works as it should + self.gre4.remove_vpp_config() + self.logger.debug(self.vapi.ppcli("show interface")) + + def test_fif6(self): + """ Fragments in fragments (6o6) """ + # TODO this should be ideally in setUpClass, but then we hit a bug + # with VppIpRoute incorrectly reporting it's present when it's not + # so we need to manually remove the vpp config, thus we cannot have + # it shared for multiple test cases + self.tun_ip6 = "1002::1" + + self.gre6 = VppGre6Interface(self, self.src_if.local_ip6, self.tun_ip6) + self.gre6.add_vpp_config() + self.gre6.admin_up() + self.gre6.config_ip6() + + self.vapi.ip_reassembly_enable_disable( + sw_if_index=self.gre6.sw_if_index, enable_ip6=True) + + self.route6 = VppIpRoute(self, self.tun_ip6, 128, + [VppRoutePath(self.src_if.remote_ip6, + self.src_if.sw_if_index, + proto=DpoProto.DPO_PROTO_IP6)], + is_ip6=1) + self.route6.add_vpp_config() + + self.reset_packet_infos() + for i in range(test_packet_count): + info = self.create_packet_info(self.src_if, self.dst_if) + payload = self.info_to_payload(info) + # Ethernet header here is only for size calculation, thus it + # doesn't matter how it's initialized. This is to ensure that + # reassembled packet is not > 9000 bytes, so that it's not dropped + p = (Ether() / + IPv6(src=self.src_if.remote_ip6, dst=self.dst_if.remote_ip6) / + UDP(sport=1234, dport=5678) / + Raw(payload)) + size = self.packet_sizes[(i // 2) % len(self.packet_sizes)] + self.extend_packet(p, size, self.padding) + info.data = p[IPv6] # use only IPv6 part, without ethernet header + + fragments = [x for _, i in self._packet_infos.iteritems() + for x in fragment_rfc8200( + i.data, i.index, 400)] + + encapped_fragments = \ + [Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac) / + IPv6(src=self.tun_ip6, dst=self.src_if.local_ip6) / + GRE() / + p + for p in fragments] + + fragmented_encapped_fragments = \ + [x for p in encapped_fragments for x in ( + fragment_rfc8200( + p, + 2 * len(self._packet_infos) + p[IPv6ExtHdrFragment].id, + 200) + if IPv6ExtHdrFragment in p else [p] + ) + ] + + self.src_if.add_stream(fragmented_encapped_fragments) + + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + + self.src_if.assert_nothing_captured() + packets = self.dst_if.get_capture(len(self._packet_infos)) + self.verify_capture(packets, IPv6) + + # TODO remove gre vpp config by hand until VppIpRoute gets fixed + # so that it's query_vpp_config() works as it should + # self.gre6.remove_vpp_config() + + +if __name__ == '__main__': + unittest.main(testRunner=VppTestRunner) diff --git a/test/test_reassembly.py.disabled b/test/test_reassembly.py.disabled deleted file mode 100644 index a2d77d493de..00000000000 --- a/test/test_reassembly.py.disabled +++ /dev/null @@ -1,966 +0,0 @@ -#!/usr/bin/env python -import unittest -from random import shuffle - -from framework import VppTestCase, VppTestRunner - -from scapy.packet import Raw -from scapy.layers.l2 import Ether, GRE -from scapy.layers.inet import IP, UDP -from util import ppp, fragment_rfc791, fragment_rfc8200 -from vpp_punt_socket import VppUDSPuntSocket -from scapy.layers.inet6 import IPv6, IPv6ExtHdrFragment, ICMPv6ParamProblem,\ - ICMPv6TimeExceeded -from vpp_gre_interface import VppGreInterface, VppGre6Interface -from vpp_ip_route import VppIpRoute, VppRoutePath, DpoProto - -test_packet_count = 257 - - -class TestIPv4Reassembly(VppTestCase): - """ IPv4 Reassembly """ - - @classmethod - def setUpClass(cls): - super(TestIPv4Reassembly, cls).setUpClass() - - cls.create_pg_interfaces([0]) - cls.pg_if = cls.pg0 - - # setup all interfaces - for i in cls.pg_interfaces: - i.admin_up() - i.config_ip4() - i.resolve_arp() - - cls.punt_port = 9999 - cls.punt_socket = VppUDSPuntSocket(cls, cls.punt_port) - - # packet sizes - cls.packet_sizes = [64, 512, 1518, 9018] - cls.padding = " abcdefghijklmn" - cls.create_stream(cls.packet_sizes) - cls.create_fragments() - - def setUp(self): - """ Test setup - force timeout on existing reassemblies """ - super(TestIPv4Reassembly, self).setUp() - self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000, - expire_walk_interval_ms=10) - self.sleep(.25) - self.vapi.ip_reassembly_set(timeout_ms=1000000, max_reassemblies=1000, - expire_walk_interval_ms=10000) - - def tearDown(self): - super(TestIPv4Reassembly, self).tearDown() - self.logger.debug(self.vapi.ppcli("show ip4-reassembly details")) - - @classmethod - def create_stream(cls, packet_sizes, packet_count=test_packet_count): - """Create input packet stream for defined interface. - - :param list packet_sizes: Required packet sizes. - """ - for i in range(0, packet_count): - info = cls.create_packet_info(cls.pg_if, cls.pg_if) - payload = cls.info_to_payload(info) - p = (Ether(dst=cls.pg_if.local_mac, src=cls.pg_if.remote_mac) / - IP(id=info.index, src=cls.pg_if.remote_ip4, - dst=cls.pg_if.local_ip4) / - UDP(sport=1234, dport=cls.punt_port) / - Raw(payload)) - size = packet_sizes[(i // 2) % len(packet_sizes)] - cls.extend_packet(p, size, cls.padding) - info.data = p - - @classmethod - def create_fragments(cls): - infos = cls._packet_infos - cls.pkt_infos = [] - for index, info in infos.iteritems(): - p = info.data - # self.logger.debug(ppp("Packet:", p.__class__(str(p)))) - fragments_400 = fragment_rfc791(p, 400) - fragments_300 = fragment_rfc791(p, 300) - fragments_200 = [ - x for f in fragments_400 for x in fragment_rfc791(f, 200)] - cls.pkt_infos.append( - (index, fragments_400, fragments_300, fragments_200)) - cls.fragments_400 = [ - x for (_, frags, _, _) in cls.pkt_infos for x in frags] - cls.fragments_300 = [ - x for (_, _, frags, _) in cls.pkt_infos for x in frags] - cls.fragments_200 = [ - x for (_, _, _, frags) in cls.pkt_infos for x in frags] - cls.logger.debug("Fragmented %s packets into %s 400-byte fragments, " - "%s 300-byte fragments and %s 200-byte fragments" % - (len(infos), len(cls.fragments_400), - len(cls.fragments_300), len(cls.fragments_200))) - - def verify_capture(self, capture, dropped_packet_indexes=[]): - """Verify captured packet stream. - - :param list capture: Captured packet stream. - """ - info = None - seen = set() - for packet in capture: - try: - sw_if_index = packet['sw_if_index'] - punt_action = packet['punt_action'] - packet = Ether(packet['packet']) - self.logger.debug(ppp("Got packet from %s, action %s" % - (sw_if_index, punt_action), packet)) - ip = packet[IP] - udp = packet[UDP] - payload_info = self.payload_to_info(str(packet[Raw])) - packet_index = payload_info.index - self.assertTrue( - packet_index not in dropped_packet_indexes, - ppp("Packet received, but should be dropped:", packet)) - if packet_index in seen: - raise Exception(ppp("Duplicate packet received", packet)) - seen.add(packet_index) - self.assertEqual(payload_info.dst, self.pg_if.sw_if_index) - info = self._packet_infos[packet_index] - self.assertTrue(info is not None) - self.assertEqual(packet_index, info.index) - saved_packet = info.data - self.assertEqual(ip.src, saved_packet[IP].src) - self.assertEqual(ip.dst, saved_packet[IP].dst) - self.assertEqual(udp.payload, saved_packet[UDP].payload) - except: - self.logger.error(ppp("Unexpected or invalid packet:", packet)) - raise - for index in self._packet_infos: - self.assertTrue(index in seen or index in dropped_packet_indexes, - "Packet with packet_index %d not received" % index) - - def test_reassembly(self): - """ basic reassembly """ - - self.pg_enable_capture() - self.pg_if.add_stream(self.fragments_200) - self.pg_start() - - packets = self.punt_socket.wait_for_packets(len(self.pkt_infos)) - self.verify_capture(packets) - self.pg_if.assert_nothing_captured() - - # run it all again to verify correctness - self.pg_enable_capture() - self.pg_if.add_stream(self.fragments_200) - self.pg_start() - - packets = self.punt_socket.wait_for_packets(len(self.pkt_infos)) - self.verify_capture(packets) - self.pg_if.assert_nothing_captured() - - def test_reversed(self): - """ reverse order reassembly """ - - fragments = list(self.fragments_200) - fragments.reverse() - - self.pg_enable_capture() - self.pg_if.add_stream(fragments) - self.pg_start() - - packets = self.punt_socket.wait_for_packets(len(self.packet_infos)) - self.verify_capture(packets) - self.pg_if.assert_nothing_captured() - - # run it all again to verify correctness - self.pg_enable_capture() - self.pg_if.add_stream(fragments) - self.pg_start() - - packets = self.punt_socket.wait_for_packets(len(self.packet_infos)) - self.verify_capture(packets) - self.pg_if.assert_nothing_captured() - - def test_random(self): - """ random order reassembly """ - - fragments = list(self.fragments_200) - shuffle(fragments) - - self.pg_enable_capture() - self.pg_if.add_stream(fragments) - self.pg_start() - - packets = self.punt_socket.wait_for_packets(len(self.packet_infos)) - self.verify_capture(packets) - self.pg_if.assert_nothing_captured() - - # run it all again to verify correctness - self.pg_enable_capture() - self.pg_if.add_stream(fragments) - self.pg_start() - - packets = self.punt_socket.wait_for_packets(len(self.packet_infos)) - self.verify_capture(packets) - self.pg_if.assert_nothing_captured() - - def test_duplicates(self): - """ duplicate fragments """ - - fragments = [ - x for (_, frags, _, _) in self.pkt_infos - for x in frags - for _ in range(0, min(2, len(frags))) - ] - - self.pg_enable_capture() - self.pg_if.add_stream(fragments) - self.pg_start() - - packets = self.punt_socket.wait_for_packets(len(self.pkt_infos)) - self.verify_capture(packets) - self.pg_if.assert_nothing_captured() - - def test_overlap1(self): - """ overlapping fragments case #1 """ - - fragments = [] - for _, _, frags_300, frags_200 in self.pkt_infos: - if len(frags_300) == 1: - fragments.extend(frags_300) - else: - for i, j in zip(frags_200, frags_300): - fragments.extend(i) - fragments.extend(j) - - self.pg_enable_capture() - self.pg_if.add_stream(fragments) - self.pg_start() - - packets = self.punt_socket.wait_for_packets(len(self.pkt_infos)) - self.verify_capture(packets) - self.pg_if.assert_nothing_captured() - - # run it all to verify correctness - self.pg_enable_capture() - self.pg_if.add_stream(fragments) - self.pg_start() - - packets = self.punt_socket.wait_for_packets(len(self.pkt_infos)) - self.verify_capture(packets) - self.pg_if.assert_nothing_captured() - - def test_overlap2(self): - """ overlapping fragments case #2 """ - - fragments = [] - for _, _, frags_300, frags_200 in self.pkt_infos: - if len(frags_300) == 1: - fragments.extend(frags_300) - else: - # care must be taken here so that there are no fragments - # received by vpp after reassembly is finished, otherwise - # new reassemblies will be started and packet generator will - # freak out when it detects unfreed buffers - zipped = zip(frags_300, frags_200) - for i, j in zipped[:-1]: - fragments.extend(i) - fragments.extend(j) - fragments.append(zipped[-1][0]) - - self.pg_enable_capture() - self.pg_if.add_stream(fragments) - self.pg_start() - - packets = self.punt_socket.wait_for_packets(len(self.pkt_infos)) - self.verify_capture(packets) - self.pg_if.assert_nothing_captured() - - # run it all to verify correctness - self.pg_enable_capture() - self.pg_if.add_stream(fragments) - self.pg_start() - - packets = self.punt_socket.wait_for_packets(len(self.pkt_infos)) - self.verify_capture(packets) - self.pg_if.assert_nothing_captured() - - def test_timeout_inline(self): - """ timeout (inline) """ - - dropped_packet_indexes = set( - index for (index, frags, _, _) in self.pkt_infos if len(frags) > 1 - ) - - self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000, - expire_walk_interval_ms=10000) - - self.pg_enable_capture() - self.pg_if.add_stream(self.fragments_400) - self.pg_start() - - packets = self.punt_socket.wait_for_packets( - len(self.pkt_infos) - len(dropped_packet_indexes)) - self.verify_capture(packets, dropped_packet_indexes) - self.pg_if.assert_nothing_captured() - - def test_timeout_cleanup(self): - """ timeout (cleanup) """ - - # whole packets + fragmented packets sans last fragment - fragments = [ - x for (_, frags_400, _, _) in self.pkt_infos - for x in frags_400[:-1 if len(frags_400) > 1 else None] - ] - - # last fragments for fragmented packets - fragments2 = [frags_400[-1] - for (_, frags_400, _, _) in self.pkt_infos - if len(frags_400) > 1] - - dropped_packet_indexes = set( - index for (index, frags_400, _, _) in self.pkt_infos - if len(frags_400) > 1) - - self.vapi.ip_reassembly_set(timeout_ms=100, max_reassemblies=1000, - expire_walk_interval_ms=50) - - self.pg_enable_capture() - self.pg_if.add_stream(fragments) - self.pg_start() - - self.sleep(.25, "wait before sending rest of fragments") - - self.pg_if.add_stream(fragments2) - self.pg_start() - self.sleep(.25, "wait for vpp to process packets") - - packets = self.punt_socket.wait_for_packets( - len(self.pkt_infos) - len(dropped_packet_indexes)) - self.verify_capture(packets, dropped_packet_indexes) - self.pg_if.assert_nothing_captured() - - def test_disabled(self): - """ reassembly disabled """ - - dropped_packet_indexes = set( - index for (index, frags_400, _, _) in self.pkt_infos - if len(frags_400) > 1) - - self.vapi.ip_reassembly_set(timeout_ms=1000, max_reassemblies=0, - expire_walk_interval_ms=10000) - - self.pg_enable_capture() - self.pg_if.add_stream(self.fragments_400) - self.pg_start() - - packets = self.punt_socket.wait_for_packets( - len(self.pkt_infos) - len(dropped_packet_indexes)) - self.verify_capture(packets, dropped_packet_indexes) - self.pg_if.assert_nothing_captured() - - -class TestIPv6Reassembly(VppTestCase): - """ IPv6 Reassembly """ - - @classmethod - def setUpClass(cls): - super(TestIPv6Reassembly, cls).setUpClass() - - cls.create_pg_interfaces([0]) - cls.pg_if = cls.pg0 - - # setup all interfaces - for i in cls.pg_interfaces: - i.admin_up() - i.config_ip6() - i.resolve_ndp() - - cls.punt_port = 9999 - cls.punt_socket = VppUDSPuntSocket(cls, cls.punt_port, is_ip4=0) - - # packet sizes - cls.packet_sizes = [64, 512, 1518, 9018] - cls.padding = " abcdefghijklmn" - cls.create_stream(cls.packet_sizes) - cls.create_fragments() - - def setUp(self): - """ Test setup - force timeout on existing reassemblies """ - super(TestIPv6Reassembly, self).setUp() - self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000, - expire_walk_interval_ms=10, is_ip6=1) - self.sleep(.25) - self.vapi.ip_reassembly_set(timeout_ms=1000000, max_reassemblies=1000, - expire_walk_interval_ms=10000, is_ip6=1) - - def tearDown(self): - super(TestIPv6Reassembly, self).tearDown() - self.logger.debug(self.vapi.ppcli("show ip6-reassembly details")) - - @classmethod - def create_stream(cls, packet_sizes, packet_count=test_packet_count): - """Create input packet stream for defined interface. - - :param list packet_sizes: Required packet sizes. - """ - for i in range(0, packet_count): - info = cls.create_packet_info(cls.pg_if, cls.pg_if) - payload = cls.info_to_payload(info) - p = (Ether(dst=cls.pg_if.local_mac, src=cls.pg_if.remote_mac) / - IPv6(src=cls.pg_if.remote_ip6, - dst=cls.pg_if.local_ip6) / - UDP(sport=1234, dport=cls.punt_port) / - Raw(payload)) - size = packet_sizes[(i // 2) % len(packet_sizes)] - cls.extend_packet(p, size, cls.padding) - info.data = p - - @classmethod - def create_fragments(cls): - infos = cls._packet_infos - cls.pkt_infos = [] - for index, info in infos.iteritems(): - p = info.data - # self.logger.debug(ppp("Packet:", p.__class__(str(p)))) - fragments_400 = fragment_rfc8200(p, info.index, 400) - fragments_300 = fragment_rfc8200(p, info.index, 300) - cls.pkt_infos.append((index, fragments_400, fragments_300)) - cls.fragments_400 = [ - x for _, frags, _ in cls.pkt_infos for x in frags] - cls.fragments_300 = [ - x for _, _, frags in cls.pkt_infos for x in frags] - cls.logger.debug("Fragmented %s packets into %s 400-byte fragments, " - "and %s 300-byte fragments" % - (len(infos), len(cls.fragments_400), - len(cls.fragments_300))) - - def verify_capture(self, capture, dropped_packet_indexes=[]): - """Verify captured packet strea . - - :param list capture: Captured packet stream. - """ - info = None - seen = set() - for packet in capture: - try: - sw_if_index = packet['sw_if_index'] - punt_action = packet['punt_action'] - packet = Ether(packet['packet']) - self.logger.debug(ppp("Got packet from %s, action %s" % - (sw_if_index, punt_action), packet)) - ip = packet[IPv6] - udp = packet[UDP] - payload_info = self.payload_to_info(str(packet[Raw])) - packet_index = payload_info.index - self.assertTrue( - packet_index not in dropped_packet_indexes, - ppp("Packet received, but should be dropped:", packet)) - if packet_index in seen: - raise Exception(ppp("Duplicate packet received", packet)) - seen.add(packet_index) - self.assertEqual(payload_info.dst, self.pg_if.sw_if_index) - info = self._packet_infos[packet_index] - self.assertTrue(info is not None) - self.assertEqual(packet_index, info.index) - saved_packet = info.data - self.assertEqual(ip.src, saved_packet[IPv6].src) - self.assertEqual(ip.dst, saved_packet[IPv6].dst) - self.assertEqual(udp.payload, saved_packet[UDP].payload) - except: - self.logger.error(ppp("Unexpected or invalid packet:", packet)) - raise - for index in self._packet_infos: - self.assertTrue(index in seen or index in dropped_packet_indexes, - "Packet with packet_index %d not received" % index) - - def test_reassembly(self): - """ basic reassembly """ - - self.pg_enable_capture() - self.pg_if.add_stream(self.fragments_400) - self.pg_start() - - packets = self.punt_socket.wait_for_packets(len(self.pkt_infos)) - self.verify_capture(packets) - self.pg_if.assert_nothing_captured() - - # run it all again to verify correctness - self.pg_enable_capture() - self.pg_if.add_stream(self.fragments_400) - self.pg_start() - - packets = self.punt_socket.wait_for_packets(len(self.pkt_infos)) - self.verify_capture(packets) - self.pg_if.assert_nothing_captured() - - def test_reversed(self): - """ reverse order reassembly """ - - fragments = list(self.fragments_400) - fragments.reverse() - - self.pg_enable_capture() - self.pg_if.add_stream(fragments) - self.pg_start() - - packets = self.punt_socket.wait_for_packets(len(self.pkt_infos)) - self.verify_capture(packets) - self.pg_if.assert_nothing_captured() - - # run it all again to verify correctness - self.pg_enable_capture() - self.pg_if.add_stream(fragments) - self.pg_start() - - packets = self.punt_socket.wait_for_packets(len(self.pkt_infos)) - self.verify_capture(packets) - self.pg_if.assert_nothing_captured() - - def test_random(self): - """ random order reassembly """ - - fragments = list(self.fragments_400) - shuffle(fragments) - - self.pg_enable_capture() - self.pg_if.add_stream(fragments) - self.pg_start() - - packets = self.punt_socket.wait_for_packets(len(self.pkt_infos)) - self.verify_capture(packets) - self.pg_if.assert_nothing_captured() - - # run it all again to verify correctness - self.pg_enable_capture() - self.pg_if.add_stream(fragments) - self.pg_start() - - packets = self.punt_socket.wait_for_packets(len(self.pkt_infos)) - self.verify_capture(packets) - self.pg_if.assert_nothing_captured() - - def test_duplicates(self): - """ duplicate fragments """ - - fragments = [ - x for (_, frags, _) in self.pkt_infos - for x in frags - for _ in range(0, min(2, len(frags))) - ] - - self.pg_enable_capture() - self.pg_if.add_stream(fragments) - self.pg_start() - - packets = self.punt_socket.wait_for_packets(len(self.pkt_infos)) - self.verify_capture(packets) - self.pg_if.assert_nothing_captured() - - def test_overlap1(self): - """ overlapping fragments case #1 """ - - fragments = [] - for _, frags_400, frags_300 in self.pkt_infos: - if len(frags_300) == 1: - fragments.extend(frags_400) - else: - for i, j in zip(frags_300, frags_400): - fragments.extend(i) - fragments.extend(j) - - dropped_packet_indexes = set( - index for (index, _, frags) in self.pkt_infos if len(frags) > 1 - ) - - self.pg_enable_capture() - self.pg_if.add_stream(fragments) - self.pg_start() - - self.sleep(.1, "wait for vpp to process packets") - packets = self.punt_socket.wait_for_packets( - len(self.pkt_infos) - len(dropped_packet_indexes)) - self.verify_capture(packets, dropped_packet_indexes) - self.pg_if.assert_nothing_captured() - - def test_overlap2(self): - """ overlapping fragments case #2 """ - - fragments = [] - for _, frags_400, frags_30 in self.pkt_infos: - if len(frags_400) == 1: - fragments.extend(frags_400) - else: - # care must be taken here so that there are no fragments - # received by vpp after reassembly is finished, otherwise - # new reassemblies will be started and packet generator will - # freak out when it detects unfreed buffers - zipped = zip(frags_400, frags_30) - for i, j in zipped[:-1]: - fragments.extend(i) - fragments.extend(j) - fragments.append(zipped[-1][0]) - - dropped_packet_indexes = set( - index for (index, _, frags) in self.pkt_infos if len(frags) > 1 - ) - - self.pg_enable_capture() - self.pg_if.add_stream(fragments) - self.pg_start() - - self.sleep(.1, "wait for vpp to process packets") - packets = self.punt_socket.wait_for_packets( - len(self.pkt_infos) - len(dropped_packet_indexes)) - self.verify_capture(packets, dropped_packet_indexes) - self.pg_if.assert_nothing_captured() - - def test_timeout_inline(self): - """ timeout (inline) """ - - dropped_packet_indexes = set( - index for (index, frags, _) in self.pkt_infos if len(frags) > 1 - ) - - self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000, - expire_walk_interval_ms=10000, is_ip6=1) - - self.pg_enable_capture() - self.pg_if.add_stream(self.fragments_400) - self.pg_start() - - packets = self.punt_socket.wait_for_packets( - len(self.pkt_infos) - len(dropped_packet_indexes)) - self.verify_capture(packets, dropped_packet_indexes) - pkts = self.pg_if.get_capture( - expected_count=len(dropped_packet_indexes)) - for icmp in pkts: - self.assertIn(ICMPv6TimeExceeded, icmp) - self.assertIn(IPv6ExtHdrFragment, icmp) - self.assertIn(icmp[IPv6ExtHdrFragment].id, dropped_packet_indexes) - dropped_packet_indexes.remove(icmp[IPv6ExtHdrFragment].id) - - def test_timeout_cleanup(self): - """ timeout (cleanup) """ - - # whole packets + fragmented packets sans last fragment - fragments = [ - x for (_, frags_400, _) in self.pkt_infos - for x in frags_400[:-1 if len(frags_400) > 1 else None] - ] - - # last fragments for fragmented packets - fragments2 = [frags_400[-1] - for (_, frags_400, _) in self.pkt_infos - if len(frags_400) > 1] - - dropped_packet_indexes = set( - index for (index, frags_400, _) in self.pkt_infos - if len(frags_400) > 1) - - self.vapi.ip_reassembly_set(timeout_ms=100, max_reassemblies=1000, - expire_walk_interval_ms=50) - - self.vapi.ip_reassembly_set(timeout_ms=100, max_reassemblies=1000, - expire_walk_interval_ms=50, is_ip6=1) - - self.pg_enable_capture() - self.pg_if.add_stream(fragments) - self.pg_start() - - self.sleep(.25, "wait before sending rest of fragments") - - self.pg_if.add_stream(fragments2) - self.pg_start() - self.sleep(.25, "wait for vpp to process packets") - - packets = self.punt_socket.wait_for_packets( - len(self.pkt_infos) - len(dropped_packet_indexes)) - self.verify_capture(packets, dropped_packet_indexes) - pkts = self.pg_if.get_capture( - expected_count=len(dropped_packet_indexes)) - for icmp in pkts: - self.assertIn(ICMPv6TimeExceeded, icmp) - self.assertIn(IPv6ExtHdrFragment, icmp) - self.assertIn(icmp[IPv6ExtHdrFragment].id, dropped_packet_indexes) - dropped_packet_indexes.remove(icmp[IPv6ExtHdrFragment].id) - - def test_disabled(self): - """ reassembly disabled """ - - dropped_packet_indexes = set( - index for (index, frags_400, _) in self.pkt_infos - if len(frags_400) > 1) - - self.vapi.ip_reassembly_set(timeout_ms=1000, max_reassemblies=0, - expire_walk_interval_ms=10000, is_ip6=1) - - self.pg_enable_capture() - self.pg_if.add_stream(self.fragments_400) - self.pg_start() - - packets = self.punt_socket.wait_for_packets( - len(self.pkt_infos) - len(dropped_packet_indexes)) - self.verify_capture(packets, dropped_packet_indexes) - self.pg_if.assert_nothing_captured() - - def test_missing_upper(self): - """ missing upper layer """ - p = (Ether(dst=self.pg_if.local_mac, src=self.pg_if.remote_mac) / - IPv6(src=self.pg_if.remote_ip6, - dst=self.pg_if.local_ip6) / - UDP(sport=1234, dport=self.punt_port) / - Raw()) - self.extend_packet(p, 1000, self.padding) - fragments = fragment_rfc8200(p, 1, 500) - bad_fragment = p.__class__(str(fragments[1])) - bad_fragment[IPv6ExtHdrFragment].nh = 59 - bad_fragment[IPv6ExtHdrFragment].offset = 0 - self.pg_enable_capture() - self.pg_if.add_stream([bad_fragment]) - self.pg_start() - pkts = self.pg_if.get_capture(expected_count=1) - icmp = pkts[0] - self.assertIn(ICMPv6ParamProblem, icmp) - self.assert_equal(icmp[ICMPv6ParamProblem].code, 3, "ICMP code") - - def test_invalid_frag_size(self): - """ fragment size not a multiple of 8 """ - p = (Ether(dst=self.pg_if.local_mac, src=self.pg_if.remote_mac) / - IPv6(src=self.pg_if.remote_ip6, - dst=self.pg_if.local_ip6) / - UDP(sport=1234, dport=self.punt_port) / - Raw()) - self.extend_packet(p, 1000, self.padding) - fragments = fragment_rfc8200(p, 1, 500) - bad_fragment = fragments[0] - self.extend_packet(bad_fragment, len(bad_fragment) + 5) - self.pg_enable_capture() - self.pg_if.add_stream([bad_fragment]) - self.pg_start() - pkts = self.pg_if.get_capture(expected_count=1) - icmp = pkts[0] - self.assertIn(ICMPv6ParamProblem, icmp) - self.assert_equal(icmp[ICMPv6ParamProblem].code, 0, "ICMP code") - - def test_invalid_packet_size(self): - """ total packet size > 65535 """ - p = (Ether(dst=self.pg_if.local_mac, src=self.pg_if.remote_mac) / - IPv6(src=self.pg_if.remote_ip6, - dst=self.pg_if.local_ip6) / - UDP(sport=1234, dport=self.punt_port) / - Raw()) - self.extend_packet(p, 1000, self.padding) - fragments = fragment_rfc8200(p, 1, 500) - bad_fragment = fragments[1] - bad_fragment[IPv6ExtHdrFragment].offset = 65500 - self.pg_enable_capture() - self.pg_if.add_stream([bad_fragment]) - self.pg_start() - pkts = self.pg_if.get_capture(expected_count=1) - icmp = pkts[0] - self.assertIn(ICMPv6ParamProblem, icmp) - self.assert_equal(icmp[ICMPv6ParamProblem].code, 0, "ICMP code") - - -class TestFIFReassembly(VppTestCase): - """ Fragments in fragments reassembly """ - - @classmethod - def setUpClass(cls): - super(TestFIFReassembly, cls).setUpClass() - - cls.create_pg_interfaces([0]) - cls.pg_if = cls.pg0 - cls.pg_if.admin_up() - cls.pg_if.config_ip4() - cls.pg_if.resolve_arp() - cls.pg_if.config_ip6() - cls.pg_if.resolve_ndp() - - cls.punt_port = 9999 - cls.punt4_socket = VppUDSPuntSocket(cls, cls.punt_port) - cls.punt6_socket = VppUDSPuntSocket(cls, cls.punt_port, is_ip4=0) - cls.packet_sizes = [64, 512, 1518, 9018] - cls.padding = " abcdefghijklmn" - - def setUp(self): - """ Test setup - force timeout on existing reassemblies """ - super(TestFIFReassembly, self).setUp() - self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000, - expire_walk_interval_ms=10) - self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000, - expire_walk_interval_ms=10, is_ip6=1) - self.sleep(.25) - self.vapi.ip_reassembly_set(timeout_ms=1000000, max_reassemblies=1000, - expire_walk_interval_ms=10000) - self.vapi.ip_reassembly_set(timeout_ms=1000000, max_reassemblies=1000, - expire_walk_interval_ms=10000, is_ip6=1) - - def tearDown(self): - self.logger.debug(self.vapi.ppcli("show ip4-reassembly details")) - self.logger.debug(self.vapi.ppcli("show ip6-reassembly details")) - super(TestFIFReassembly, self).tearDown() - - def verify_capture(self, capture, ip_class, dropped_packet_indexes=[]): - """Verify captured packet stream. - - :param list capture: Captured packet stream. - """ - info = None - seen = set() - for packet in capture: - try: - sw_if_index = packet['sw_if_index'] - punt_action = packet['punt_action'] - packet = Ether(packet['packet']) - self.logger.debug(ppp("Got packet from %s, action %s" % - (sw_if_index, punt_action), packet)) - ip = packet[ip_class] - udp = packet[UDP] - payload_info = self.payload_to_info(str(packet[Raw])) - packet_index = payload_info.index - self.assertTrue( - packet_index not in dropped_packet_indexes, - ppp("Packet received, but should be dropped:", packet)) - if packet_index in seen: - raise Exception(ppp("Duplicate packet received", packet)) - seen.add(packet_index) - self.assertEqual(payload_info.dst, self.pg_if.sw_if_index) - info = self._packet_infos[packet_index] - self.assertTrue(info is not None) - self.assertEqual(packet_index, info.index) - saved_packet = info.data - self.assertEqual(ip.src, saved_packet[ip_class].src) - self.assertEqual(ip.dst, saved_packet[ip_class].dst) - self.assertEqual(udp.payload, saved_packet[UDP].payload) - except: - self.logger.error(ppp("Unexpected or invalid packet:", packet)) - raise - for index in self._packet_infos: - self.assertTrue(index in seen or index in dropped_packet_indexes, - "Packet with packet_index %d not received" % index) - - def test_fif4(self): - """ Fragments in fragments (4o4) """ - - # TODO this should be ideally in setUpClass, but then we hit a bug - # with VppIpRoute incorrectly reporting it's present when it's not - # so we need to manually remove the vpp config, thus we cannot have - # it shared for multiple test cases - self.tun_ip4 = "1.1.1.2" - - self.gre4 = VppGreInterface(self, self.pg0.local_ip4, self.tun_ip4) - self.gre4.add_vpp_config() - self.gre4.admin_up() - self.gre4.config_ip4() - - self.route4 = VppIpRoute(self, self.tun_ip4, 32, - [VppRoutePath(self.pg0.remote_ip4, - self.pg0.sw_if_index)]) - self.route4.add_vpp_config() - - self.reset_packet_infos() - for i in range(test_packet_count): - info = self.create_packet_info(self.pg0, self.pg0) - payload = self.info_to_payload(info) - p = (IP(id=i, src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) / - UDP(sport=1234, dport=self.punt_port) / - Raw(payload)) - size = self.packet_sizes[(i // 2) % len(self.packet_sizes)] - self.extend_packet(p, size, self.padding) - info.data = p - - fragments = [x for _, p in self._packet_infos.iteritems() - for x in fragment_rfc791(p.data, 400)] - - encapped_fragments = \ - [Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / - IP(src=self.tun_ip4, dst=self.pg0.local_ip4) / - GRE() / - p - for p in fragments] - - fragmented_encapped_fragments = \ - [x for p in encapped_fragments - for x in fragment_rfc791(p, 200)] - - self.pg0.add_stream(fragmented_encapped_fragments) - - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - - self.pg0.assert_nothing_captured() - packets = self.punt4_socket.wait_for_packets(len(self._packet_infos)) - self.verify_capture(packets, IP) - - # TODO remove gre vpp config by hand until VppIpRoute gets fixed - # so that it's query_vpp_config() works as it should - self.gre4.remove_vpp_config() - - def test_fif6(self): - """ Fragments in fragments (6o6) """ - # TODO this should be ideally in setUpClass, but then we hit a bug - # with VppIpRoute incorrectly reporting it's present when it's not - # so we need to manually remove the vpp config, thus we cannot have - # it shared for multiple test cases - self.tun_ip6 = "1002::1" - - self.gre6 = VppGre6Interface(self, self.pg0.local_ip6, self.tun_ip6) - self.gre6.add_vpp_config() - self.gre6.admin_up() - self.gre6.config_ip6() - - self.route6 = VppIpRoute(self, self.tun_ip6, 128, - [VppRoutePath(self.pg0.remote_ip6, - self.pg0.sw_if_index, - proto=DpoProto.DPO_PROTO_IP6)], - is_ip6=1) - self.route6.add_vpp_config() - - self.reset_packet_infos() - for i in range(test_packet_count): - info = self.create_packet_info(self.pg0, self.pg0) - payload = self.info_to_payload(info) - p = (IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6) / - UDP(sport=1234, dport=self.punt_port) / - Raw(payload)) - size = self.packet_sizes[(i // 2) % len(self.packet_sizes)] - self.extend_packet(p, size, self.padding) - info.data = p - - fragments = [x for _, i in self._packet_infos.iteritems() - for x in fragment_rfc8200( - i.data, i.index, 400)] - - encapped_fragments = \ - [Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / - IPv6(src=self.tun_ip6, dst=self.pg0.local_ip6) / - GRE() / - p - for p in fragments] - - fragmented_encapped_fragments = \ - [x for p in encapped_fragments for x in ( - fragment_rfc8200( - p, - 2 * len(self._packet_infos) + p[IPv6ExtHdrFragment].id, - 200) - if IPv6ExtHdrFragment in p else [p] - ) - ] - - self.pg0.add_stream(fragmented_encapped_fragments) - - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - - self.pg0.assert_nothing_captured() - packets = self.punt6_socket.wait_for_packets(len(self._packet_infos)) - self.verify_capture(packets, IPv6) - - # TODO remove gre vpp config by hand until VppIpRoute gets fixed - # so that it's query_vpp_config() works as it should - self.gre6.remove_vpp_config() - - -if __name__ == '__main__': - unittest.main(testRunner=VppTestRunner) diff --git a/test/vpp_papi_provider.py b/test/vpp_papi_provider.py index dd553cb473e..52fc3641672 100644 --- a/test/vpp_papi_provider.py +++ b/test/vpp_papi_provider.py @@ -3203,6 +3203,15 @@ class VppPapiProvider(object): """ Get IP reassembly parameters """ return self.api(self.papi.ip_reassembly_get, {'is_ip6': is_ip6}) + def ip_reassembly_enable_disable(self, sw_if_index, enable_ip4=False, + enable_ip6=False): + """ Enable/disable IP reassembly """ + return self.api(self.papi.ip_reassembly_enable_disable, + {'sw_if_index': sw_if_index, + 'enable_ip4': 1 if enable_ip4 else 0, + 'enable_ip6': 1 if enable_ip6 else 0, + }) + def gbp_endpoint_add_del(self, is_add, sw_if_index, addr, is_ip6, epg): """ GBP endpoint Add/Del """ return self.api(self.papi.gbp_endpoint_add_del, diff --git a/test/vpp_punt_socket.py b/test/vpp_punt_socket.py deleted file mode 100644 index 5004b23396d..00000000000 --- a/test/vpp_punt_socket.py +++ /dev/null @@ -1,75 +0,0 @@ -from socket import socket, AF_UNIX, SOCK_DGRAM -from select import select -from time import time -from struct import unpack, calcsize -from util import ppc -from scapy.layers.l2 import Ether - -client_uds_socket_name = "client-uds-socket" -vpp_uds_socket_name = "vpp-uds-socket" - -VPP_PUNT_HEADER_FMT = '=Ii' -VPP_PUNT_HEADER_SIZE = calcsize(VPP_PUNT_HEADER_FMT) - - -class VppPuntAction: - PUNT_L2 = 0 - PUNT_IP4_ROUTED = 1 - PUNT_IP6_ROUTED = 2 - - -class VppUDSPuntSocket(object): - def __init__(self, testcase, port, is_ip4=1, l4_protocol=0x11): - client_path = '%s/%s-%s-%s' % (testcase.tempdir, - client_uds_socket_name, - "4" if is_ip4 else "6", port) - testcase.vapi.punt_socket_register( - port, client_path, is_ip4=is_ip4, l4_protocol=l4_protocol) - self.testcase = testcase - self.uds = socket(AF_UNIX, SOCK_DGRAM) - self.uds.bind(client_path) - self.uds.connect(testcase.punt_socket_path) - - def wait_for_packets(self, count, timeout=1): - packets = [] - now = time() - deadline = now + timeout - while len(packets) < count and now < deadline: - r, w, e = select([self.uds], [], [self.uds], deadline - now) - if self.uds in r: - x = self.uds.recv(1024 * 1024) - sw_if_index, punt_action = unpack( - VPP_PUNT_HEADER_FMT, x[:VPP_PUNT_HEADER_SIZE]) - packets.append({'sw_if_index': sw_if_index, - 'punt_action': punt_action, - 'packet': x[VPP_PUNT_HEADER_SIZE:]}) - - if self.uds in e: - raise Exception("select() indicates error on UDS socket") - now = time() - - if len(packets) != count: - raise Exception("Unexpected packet count received, got %s packets," - " expected %s packets" % (len(packets), count)) - self.testcase.logger.debug( - "Got %s packets via punt socket" % len(packets)) - return packets - - def assert_nothing_captured(self, timeout=.25): - packets = [] - now = time() - deadline = now + timeout - while now < deadline: - r, w, e = select([self.uds], [], [self.uds], deadline - now) - if self.uds in r: - x = self.uds.recv(1024 * 1024) - packets.append(Ether(x[VPP_PUNT_HEADER_SIZE:])) - if self.uds in e: - raise Exception("select() indicates error on UDS socket") - now = time() - - if len(packets) > 0: - self.testcase.logger.error( - ppc("Unexpected packets captured:", packets)) - raise Exception("Unexpected packet count received, got %s packets," - " expected no packets" % len(packets)) -- cgit 1.2.3-korg