#!/usr/bin/env python3
+import socket
from socket import inet_pton, inet_ntop
import unittest
from parameterized import parameterized
import scapy.compat
import scapy.layers.inet6 as inet6
+from scapy.layers.inet import UDP
from scapy.contrib.mpls import MPLS
from scapy.layers.inet6 import IPv6, ICMPv6ND_NS, ICMPv6ND_RS, \
ICMPv6ND_RA, ICMPv6NDOptMTU, ICMPv6NDOptSrcLLAddr, ICMPv6NDOptPrefixInfo, \
in6_mactoifaceid
from six import moves
-from framework import VppTestCase, VppTestRunner
+from framework import VppTestCase, VppTestRunner, tag_run_solo
from util import ppp, ip6_normalize, mk_ll_addr
-from vpp_ip import DpoProto
+from vpp_papi import VppEnum
+from vpp_ip import DpoProto, VppIpPuntPolicer, VppIpPuntRedirect, VppIpPathMtu
from vpp_ip_route import VppIpRoute, VppRoutePath, find_route, VppIpMRoute, \
- VppMRoutePath, MRouteItfFlags, MRouteEntryFlags, VppMplsIpBind, \
+ VppMRoutePath, VppMplsIpBind, \
VppMplsRoute, VppMplsTable, VppIpTable, FibPathType, FibPathProto, \
- VppIpInterfaceAddress, find_route_in_dump, find_mroute_in_dump
+ VppIpInterfaceAddress, find_route_in_dump, find_mroute_in_dump, \
+ VppIp6LinkLocalAddress, VppIpRouteV2
from vpp_neighbor import find_nbr, VppNeighbor
+from vpp_ipip_tun_interface import VppIpIpTunInterface
from vpp_pg_interface import is_ipv6_misc
from vpp_sub_interface import VppSubInterface, VppDot1QSubint
-from vpp_policer import VppPolicer
+from vpp_policer import VppPolicer, PolicerAction
from ipaddress import IPv6Network, IPv6Address
AF_INET6 = socket.AF_INET6
self.assertEqual(ip.dst, dip)
+@tag_run_solo
class TestIPv6(TestIPv6ND):
""" IPv6 Test Case """
subitf.admin_up()
subitf.config_ip6()
- rxs = self.pg1._get_capture(timeout=2, filter_out_fn=None)
+ rxs = self.pg1._get_capture(timeout=4, filter_out_fn=None)
#
# hunt for the MLD on vlan 99
self.assertEqual(mld.records_number, 4)
+class TestIPv6RouteLookup(VppTestCase):
+ """ IPv6 Route Lookup Test Case """
+ routes = []
+
+ def route_lookup(self, prefix, exact):
+ return self.vapi.api(self.vapi.papi.ip_route_lookup,
+ {
+ 'table_id': 0,
+ 'exact': exact,
+ 'prefix': prefix,
+ })
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestIPv6RouteLookup, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIPv6RouteLookup, cls).tearDownClass()
+
+ def setUp(self):
+ super(TestIPv6RouteLookup, self).setUp()
+
+ drop_nh = VppRoutePath("::1", 0xffffffff,
+ type=FibPathType.FIB_PATH_TYPE_DROP)
+
+ # Add 3 routes
+ r = VppIpRoute(self, "2001:1111::", 32, [drop_nh])
+ r.add_vpp_config()
+ self.routes.append(r)
+
+ r = VppIpRoute(self, "2001:1111:2222::", 48, [drop_nh])
+ r.add_vpp_config()
+ self.routes.append(r)
+
+ r = VppIpRoute(self, "2001:1111:2222::1", 128, [drop_nh])
+ r.add_vpp_config()
+ self.routes.append(r)
+
+ def tearDown(self):
+ # Remove the routes we added
+ for r in self.routes:
+ r.remove_vpp_config()
+
+ super(TestIPv6RouteLookup, self).tearDown()
+
+ def test_exact_match(self):
+ # Verify we find the host route
+ prefix = "2001:1111:2222::1/128"
+ result = self.route_lookup(prefix, True)
+ assert (prefix == str(result.route.prefix))
+
+ # Verify we find a middle prefix route
+ prefix = "2001:1111:2222::/48"
+ result = self.route_lookup(prefix, True)
+ assert (prefix == str(result.route.prefix))
+
+ # Verify we do not find an available LPM.
+ with self.vapi.assert_negative_api_retval():
+ self.route_lookup("2001::2/128", True)
+
+ def test_longest_prefix_match(self):
+ # verify we find lpm
+ lpm_prefix = "2001:1111:2222::/48"
+ result = self.route_lookup("2001:1111:2222::2/128", False)
+ assert (lpm_prefix == str(result.route.prefix))
+
+ # Verify we find the exact when not requested
+ result = self.route_lookup(lpm_prefix, False)
+ assert (lpm_prefix == str(result.route.prefix))
+
+ # Can't seem to delete the default route so no negative LPM test.
+
+
class TestIPv6IfAddrRoute(VppTestCase):
""" IPv6 Interface Addr Route Test Case """
self.assertFalse(find_route(self, addr1, 128))
self.assertFalse(find_route(self, addr2, 128))
+ def test_ipv6_ifaddr_del(self):
+ """ Delete an interface address that does not exist """
+
+ loopbacks = self.create_loopback_interfaces(1)
+ lo = self.lo_interfaces[0]
+
+ lo.config_ip6()
+ lo.admin_up()
+
+ #
+ # try and remove pg0's subnet from lo
+ #
+ with self.vapi.assert_negative_api_retval():
+ self.vapi.sw_interface_add_del_address(
+ sw_if_index=lo.sw_if_index,
+ prefix=self.pg0.local_ip6_prefix,
+ is_add=0)
+
class TestICMPv6Echo(VppTestCase):
""" ICMPv6 Echo Test Case """
for i in self.pg_interfaces:
i.admin_up()
i.config_ip6()
+ i.resolve_ndp(link_layer=True)
i.resolve_ndp()
def tearDown(self):
- Check outgoing ICMPv6 Echo Reply message on pg0 interface.
"""
- icmpv6_id = 0xb
- icmpv6_seq = 5
- icmpv6_data = b'\x0a' * 18
- p_echo_request = (Ether(src=self.pg0.remote_mac,
- dst=self.pg0.local_mac) /
- IPv6(src=self.pg0.remote_ip6,
- dst=self.pg0.local_ip6) /
- ICMPv6EchoRequest(
- id=icmpv6_id,
- seq=icmpv6_seq,
- data=icmpv6_data))
-
- self.pg0.add_stream(p_echo_request)
+ # test both with global and local ipv6 addresses
+ dsts = (self.pg0.local_ip6, self.pg0.local_ip6_ll)
+ id = 0xb
+ seq = 5
+ data = b'\x0a' * 18
+ p = list()
+ for dst in dsts:
+ p.append((Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
+ IPv6(src=self.pg0.remote_ip6, dst=dst) /
+ ICMPv6EchoRequest(id=id, seq=seq, data=data)))
+
+ self.pg0.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
-
- rx = self.pg0.get_capture(1)
- rx = rx[0]
- ether = rx[Ether]
- ipv6 = rx[IPv6]
- icmpv6 = rx[ICMPv6EchoReply]
-
- self.assertEqual(ether.src, self.pg0.local_mac)
- self.assertEqual(ether.dst, self.pg0.remote_mac)
-
- self.assertEqual(ipv6.src, self.pg0.local_ip6)
- self.assertEqual(ipv6.dst, self.pg0.remote_ip6)
-
- self.assertEqual(
- icmp6types[icmpv6.type], "Echo Reply")
- self.assertEqual(icmpv6.id, icmpv6_id)
- self.assertEqual(icmpv6.seq, icmpv6_seq)
- self.assertEqual(icmpv6.data, icmpv6_data)
+ rxs = self.pg0.get_capture(len(dsts))
+
+ for rx, dst in zip(rxs, dsts):
+ ether = rx[Ether]
+ ipv6 = rx[IPv6]
+ icmpv6 = rx[ICMPv6EchoReply]
+ self.assertEqual(ether.src, self.pg0.local_mac)
+ self.assertEqual(ether.dst, self.pg0.remote_mac)
+ self.assertEqual(ipv6.src, dst)
+ self.assertEqual(ipv6.dst, self.pg0.remote_ip6)
+ self.assertEqual(icmp6types[icmpv6.type], "Echo Reply")
+ self.assertEqual(icmpv6.id, id)
+ self.assertEqual(icmpv6.seq, seq)
+ self.assertEqual(icmpv6.data, data)
class TestIPv6RD(TestIPv6ND):
def test_ip_disabled(self):
""" IP Disabled """
+ MRouteItfFlags = VppEnum.vl_api_mfib_itf_flags_t
+ MRouteEntryFlags = VppEnum.vl_api_mfib_entry_flags_t
#
# An (S,G).
# one accepting interface, pg0, 2 forwarding interfaces
self,
"::",
"ffef::1", 128,
- MRouteEntryFlags.MFIB_ENTRY_FLAG_NONE,
+ MRouteEntryFlags.MFIB_API_ENTRY_FLAG_NONE,
[VppMRoutePath(self.pg1.sw_if_index,
- MRouteItfFlags.MFIB_ITF_FLAG_ACCEPT),
+ MRouteItfFlags.MFIB_API_ITF_FLAG_ACCEPT),
VppMRoutePath(self.pg0.sw_if_index,
- MRouteItfFlags.MFIB_ITF_FLAG_FORWARD)])
+ MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD)])
route_ff_01.add_vpp_config()
pu = (Ether(src=self.pg1.remote_mac,
def send_and_expect_load_balancing(self, input, pkts, outputs):
self.pg_send(input, pkts)
+ rxs = []
for oo in outputs:
rx = oo._get_capture(1)
self.assertNotEqual(0, len(rx))
+ rxs.append(rx)
+ return rxs
def send_and_expect_one_itf(self, input, pkts, itf):
self.pg_send(input, pkts)
# be guaranteed. But with 64 different packets we do expect some
# balancing. So instead just ensure there is traffic on each link.
#
- self.send_and_expect_load_balancing(self.pg0, port_ip_pkts,
- [self.pg1, self.pg2])
+ rx = self.send_and_expect_load_balancing(self.pg0, port_ip_pkts,
+ [self.pg1, self.pg2])
+ n_ip_pg0 = len(rx[0])
self.send_and_expect_load_balancing(self.pg0, src_ip_pkts,
[self.pg1, self.pg2])
self.send_and_expect_load_balancing(self.pg0, port_mpls_pkts,
[self.pg1, self.pg2])
self.send_and_expect_load_balancing(self.pg0, src_mpls_pkts,
[self.pg1, self.pg2])
- self.send_and_expect_load_balancing(self.pg0, port_mpls_neos_pkts,
- [self.pg1, self.pg2])
+ rx = self.send_and_expect_load_balancing(self.pg0, port_mpls_neos_pkts,
+ [self.pg1, self.pg2])
+ n_mpls_pg0 = len(rx[0])
+
+ #
+ # change the router ID and expect the distribution changes
+ #
+ self.vapi.set_ip_flow_hash_router_id(router_id=0x11111111)
+
+ rx = self.send_and_expect_load_balancing(self.pg0, port_ip_pkts,
+ [self.pg1, self.pg2])
+ self.assertNotEqual(n_ip_pg0, len(rx[0]))
+
+ rx = self.send_and_expect_load_balancing(self.pg0, src_mpls_pkts,
+ [self.pg1, self.pg2])
+ self.assertNotEqual(n_mpls_pg0, len(rx[0]))
#
# The packets with Entropy label in should not load-balance,
# - now only the stream with differing source address will
# load-balance
#
- self.vapi.set_ip_flow_hash(vrf_id=0, src=1, dst=1, sport=0, dport=0,
- is_ipv6=1)
+ self.vapi.set_ip_flow_hash(vrf_id=0, src=1, dst=1, proto=1,
+ sport=0, dport=0, is_ipv6=1)
self.send_and_expect_load_balancing(self.pg0, src_ip_pkts,
[self.pg1, self.pg2])
# change the flow hash config back to defaults
#
self.vapi.set_ip_flow_hash(vrf_id=0, src=1, dst=1, sport=1, dport=1,
- is_ipv6=1)
+ proto=1, is_ipv6=1)
#
# Recursive prefixes
self.send_and_expect_one_itf(self.pg0, port_pkts, self.pg3)
-class TestIP6Punt(VppTestCase):
- """ IPv6 Punt Police/Redirect """
-
- @classmethod
- def setUpClass(cls):
- super(TestIP6Punt, cls).setUpClass()
-
- @classmethod
- def tearDownClass(cls):
- super(TestIP6Punt, cls).tearDownClass()
-
- def setUp(self):
- super(TestIP6Punt, self).setUp()
+class IP6PuntSetup(object):
+ """ Setup for IPv6 Punt Police/Redirect """
+ def punt_setup(self):
self.create_pg_interfaces(range(4))
for i in self.pg_interfaces:
i.config_ip6()
i.resolve_ndp()
- def tearDown(self):
- super(TestIP6Punt, self).tearDown()
+ self.pkt = (Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IPv6(src=self.pg0.remote_ip6,
+ dst=self.pg0.local_ip6) /
+ inet6.TCP(sport=1234, dport=1234) /
+ Raw(b'\xa5' * 100))
+
+ def punt_teardown(self):
for i in self.pg_interfaces:
i.unconfig_ip6()
i.admin_down()
+
+class TestIP6Punt(IP6PuntSetup, VppTestCase):
+ """ IPv6 Punt Police/Redirect """
+
+ def setUp(self):
+ super(TestIP6Punt, self).setUp()
+ super(TestIP6Punt, self).punt_setup()
+
+ def tearDown(self):
+ super(TestIP6Punt, self).punt_teardown()
+ super(TestIP6Punt, self).tearDown()
+
def test_ip_punt(self):
""" IP6 punt police and redirect """
- p = (Ether(src=self.pg0.remote_mac,
- dst=self.pg0.local_mac) /
- IPv6(src=self.pg0.remote_ip6,
- dst=self.pg0.local_ip6) /
- inet6.TCP(sport=1234, dport=1234) /
- Raw(b'\xa5' * 100))
-
- pkts = p * 1025
+ pkts = self.pkt * 1025
#
# Configure a punt redirect via pg1.
#
nh_addr = self.pg1.remote_ip6
- self.vapi.ip_punt_redirect(self.pg0.sw_if_index,
- self.pg1.sw_if_index,
- nh_addr)
+ ip_punt_redirect = VppIpPuntRedirect(self, self.pg0.sw_if_index,
+ self.pg1.sw_if_index, nh_addr)
+ ip_punt_redirect.add_vpp_config()
self.send_and_expect(self.pg0, pkts, self.pg1)
#
policer = VppPolicer(self, "ip6-punt", 400, 0, 10, 0, rate_type=1)
policer.add_vpp_config()
- self.vapi.ip_punt_police(policer.policer_index, is_ip6=1)
+ ip_punt_policer = VppIpPuntPolicer(self, policer.policer_index,
+ is_ip6=True)
+ ip_punt_policer.add_vpp_config()
self.vapi.cli("clear trace")
self.pg0.add_stream(pkts)
# but not equal to the number sent, since some were policed
#
rx = self.pg1._get_capture(1)
+ stats = policer.get_stats()
+
+ # Single rate policer - expect conform, violate but no exceed
+ self.assertGreater(stats['conform_packets'], 0)
+ self.assertEqual(stats['exceed_packets'], 0)
+ self.assertGreater(stats['violate_packets'], 0)
+
self.assertGreater(len(rx), 0)
self.assertLess(len(rx), len(pkts))
#
# remove the policer. back to full rx
#
- self.vapi.ip_punt_police(policer.policer_index, is_add=0, is_ip6=1)
+ ip_punt_policer.remove_vpp_config()
policer.remove_vpp_config()
self.send_and_expect(self.pg0, pkts, self.pg1)
#
# remove the redirect. expect full drop.
#
- self.vapi.ip_punt_redirect(self.pg0.sw_if_index,
- self.pg1.sw_if_index,
- nh_addr,
- is_add=0)
+ ip_punt_redirect.remove_vpp_config()
self.send_and_assert_no_replies(self.pg0, pkts,
"IP no punt config")
#
# Add a redirect that is not input port selective
#
- self.vapi.ip_punt_redirect(0xffffffff,
- self.pg1.sw_if_index,
- nh_addr)
+ ip_punt_redirect = VppIpPuntRedirect(self, 0xffffffff,
+ self.pg1.sw_if_index, nh_addr)
+ ip_punt_redirect.add_vpp_config()
self.send_and_expect(self.pg0, pkts, self.pg1)
-
- self.vapi.ip_punt_redirect(0xffffffff,
- self.pg1.sw_if_index,
- nh_addr,
- is_add=0)
+ ip_punt_redirect.remove_vpp_config()
def test_ip_punt_dump(self):
""" IP6 punt redirect dump"""
#
# Configure a punt redirects
#
- nh_addr = self.pg3.remote_ip6
- self.vapi.ip_punt_redirect(self.pg0.sw_if_index,
- self.pg3.sw_if_index,
- nh_addr)
- self.vapi.ip_punt_redirect(self.pg1.sw_if_index,
- self.pg3.sw_if_index,
- nh_addr)
- self.vapi.ip_punt_redirect(self.pg2.sw_if_index,
- self.pg3.sw_if_index,
- '0::0')
+ nh_address = self.pg3.remote_ip6
+ ipr_03 = VppIpPuntRedirect(self, self.pg0.sw_if_index,
+ self.pg3.sw_if_index, nh_address)
+ ipr_13 = VppIpPuntRedirect(self, self.pg1.sw_if_index,
+ self.pg3.sw_if_index, nh_address)
+ ipr_23 = VppIpPuntRedirect(self, self.pg2.sw_if_index,
+ self.pg3.sw_if_index, '0::0')
+ ipr_03.add_vpp_config()
+ ipr_13.add_vpp_config()
+ ipr_23.add_vpp_config()
#
# Dump pg0 punt redirects
#
- punts = self.vapi.ip_punt_redirect_dump(self.pg0.sw_if_index,
- is_ipv6=1)
- for p in punts:
- self.assertEqual(p.punt.rx_sw_if_index, self.pg0.sw_if_index)
+ self.assertTrue(ipr_03.query_vpp_config())
+ self.assertTrue(ipr_13.query_vpp_config())
+ self.assertTrue(ipr_23.query_vpp_config())
#
# Dump punt redirects for all interfaces
self.assertEqual(str(punts[2].punt.nh), '::')
+class TestIP6PuntHandoff(IP6PuntSetup, VppTestCase):
+ """ IPv6 Punt Police/Redirect """
+ worker_config = "workers 2"
+
+ def setUp(self):
+ super(TestIP6PuntHandoff, self).setUp()
+ super(TestIP6PuntHandoff, self).punt_setup()
+
+ def tearDown(self):
+ super(TestIP6PuntHandoff, self).punt_teardown()
+ super(TestIP6PuntHandoff, self).tearDown()
+
+ def test_ip_punt(self):
+ """ IP6 punt policer thread handoff """
+ pkts = self.pkt * NUM_PKTS
+
+ #
+ # Configure a punt redirect via pg1.
+ #
+ nh_addr = self.pg1.remote_ip6
+ ip_punt_redirect = VppIpPuntRedirect(self, self.pg0.sw_if_index,
+ self.pg1.sw_if_index, nh_addr)
+ ip_punt_redirect.add_vpp_config()
+
+ action_tx = PolicerAction(
+ VppEnum.vl_api_sse2_qos_action_type_t.SSE2_QOS_ACTION_API_TRANSMIT,
+ 0)
+ #
+ # This policer drops no packets, we are just
+ # testing that they get to the right thread.
+ #
+ policer = VppPolicer(self, "ip6-punt", 400, 0, 10, 0, 1,
+ 0, 0, False, action_tx, action_tx, action_tx)
+ policer.add_vpp_config()
+ ip_punt_policer = VppIpPuntPolicer(self, policer.policer_index,
+ is_ip6=True)
+ ip_punt_policer.add_vpp_config()
+
+ for worker in [0, 1]:
+ self.send_and_expect(self.pg0, pkts, self.pg1, worker=worker)
+ if worker == 0:
+ self.logger.debug(self.vapi.cli("show trace max 100"))
+
+ # Combined stats, all threads
+ stats = policer.get_stats()
+
+ # Single rate policer - expect conform, violate but no exceed
+ self.assertGreater(stats['conform_packets'], 0)
+ self.assertEqual(stats['exceed_packets'], 0)
+ self.assertGreater(stats['violate_packets'], 0)
+
+ # Worker 0, should have done all the policing
+ stats0 = policer.get_stats(worker=0)
+ self.assertEqual(stats, stats0)
+
+ # Worker 1, should have handed everything off
+ stats1 = policer.get_stats(worker=1)
+ self.assertEqual(stats1['conform_packets'], 0)
+ self.assertEqual(stats1['exceed_packets'], 0)
+ self.assertEqual(stats1['violate_packets'], 0)
+
+ # Bind the policer to worker 1 and repeat
+ policer.bind_vpp_config(1, True)
+ for worker in [0, 1]:
+ self.send_and_expect(self.pg0, pkts, self.pg1, worker=worker)
+ self.logger.debug(self.vapi.cli("show trace max 100"))
+
+ # The 2 workers should now have policed the same amount
+ stats = policer.get_stats()
+ stats0 = policer.get_stats(worker=0)
+ stats1 = policer.get_stats(worker=1)
+
+ self.assertGreater(stats0['conform_packets'], 0)
+ self.assertEqual(stats0['exceed_packets'], 0)
+ self.assertGreater(stats0['violate_packets'], 0)
+
+ self.assertGreater(stats1['conform_packets'], 0)
+ self.assertEqual(stats1['exceed_packets'], 0)
+ self.assertGreater(stats1['violate_packets'], 0)
+
+ self.assertEqual(stats0['conform_packets'] + stats1['conform_packets'],
+ stats['conform_packets'])
+
+ self.assertEqual(stats0['violate_packets'] + stats1['violate_packets'],
+ stats['violate_packets'])
+
+ # Unbind the policer and repeat
+ policer.bind_vpp_config(1, False)
+ for worker in [0, 1]:
+ self.send_and_expect(self.pg0, pkts, self.pg1, worker=worker)
+ self.logger.debug(self.vapi.cli("show trace max 100"))
+
+ # The policer should auto-bind to worker 0 when packets arrive
+ stats = policer.get_stats()
+ stats0new = policer.get_stats(worker=0)
+ stats1new = policer.get_stats(worker=1)
+
+ self.assertGreater(stats0new['conform_packets'],
+ stats0['conform_packets'])
+ self.assertEqual(stats0new['exceed_packets'], 0)
+ self.assertGreater(stats0new['violate_packets'],
+ stats0['violate_packets'])
+
+ self.assertEqual(stats1, stats1new)
+
+ #
+ # Clean up
+ #
+ ip_punt_policer.remove_vpp_config()
+ policer.remove_vpp_config()
+ ip_punt_redirect.remove_vpp_config()
+
+
class TestIPDeag(VppTestCase):
""" IPv6 Deaggregate Routes """
for i in self.pg_interfaces:
i.admin_up()
i.config_ip6()
- i.resolve_arp()
i.generate_remote_hosts(2)
self.tables.append(VppIpTable(self, table_id,
True).add_vpp_config())
super(TestIPReplace, self).tearDown()
for i in self.pg_interfaces:
i.admin_down()
- i.unconfig_ip4()
+ i.unconfig_ip6()
def test_replace(self):
""" IP Table Replace """
+ MRouteItfFlags = VppEnum.vl_api_mfib_itf_flags_t
+ MRouteEntryFlags = VppEnum.vl_api_mfib_entry_flags_t
N_ROUTES = 20
links = [self.pg0, self.pg1, self.pg2, self.pg3]
routes = [[], [], [], []]
multi = VppIpMRoute(
self, "::",
"ff:2001::%d" % jj, 128,
- MRouteEntryFlags.MFIB_ENTRY_FLAG_NONE,
+ MRouteEntryFlags.MFIB_API_ENTRY_FLAG_NONE,
[VppMRoutePath(self.pg0.sw_if_index,
- MRouteItfFlags.MFIB_ITF_FLAG_ACCEPT,
+ MRouteItfFlags.MFIB_API_ITF_FLAG_ACCEPT,
proto=FibPathProto.FIB_PATH_NH_PROTO_IP6),
VppMRoutePath(self.pg1.sw_if_index,
- MRouteItfFlags.MFIB_ITF_FLAG_FORWARD,
+ MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD,
proto=FibPathProto.FIB_PATH_NH_PROTO_IP6),
VppMRoutePath(self.pg2.sw_if_index,
- MRouteItfFlags.MFIB_ITF_FLAG_FORWARD,
+ MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD,
proto=FibPathProto.FIB_PATH_NH_PROTO_IP6),
VppMRoutePath(self.pg3.sw_if_index,
- MRouteItfFlags.MFIB_ITF_FLAG_FORWARD,
+ MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD,
proto=FibPathProto.FIB_PATH_NH_PROTO_IP6)],
table_id=t.table_id).add_vpp_config()
routes[ii].append({'uni': uni,
self.assertEqual(len(t.mdump()), 5)
+class TestIP6Replace(VppTestCase):
+ """ IPv4 Interface Address Replace """
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestIP6Replace, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIP6Replace, cls).tearDownClass()
+
+ def setUp(self):
+ super(TestIP6Replace, self).setUp()
+
+ self.create_pg_interfaces(range(4))
+
+ for i in self.pg_interfaces:
+ i.admin_up()
+
+ def tearDown(self):
+ super(TestIP6Replace, self).tearDown()
+ for i in self.pg_interfaces:
+ i.admin_down()
+
+ def get_n_pfxs(self, intf):
+ return len(self.vapi.ip_address_dump(intf.sw_if_index, True))
+
+ def test_replace(self):
+ """ IP interface address replace """
+
+ intf_pfxs = [[], [], [], []]
+
+ # add prefixes to each of the interfaces
+ for i in range(len(self.pg_interfaces)):
+ intf = self.pg_interfaces[i]
+
+ # 2001:16:x::1/64
+ addr = "2001:16:%d::1" % intf.sw_if_index
+ a = VppIpInterfaceAddress(self, intf, addr, 64).add_vpp_config()
+ intf_pfxs[i].append(a)
+
+ # 2001:16:x::2/64 - a different address in the same subnet as above
+ addr = "2001:16:%d::2" % intf.sw_if_index
+ a = VppIpInterfaceAddress(self, intf, addr, 64).add_vpp_config()
+ intf_pfxs[i].append(a)
+
+ # 2001:15:x::2/64 - a different address and subnet
+ addr = "2001:15:%d::2" % intf.sw_if_index
+ a = VppIpInterfaceAddress(self, intf, addr, 64).add_vpp_config()
+ intf_pfxs[i].append(a)
+
+ # a dump should n_address in it
+ for intf in self.pg_interfaces:
+ self.assertEqual(self.get_n_pfxs(intf), 3)
+
+ #
+ # remove all the address thru a replace
+ #
+ self.vapi.sw_interface_address_replace_begin()
+ self.vapi.sw_interface_address_replace_end()
+ for intf in self.pg_interfaces:
+ self.assertEqual(self.get_n_pfxs(intf), 0)
+
+ #
+ # add all the interface addresses back
+ #
+ for p in intf_pfxs:
+ for v in p:
+ v.add_vpp_config()
+ for intf in self.pg_interfaces:
+ self.assertEqual(self.get_n_pfxs(intf), 3)
+
+ #
+ # replace again, but this time update/re-add the address on the first
+ # two interfaces
+ #
+ self.vapi.sw_interface_address_replace_begin()
+
+ for p in intf_pfxs[:2]:
+ for v in p:
+ v.add_vpp_config()
+
+ self.vapi.sw_interface_address_replace_end()
+
+ # on the first two the address still exist,
+ # on the other two they do not
+ for intf in self.pg_interfaces[:2]:
+ self.assertEqual(self.get_n_pfxs(intf), 3)
+ for p in intf_pfxs[:2]:
+ for v in p:
+ self.assertTrue(v.query_vpp_config())
+ for intf in self.pg_interfaces[2:]:
+ self.assertEqual(self.get_n_pfxs(intf), 0)
+
+ #
+ # add all the interface addresses back on the last two
+ #
+ for p in intf_pfxs[2:]:
+ for v in p:
+ v.add_vpp_config()
+ for intf in self.pg_interfaces:
+ self.assertEqual(self.get_n_pfxs(intf), 3)
+
+ #
+ # replace again, this time add different prefixes on all the interfaces
+ #
+ self.vapi.sw_interface_address_replace_begin()
+
+ pfxs = []
+ for intf in self.pg_interfaces:
+ # 2001:18:x::1/64
+ addr = "2001:18:%d::1" % intf.sw_if_index
+ pfxs.append(VppIpInterfaceAddress(self, intf, addr,
+ 64).add_vpp_config())
+
+ self.vapi.sw_interface_address_replace_end()
+
+ # only .18 should exist on each interface
+ for intf in self.pg_interfaces:
+ self.assertEqual(self.get_n_pfxs(intf), 1)
+ for pfx in pfxs:
+ self.assertTrue(pfx.query_vpp_config())
+
+ #
+ # remove everything
+ #
+ self.vapi.sw_interface_address_replace_begin()
+ self.vapi.sw_interface_address_replace_end()
+ for intf in self.pg_interfaces:
+ self.assertEqual(self.get_n_pfxs(intf), 0)
+
+ #
+ # add prefixes to each interface. post-begin add the prefix from
+ # interface X onto interface Y. this would normally be an error
+ # since it would generate a 'duplicate address' warning. but in
+ # this case, since what is newly downloaded is sane, it's ok
+ #
+ for intf in self.pg_interfaces:
+ # 2001:18:x::1/64
+ addr = "2001:18:%d::1" % intf.sw_if_index
+ VppIpInterfaceAddress(self, intf, addr, 64).add_vpp_config()
+
+ self.vapi.sw_interface_address_replace_begin()
+
+ pfxs = []
+ for intf in self.pg_interfaces:
+ # 2001:18:x::1/64
+ addr = "2001:18:%d::1" % (intf.sw_if_index + 1)
+ pfxs.append(VppIpInterfaceAddress(self, intf,
+ addr, 64).add_vpp_config())
+
+ self.vapi.sw_interface_address_replace_end()
+
+ self.logger.info(self.vapi.cli("sh int addr"))
+
+ for intf in self.pg_interfaces:
+ self.assertEqual(self.get_n_pfxs(intf), 1)
+ for pfx in pfxs:
+ self.assertTrue(pfx.query_vpp_config())
+
+
+class TestIP6LinkLocal(VppTestCase):
+ """ IPv6 Link Local """
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestIP6LinkLocal, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIP6LinkLocal, cls).tearDownClass()
+
+ def setUp(self):
+ super(TestIP6LinkLocal, self).setUp()
+
+ self.create_pg_interfaces(range(2))
+
+ for i in self.pg_interfaces:
+ i.admin_up()
+
+ def tearDown(self):
+ super(TestIP6LinkLocal, self).tearDown()
+ for i in self.pg_interfaces:
+ i.admin_down()
+
+ def test_ip6_ll(self):
+ """ IPv6 Link Local """
+
+ #
+ # two APIs to add a link local address.
+ # 1 - just like any other prefix
+ # 2 - with the special set LL API
+ #
+
+ #
+ # First with the API to set a 'normal' prefix
+ #
+ ll1 = "fe80:1::1"
+ ll2 = "fe80:2::2"
+ ll3 = "fe80:3::3"
+
+ VppIpInterfaceAddress(self, self.pg0, ll1, 128).add_vpp_config()
+
+ #
+ # should be able to ping the ll
+ #
+ p_echo_request_1 = (Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IPv6(src=ll2,
+ dst=ll1) /
+ ICMPv6EchoRequest())
+
+ self.send_and_expect(self.pg0, [p_echo_request_1], self.pg0)
+
+ #
+ # change the link-local on pg0
+ #
+ v_ll3 = VppIpInterfaceAddress(self, self.pg0,
+ ll3, 128).add_vpp_config()
+
+ p_echo_request_3 = (Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IPv6(src=ll2,
+ dst=ll3) /
+ ICMPv6EchoRequest())
+
+ self.send_and_expect(self.pg0, [p_echo_request_3], self.pg0)
+
+ #
+ # set a normal v6 prefix on the link
+ #
+ self.pg0.config_ip6()
+
+ self.send_and_expect(self.pg0, [p_echo_request_3], self.pg0)
+
+ # the link-local cannot be removed
+ with self.vapi.assert_negative_api_retval():
+ v_ll3.remove_vpp_config()
+
+ #
+ # Use the specific link-local API on pg1
+ #
+ VppIp6LinkLocalAddress(self, self.pg1, ll1).add_vpp_config()
+ self.send_and_expect(self.pg1, [p_echo_request_1], self.pg1)
+
+ VppIp6LinkLocalAddress(self, self.pg1, ll3).add_vpp_config()
+ self.send_and_expect(self.pg1, [p_echo_request_3], self.pg1)
+
+
+class TestIPv6PathMTU(VppTestCase):
+ """ IPv6 Path MTU """
+
+ def setUp(self):
+ super(TestIPv6PathMTU, self).setUp()
+
+ self.create_pg_interfaces(range(2))
+
+ # setup all interfaces
+ for i in self.pg_interfaces:
+ i.admin_up()
+ i.config_ip6()
+ i.resolve_ndp()
+
+ def tearDown(self):
+ super(TestIPv6PathMTU, self).tearDown()
+ for i in self.pg_interfaces:
+ i.unconfig_ip6()
+ i.admin_down()
+
+ def test_path_mtu_local(self):
+ """ Path MTU for attached neighbour """
+
+ self.vapi.cli("set log class ip level debug")
+ #
+ # The goal here is not test that fragmentation works correctly,
+ # that's done elsewhere, the intent is to ensure that the Path MTU
+ # settings are honoured.
+ #
+
+ #
+ # IPv6 will only frag locally generated packets, so use tunnelled
+ # packets post encap
+ #
+ tun = VppIpIpTunInterface(
+ self,
+ self.pg1,
+ self.pg1.local_ip6,
+ self.pg1.remote_ip6)
+ tun.add_vpp_config()
+ tun.admin_up()
+ tun.config_ip6()
+
+ # set the interface MTU to a reasonable value
+ self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
+ [2800, 0, 0, 0])
+
+ p_2k = (Ether(dst=self.pg0.local_mac,
+ src=self.pg0.remote_mac) /
+ IPv6(src=self.pg0.remote_ip6,
+ dst=tun.remote_ip6) /
+ UDP(sport=1234, dport=5678) /
+ Raw(b'0xa' * 1000))
+ p_1k = (Ether(dst=self.pg0.local_mac,
+ src=self.pg0.remote_mac) /
+ IPv6(src=self.pg0.remote_ip6,
+ dst=tun.remote_ip6) /
+ UDP(sport=1234, dport=5678) /
+ Raw(b'0xa' * 600))
+
+ nbr = VppNeighbor(self,
+ self.pg1.sw_if_index,
+ self.pg1.remote_mac,
+ self.pg1.remote_ip6).add_vpp_config()
+
+ # this is now the interface MTU frags
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1)
+
+ # drop the path MTU for this neighbour to below the interface MTU
+ # expect more frags
+ pmtu = VppIpPathMtu(self, self.pg1.remote_ip6, 1300).add_vpp_config()
+
+ # print/format the adj delegate and trackers
+ self.logger.info(self.vapi.cli("sh ip pmtu"))
+ self.logger.info(self.vapi.cli("sh adj 7"))
+
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # increase the path MTU to more than the interface
+ # expect to use the interface MTU
+ pmtu.modify(8192)
+
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1)
+
+ # go back to an MTU from the path
+ pmtu.modify(1300)
+
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # raise the interface's MTU
+ # should still use that of the path
+ self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
+ [2000, 0, 0, 0])
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # set path high and interface low
+ pmtu.modify(2000)
+ self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
+ [1300, 0, 0, 0])
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # remove the path MTU
+ self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
+ [2800, 0, 0, 0])
+ pmtu.modify(0)
+
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1)
+
+ def test_path_mtu_remote(self):
+ """ Path MTU for remote neighbour """
+
+ self.vapi.cli("set log class ip level debug")
+ #
+ # The goal here is not test that fragmentation works correctly,
+ # that's done elsewhere, the intent is to ensure that the Path MTU
+ # settings are honoured.
+ #
+ tun_dst = "2001::1"
+
+ route = VppIpRoute(
+ self, tun_dst, 64,
+ [VppRoutePath(self.pg1.remote_ip6,
+ self.pg1.sw_if_index)]).add_vpp_config()
+
+ #
+ # IPv6 will only frag locally generated packets, so use tunnelled
+ # packets post encap
+ #
+ tun = VppIpIpTunInterface(
+ self,
+ self.pg1,
+ self.pg1.local_ip6,
+ tun_dst)
+ tun.add_vpp_config()
+ tun.admin_up()
+ tun.config_ip6()
+
+ # set the interface MTU to a reasonable value
+ self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
+ [2800, 0, 0, 0])
+
+ p_2k = (Ether(dst=self.pg0.local_mac,
+ src=self.pg0.remote_mac) /
+ IPv6(src=self.pg0.remote_ip6,
+ dst=tun.remote_ip6) /
+ UDP(sport=1234, dport=5678) /
+ Raw(b'0xa' * 1000))
+ p_1k = (Ether(dst=self.pg0.local_mac,
+ src=self.pg0.remote_mac) /
+ IPv6(src=self.pg0.remote_ip6,
+ dst=tun.remote_ip6) /
+ UDP(sport=1234, dport=5678) /
+ Raw(b'0xa' * 600))
+
+ nbr = VppNeighbor(self,
+ self.pg1.sw_if_index,
+ self.pg1.remote_mac,
+ self.pg1.remote_ip6).add_vpp_config()
+
+ # this is now the interface MTU frags
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1)
+
+ # drop the path MTU for this neighbour to below the interface MTU
+ # expect more frags
+ pmtu = VppIpPathMtu(self, tun_dst, 1300).add_vpp_config()
+
+ # print/format the fib entry/dpo
+ self.logger.info(self.vapi.cli("sh ip6 fib 2001::1"))
+
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # increase the path MTU to more than the interface
+ # expect to use the interface MTU
+ pmtu.modify(8192)
+
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1)
+
+ # go back to an MTU from the path
+ pmtu.modify(1300)
+
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # raise the interface's MTU
+ # should still use that of the path
+ self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
+ [2000, 0, 0, 0])
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # turn the tun_dst into an attached neighbour
+ route.modify([VppRoutePath("::",
+ self.pg1.sw_if_index)])
+ nbr2 = VppNeighbor(self,
+ self.pg1.sw_if_index,
+ self.pg1.remote_mac,
+ tun_dst).add_vpp_config()
+
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # add back to not attached
+ nbr2.remove_vpp_config()
+ route.modify([VppRoutePath(self.pg1.remote_ip6,
+ self.pg1.sw_if_index)])
+
+ # set path high and interface low
+ pmtu.modify(2000)
+ self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
+ [1300, 0, 0, 0])
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # remove the path MTU
+ self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
+ [2800, 0, 0, 0])
+ pmtu.remove_vpp_config()
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1)
+
+
+class TestIPFibSource(VppTestCase):
+ """ IPv6 Table FibSource """
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestIPFibSource, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIPFibSource, cls).tearDownClass()
+
+ def setUp(self):
+ super(TestIPFibSource, self).setUp()
+
+ self.create_pg_interfaces(range(2))
+
+ for i in self.pg_interfaces:
+ i.admin_up()
+ i.config_ip6()
+ i.resolve_arp()
+ i.generate_remote_hosts(2)
+ i.configure_ipv6_neighbors()
+
+ def tearDown(self):
+ super(TestIPFibSource, self).tearDown()
+ for i in self.pg_interfaces:
+ i.admin_down()
+ i.unconfig_ip4()
+
+ def test_fib_source(self):
+ """ IP Table FibSource """
+
+ routes = self.vapi.ip_route_v2_dump(0, True)
+
+ # 2 interfaces (4 routes) + 2 specials + 4 neighbours = 10 routes
+ self.assertEqual(len(routes), 10)
+
+ # dump all the sources in the FIB
+ sources = self.vapi.fib_source_dump()
+ for source in sources:
+ if (source.src.name == "API"):
+ api_source = source.src
+ if (source.src.name == "interface"):
+ intf_source = source.src
+ if (source.src.name == "adjacency"):
+ adj_source = source.src
+ if (source.src.name == "special"):
+ special_source = source.src
+ if (source.src.name == "default-route"):
+ dr_source = source.src
+
+ # dump the individual route types
+ routes = self.vapi.ip_route_v2_dump(0, True, src=adj_source.id)
+ self.assertEqual(len(routes), 4)
+ routes = self.vapi.ip_route_v2_dump(0, True, src=intf_source.id)
+ self.assertEqual(len(routes), 4)
+ routes = self.vapi.ip_route_v2_dump(0, True, src=special_source.id)
+ self.assertEqual(len(routes), 1)
+ routes = self.vapi.ip_route_v2_dump(0, True, src=dr_source.id)
+ self.assertEqual(len(routes), 1)
+
+ # add a new soure that'a better than the API
+ self.vapi.fib_source_add(src={'name': "bgp",
+ "priority": api_source.priority - 1})
+
+ # dump all the sources to check our new one is there
+ sources = self.vapi.fib_source_dump()
+
+ for source in sources:
+ if (source.src.name == "bgp"):
+ bgp_source = source.src
+
+ self.assertTrue(bgp_source)
+ self.assertEqual(bgp_source.priority,
+ api_source.priority - 1)
+
+ # add a route with the default API source
+ r1 = VppIpRouteV2(
+ self, "2001::1", 128,
+ [VppRoutePath(self.pg0.remote_ip6,
+ self.pg0.sw_if_index)]).add_vpp_config()
+
+ r2 = VppIpRouteV2(self, "2001::1", 128,
+ [VppRoutePath(self.pg1.remote_ip6,
+ self.pg1.sw_if_index)],
+ src=bgp_source.id).add_vpp_config()
+
+ # ensure the BGP source takes priority
+ p = (Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IPv6(src=self.pg0.remote_ip6, dst="2001::1") /
+ inet6.UDP(sport=1234, dport=1234) /
+ Raw(b'\xa5' * 100))
+
+ self.send_and_expect(self.pg0, [p], self.pg1)
+
+ r2.remove_vpp_config()
+ r1.remove_vpp_config()
+
+ self.assertFalse(find_route(self, "2001::1", 128))
+
+
if __name__ == '__main__':
unittest.main(testRunner=VppTestRunner)
Code Review / vpp.git / blobdiff