#!/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, VppGre6Interface
+from vpp_ip import DpoProto
+from vpp_ip_route import VppIpRoute, VppRoutePath, VppIpTable
from util import ppp, ppc
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.
#
#
# 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.
#
# 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