import unittest
import socket
+from framework import tag_fixme_vpp_workers
from framework import VppTestCase, VppTestRunner
from vpp_ip import DpoProto, INVALID_INDEX
from vpp_ip_route import VppIpRoute, VppRoutePath, VppMplsRoute, \
VppMplsIpBind, VppIpMRoute, VppMRoutePath, \
- MRouteItfFlags, MRouteEntryFlags, VppIpTable, VppMplsTable, \
+ VppIpTable, VppMplsTable, \
VppMplsLabel, MplsLspMode, find_mpls_route, \
FibPathProto, FibPathType, FibPathFlags, VppMplsLabel, MplsLspMode
from vpp_mpls_tunnel_interface import VppMPLSTunnelInterface
+from vpp_papi import VppEnum
import scapy.compat
from scapy.packet import Raw
-from scapy.layers.l2 import Ether
+from scapy.layers.l2 import Ether, ARP
from scapy.layers.inet import IP, UDP, ICMP
from scapy.layers.inet6 import IPv6, ICMPv6TimeExceeded
from scapy.contrib.mpls import MPLS
NUM_PKTS = 67
+# scapy removed these attributes.
+# we asked that they be restored: https://github.com/secdev/scapy/pull/1878
+# semantic names have more meaning than numbers. so here they are.
+ARP.who_has = 1
+ARP.is_at = 2
+
def verify_filter(capture, sent):
if not len(capture) == len(sent):
rx_mpls = rx_mpls[MPLS].payload
+@tag_fixme_vpp_workers
class TestMPLS(VppTestCase):
""" MPLS Test Case """
for i in self.pg_interfaces:
i.unconfig_ip4()
i.unconfig_ip6()
- i.ip6_disable()
i.set_table_ip4(0)
i.set_table_ip6(0)
i.disable_mpls()
def test_mpls_tunnel_many(self):
""" MPLS Multiple Tunnels """
- for ii in range(10):
+ for ii in range(100):
mpls_tun = VppMPLSTunnelInterface(
self,
[VppRoutePath(self.pg0.remote_ip4,
VppMplsLabel(46, MplsLspMode.UNIFORM)])])
mpls_tun.add_vpp_config()
mpls_tun.admin_up()
+ for ii in range(100):
+ mpls_tun = VppMPLSTunnelInterface(
+ self,
+ [VppRoutePath(self.pg0.remote_ip4,
+ self.pg0.sw_if_index,
+ labels=[VppMplsLabel(44, ttl=32),
+ VppMplsLabel(46, MplsLspMode.UNIFORM)])],
+ is_l2=1)
+ mpls_tun.add_vpp_config()
+ mpls_tun.admin_up()
def test_v4_exp_null(self):
""" MPLS V4 Explicit NULL test """
def test_mcast_head(self):
""" MPLS Multicast Head-end """
+ MRouteItfFlags = VppEnum.vl_api_mfib_itf_flags_t
+ MRouteEntryFlags = VppEnum.vl_api_mfib_entry_flags_t
+
#
# Create a multicast tunnel with two replications
#
self,
"0.0.0.0",
"232.1.1.1", 32,
- 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),
VppMRoutePath(mpls_tun._sw_if_index,
- MRouteItfFlags.MFIB_ITF_FLAG_FORWARD)])
+ MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD)])
route_232_1_1_1.add_vpp_config()
self.logger.info(self.vapi.cli("sh ip mfib index 0"))
def test_mcast_ip4_tail(self):
""" MPLS IPv4 Multicast Tail """
+ MRouteItfFlags = VppEnum.vl_api_mfib_itf_flags_t
+ MRouteEntryFlags = VppEnum.vl_api_mfib_entry_flags_t
+
#
# Add a multicast route that will forward the traffic
# post-disposition
self,
"0.0.0.0",
"232.1.1.1", 32,
- MRouteEntryFlags.MFIB_ENTRY_FLAG_NONE,
+ MRouteEntryFlags.MFIB_API_ENTRY_FLAG_NONE,
table_id=1,
paths=[VppMRoutePath(self.pg1.sw_if_index,
- MRouteItfFlags.MFIB_ITF_FLAG_FORWARD)])
+ MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD)])
route_232_1_1_1.add_vpp_config()
#
def test_mcast_ip6_tail(self):
""" MPLS IPv6 Multicast Tail """
+ MRouteItfFlags = VppEnum.vl_api_mfib_itf_flags_t
+ MRouteEntryFlags = VppEnum.vl_api_mfib_entry_flags_t
+
#
# Add a multicast route that will forward the traffic
# post-disposition
self,
"::",
"ff01::1", 32,
- MRouteEntryFlags.MFIB_ENTRY_FLAG_NONE,
+ MRouteEntryFlags.MFIB_API_ENTRY_FLAG_NONE,
table_id=1,
paths=[VppMRoutePath(self.pg1.sw_if_index,
- MRouteItfFlags.MFIB_ITF_FLAG_FORWARD,
+ MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD,
proto=FibPathProto.FIB_PATH_NH_PROTO_IP6)])
route_ff.add_vpp_config()
VppMplsLabel(32),
VppMplsLabel(99)])
+ def test_attached(self):
+ """ Attach Routes with Local Label """
+
+ #
+ # test that if a local label is associated with an attached/connected
+ # prefix, that we can reach hosts in the prefix.
+ #
+ binding = VppMplsIpBind(self, 44,
+ self.pg0._local_ip4_subnet,
+ self.pg0.local_ip4_prefix_len)
+ binding.add_vpp_config()
+
+ tx = (Ether(src=self.pg1.remote_mac,
+ dst=self.pg1.local_mac) /
+ MPLS(label=44, ttl=64) /
+ IP(src=self.pg0.remote_ip4, dst=self.pg0.remote_ip4) /
+ UDP(sport=1234, dport=1234) /
+ Raw(b'\xa5' * 100))
+ rxs = self.send_and_expect(self.pg0, [tx], self.pg0)
+ for rx in rxs:
+ # if there's an ARP then the label is linked to the glean
+ # which is wrong.
+ self.assertFalse(rx.haslayer(ARP))
+ # it should be unicasted to the host
+ self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
+ self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
+
class TestMPLSDisabled(VppTestCase):
""" MPLS disabled """
def test_mpls_disabled(self):
""" MPLS Disabled """
+ self.logger.info(self.vapi.cli("show mpls interface"))
+ self.logger.info(self.vapi.cli("show mpls interface pg1"))
+ self.logger.info(self.vapi.cli("show mpls interface pg0"))
+
tx = (Ether(src=self.pg1.remote_mac,
dst=self.pg1.local_mac) /
MPLS(label=32, ttl=64) /
#
self.pg1.enable_mpls()
+ self.logger.info(self.vapi.cli("show mpls interface"))
+ self.logger.info(self.vapi.cli("show mpls interface pg1"))
+
#
# Now we get packets through
#
#
# put the connected routes back
#
+ self.logger.info(self.vapi.cli("sh log"))
self.pg2.admin_up()
self.pg2.config_ip6()
self.pg2.resolve_ndp()
tbl.add_vpp_config()
self.tables.append(tbl)
- # use pg0 as the core facing interface
+ # use pg0 as the core facing interface, don't resolve ARP
self.pg0.admin_up()
self.pg0.config_ip4()
- self.pg0.resolve_arp()
self.pg0.enable_mpls()
# use the other 2 for customer facing L2 links
self.assertEqual(rx_eth.src, tx_eth.src)
self.assertEqual(rx_eth.dst, tx_eth.dst)
+ def verify_arp_req(self, rx, smac, sip, dip):
+ ether = rx[Ether]
+ self.assertEqual(ether.dst, "ff:ff:ff:ff:ff:ff")
+ self.assertEqual(ether.src, smac)
+
+ arp = rx[ARP]
+ self.assertEqual(arp.hwtype, 1)
+ self.assertEqual(arp.ptype, 0x800)
+ self.assertEqual(arp.hwlen, 6)
+ self.assertEqual(arp.plen, 4)
+ self.assertEqual(arp.op, ARP.who_has)
+ self.assertEqual(arp.hwsrc, smac)
+ self.assertEqual(arp.hwdst, "00:00:00:00:00:00")
+ self.assertEqual(arp.psrc, sip)
+ self.assertEqual(arp.pdst, dip)
+
def test_vpws(self):
""" Virtual Private Wire Service """
#
# Inject a packet from the customer/L2 side
+ # there's no resolved ARP entry so the first packet we see should be
+ # an ARP request
#
+ tx1 = pcore[MPLS].payload
+ rx1 = self.send_and_expect(self.pg1, [tx1], self.pg0)
+
+ self.verify_arp_req(rx1[0],
+ self.pg0.local_mac,
+ self.pg0.local_ip4,
+ self.pg0.remote_ip4)
+
+ #
+ # resolve the ARP entries and send again
+ #
+ self.pg0.resolve_arp()
tx1 = pcore[MPLS].payload * NUM_PKTS
rx1 = self.send_and_expect(self.pg1, tx1, self.pg0)
def test_vpls(self):
""" Virtual Private LAN Service """
+
+ # we skipped this in the setup
+ self.pg0.resolve_arp()
+
#
# Create a L2 MPLS tunnels
#
Code Review / vpp.git / blobdiff