#!/usr/bin/env python
import unittest
-from logging import *
-
-from framework import VppTestCase, VppTestRunner
-from vpp_sub_interface import VppDot1QSubint
-from vpp_gre_interface import VppGreInterface, VppGre6Interface
-from vpp_ip import DpoProto
-from vpp_ip_route import VppIpRoute, VppRoutePath, VppIpTable
-from vpp_papi_provider import L2_VTR_OP
+import scapy.compat
from scapy.packet import Raw
from scapy.layers.l2 import Ether, Dot1Q, GRE
from scapy.layers.inet import IP, UDP
from scapy.layers.inet6 import IPv6
from scapy.volatile import RandMAC, RandIP
+from framework import VppTestCase, VppTestRunner
+from vpp_sub_interface import L2_VTR_OP, VppDot1QSubint
+from vpp_gre_interface import VppGreInterface
+from vpp_ip import DpoProto
+from vpp_ip_route import VppIpRoute, VppRoutePath, VppIpTable, FibPathProto
from util import ppp, ppc
+from vpp_papi import VppEnum
+
+class TestGREInputNodes(VppTestCase):
+ """ GRE Input Nodes Test Case """
+
+ def setUp(self):
+ super(TestGREInputNodes, self).setUp()
+
+ # create 3 pg interfaces - set one in a non-default table.
+ self.create_pg_interfaces(range(1))
+
+ for i in self.pg_interfaces:
+ i.admin_up()
+ i.config_ip4()
+
+ def tearDown(self):
+ for i in self.pg_interfaces:
+ i.unconfig_ip4()
+ i.admin_down()
+ super(TestGREInputNodes, self).tearDown()
+
+ def test_gre_input_node(self):
+ """ GRE gre input nodes not registerd unless configured """
+ pkt = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) /
+ GRE())
+
+ self.pg0.add_stream(pkt)
+ self.pg_start()
+ # no tunnel created, gre-input not registered
+ err = self.statistics.get_counter(
+ '/err/ip4-input/unknown ip protocol')[0]
+ self.assertEqual(err, 1)
+ err_count = err
+
+ # create gre tunnel
+ gre_if = VppGreInterface(self, self.pg0.local_ip4, "1.1.1.2")
+ gre_if.add_vpp_config()
-class GreTunnelTypes:
- TT_L3 = 0
- TT_TEB = 1
- TT_ERSPAN = 2
+ self.pg0.add_stream(pkt)
+ self.pg_start()
+ # tunnel created, gre-input registered
+ err = self.statistics.get_counter(
+ '/err/ip4-input/unknown ip protocol')[0]
+ # expect no new errors
+ self.assertEqual(err, err_count)
class TestGRE(VppTestCase):
def setUpClass(cls):
super(TestGRE, cls).setUpClass()
+ @classmethod
+ def tearDownClass(cls):
+ super(TestGRE, cls).tearDownClass()
+
def setUp(self):
super(TestGRE, self).setUp()
GRE() /
Ether(dst=RandMAC('*:*:*:*:*:*'),
src=RandMAC('*:*:*:*:*:*')) /
- IP(src=str(RandIP()), dst=str(RandIP())) /
+ IP(src=scapy.compat.raw(RandIP()),
+ dst=scapy.compat.raw(RandIP())) /
UDP(sport=1234, dport=1234) /
Raw(payload))
info.data = p.copy()
Ether(dst=RandMAC('*:*:*:*:*:*'),
src=RandMAC('*:*:*:*:*:*')) /
Dot1Q(vlan=vlan) /
- IP(src=str(RandIP()), dst=str(RandIP())) /
+ IP(src=scapy.compat.raw(RandIP()),
+ dst=scapy.compat.raw(RandIP())) /
UDP(sport=1234, dport=1234) /
Raw(payload))
info.data = p.copy()
self.assertEqual(rx_ip.src, tunnel_src)
self.assertEqual(rx_ip.dst, tunnel_dst)
- rx_gre = GRE(str(rx_ip[IPv6].payload))
+ rx_gre = GRE(scapy.compat.raw(rx_ip[IPv6].payload))
rx_ip = rx_gre[IPv6]
self.assertEqual(rx_ip.src, tx_ip.src)
self.assertEqual(rx_ip.src, tunnel_src)
self.assertEqual(rx_ip.dst, tunnel_dst)
- rx_gre = GRE(str(rx_ip[IPv6].payload))
+ rx_gre = GRE(scapy.compat.raw(rx_ip[IPv6].payload))
tx_ip = tx[IP]
rx_ip = rx_gre[IP]
self.assertEqual(rx_ip.src, tunnel_src)
self.assertEqual(rx_ip.dst, tunnel_dst)
- rx_gre = GRE(str(rx_ip[IP].payload))
+ rx_gre = GRE(scapy.compat.raw(rx_ip[IP].payload))
rx_ip = rx_gre[IPv6]
tx_ip = tx[IPv6]
#
# Send a packet stream that is routed into the tunnel
- # - they are all dropped since the tunnel's desintation IP
+ # - they are all dropped since the tunnel's destintation IP
# is unresolved - or resolves via the default route - which
# which is a drop.
#
self, "2001::1", 128,
[VppRoutePath("::",
gre_if.sw_if_index,
- proto=DpoProto.DPO_PROTO_IP6)],
- is_ip6=1)
+ proto=DpoProto.DPO_PROTO_IP6)])
route6_via_tun.add_vpp_config()
tx = self.create_stream_ip6(self.pg0, "2001::2", "2001::1")
# - assign an IP Address
# - Add a route via the tunnel
#
- gre_if = VppGre6Interface(self,
- self.pg2.local_ip6,
- "1002::1")
+ gre_if = VppGreInterface(self,
+ self.pg2.local_ip6,
+ "1002::1")
gre_if.add_vpp_config()
gre_if.admin_up()
gre_if.config_ip6()
- route_via_tun = VppIpRoute(
- self, "4004::1", 128,
- [VppRoutePath("0::0",
- gre_if.sw_if_index,
- proto=DpoProto.DPO_PROTO_IP6)],
- is_ip6=1)
+ route_via_tun = VppIpRoute(self, "4004::1", 128,
+ [VppRoutePath("0::0",
+ gre_if.sw_if_index)])
route_via_tun.add_vpp_config()
#
# Send a packet stream that is routed into the tunnel
- # - they are all dropped since the tunnel's desintation IP
+ # - they are all dropped since the tunnel's destintation IP
# is unresolved - or resolves via the default route - which
# which is a drop.
#
#
# Add a route that resolves the tunnel's destination
#
- route_tun_dst = VppIpRoute(
- self, "1002::1", 128,
- [VppRoutePath(self.pg2.remote_ip6,
- self.pg2.sw_if_index,
- proto=DpoProto.DPO_PROTO_IP6)],
- is_ip6=1)
+ route_tun_dst = VppIpRoute(self, "1002::1", 128,
+ [VppRoutePath(self.pg2.remote_ip6,
+ self.pg2.sw_if_index)])
route_tun_dst.add_vpp_config()
#
#
gre_if1 = VppGreInterface(self, self.pg0.local_ip4,
"2.2.2.2",
- type=GreTunnelTypes.TT_TEB)
+ type=(VppEnum.vl_api_gre_tunnel_type_t.
+ GRE_API_TUNNEL_TYPE_TEB))
gre_if2 = VppGreInterface(self, self.pg0.local_ip4,
"2.2.2.3",
- type=GreTunnelTypes.TT_TEB)
+ type=(VppEnum.vl_api_gre_tunnel_type_t.
+ GRE_API_TUNNEL_TYPE_TEB))
gre_if1.add_vpp_config()
gre_if2.add_vpp_config()
# Configure both to pop thier respective VLAN tags,
# so that during the x-coonect they will subsequently push
#
- self.vapi.sw_interface_set_l2_tag_rewrite(gre_if_12.sw_if_index,
- L2_VTR_OP.L2_POP_1,
- 12)
- self.vapi.sw_interface_set_l2_tag_rewrite(gre_if_11.sw_if_index,
- L2_VTR_OP.L2_POP_1,
- 11)
+ self.vapi.l2_interface_vlan_tag_rewrite(
+ sw_if_index=gre_if_12.sw_if_index, vtr_op=L2_VTR_OP.L2_POP_1,
+ push_dot1q=12)
+ self.vapi.l2_interface_vlan_tag_rewrite(
+ sw_if_index=gre_if_11.sw_if_index, vtr_op=L2_VTR_OP.L2_POP_1,
+ push_dot1q=11)
#
# Send traffic in both directiond - expect the VLAN tags to
route_tun1_dst.add_vpp_config()
route_tun2_dst.add_vpp_config()
+ def test_gre_loop(self):
+ """ GRE tunnel loop Tests """
+
+ #
+ # Create an L3 GRE tunnel.
+ # - set it admin up
+ # - assign an IP Addres
+ #
+ gre_if = VppGreInterface(self,
+ self.pg0.local_ip4,
+ "1.1.1.2")
+ gre_if.add_vpp_config()
+ gre_if.admin_up()
+ gre_if.config_ip4()
+
+ #
+ # add a route to the tunnel's destination that points
+ # through the tunnel, hence forming a loop in the forwarding
+ # graph
+ #
+ route_dst = VppIpRoute(self, "1.1.1.2", 32,
+ [VppRoutePath("0.0.0.0",
+ gre_if.sw_if_index)])
+ route_dst.add_vpp_config()
+
+ #
+ # packets to the tunnels destination should be dropped
+ #
+ tx = self.create_stream_ip4(self.pg0, "1.1.1.1", "1.1.1.2")
+ self.send_and_assert_no_replies(self.pg2, tx)
+
+ self.logger.info(self.vapi.ppcli("sh adj 7"))
+
+ #
+ # break the loop
+ #
+ route_dst.modify([VppRoutePath(self.pg1.remote_ip4,
+ self.pg1.sw_if_index)])
+ route_dst.add_vpp_config()
+
+ rx = self.send_and_expect(self.pg0, tx, self.pg1)
+
+ #
+ # a good route throught the tunnel to check it restacked
+ #
+ route_via_tun_2 = VppIpRoute(self, "2.2.2.2", 32,
+ [VppRoutePath("0.0.0.0",
+ gre_if.sw_if_index)])
+ route_via_tun_2.add_vpp_config()
+
+ tx = self.create_stream_ip4(self.pg0, "2.2.2.3", "2.2.2.2")
+ rx = self.send_and_expect(self.pg0, tx, self.pg1)
+ self.verify_tunneled_4o4(self.pg1, rx, tx,
+ self.pg0.local_ip4, "1.1.1.2")
+
+ #
+ # cleanup
+ #
+ route_via_tun_2.remove_vpp_config()
+ gre_if.remove_vpp_config()
+
if __name__ == '__main__':
unittest.main(testRunner=VppTestRunner)
Code Review / vpp.git / blobdiff