-#!/usr/bin/env python
+#!/usr/bin/env python3
"""IP{4,6} over IP{v,6} tunnel functional tests"""
import unittest
-from scapy.layers.inet6 import IPv6, Ether, IP, UDP, IPv6ExtHdrFragment
+from scapy.layers.inet6 import IPv6, Ether, IP, UDP, IPv6ExtHdrFragment, Raw
from scapy.all import fragment, fragment6, RandShort, defragment6
from framework import VppTestCase, VppTestRunner
from vpp_ip import DpoProto
-from vpp_ip_route import VppIpRoute, VppRoutePath, VppIpTable
+from vpp_ip_route import VppIpRoute, VppRoutePath, VppIpTable, FibPathProto
from socket import AF_INET, AF_INET6, inet_pton
from util import reassemble4
"""
+def ipip_add_tunnel(test, src, dst, table_id=0, tc_tos=0xff):
+ """ Add a IPIP tunnel """
+ return test.vapi.ipip_add_tunnel(
+ tunnel={
+ 'src': src,
+ 'dst': dst,
+ 'table_id': table_id,
+ 'instance': 0xffffffff,
+ 'tc_tos': tc_tos
+ }
+ )
+
+
class TestIPIP(VppTestCase):
""" IPIP Test Case """
p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
p_ip6 = IPv6(src="1::1", dst="DEAD::1", nh='UDP', tc=42)
p_ip4 = IP(src="1.2.3.4", dst="130.67.0.1", tos=42)
- p_payload = UDP(sport=1234, dport=1234)
+ p_payload = UDP(sport=1234, dport=1234) / Raw(b'X' * 100)
# IPv4 transport
- rv = self.vapi.ipip_add_tunnel(
- src_address=self.pg0.local_ip4n,
- dst_address=self.pg1.remote_ip4n,
- is_ipv6=0, tc_tos=0xFF)
+ rv = ipip_add_tunnel(self,
+ self.pg0.local_ip4,
+ self.pg1.remote_ip4,
+ tc_tos=0xFF)
sw_if_index = rv.sw_if_index
# Set interface up and enable IP on it
self, "130.67.0.0", 16,
[VppRoutePath("0.0.0.0",
sw_if_index,
- proto=DpoProto.DPO_PROTO_IP4)], is_ip6=0)
+ proto=FibPathProto.FIB_PATH_NH_PROTO_IP4)])
ip4_via_tunnel.add_vpp_config()
ip6_via_tunnel = VppIpRoute(
self, "dead::", 16,
[VppRoutePath("::",
sw_if_index,
- proto=DpoProto.DPO_PROTO_IP6)], is_ip6=1)
+ proto=FibPathProto.FIB_PATH_NH_PROTO_IP6)])
ip6_via_tunnel.add_vpp_config()
# IPv6 in to IPv4 tunnel
rx = self.send_and_expect(self.pg0, p6 * 10, self.pg1)
for p in rx:
self.validate(p[1], p6_reply)
+ self.assert_packet_checksums_valid(p)
# IPv4 in to IPv4 tunnel
p4 = (p_ether / p_ip4 / p_payload)
rx = self.send_and_expect(self.pg0, p4 * 10, self.pg1)
for p in rx:
self.validate(p[1], p4_reply)
+ self.assert_packet_checksums_valid(p)
# Decapsulation
p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
rx = self.send_and_expect(self.pg1, p4 * 10, self.pg0)
for p in rx:
self.validate(p[1], p4_reply)
+ self.assert_packet_checksums_valid(p)
- err = self.statistics.get_counter(
+ err = self.statistics.get_err_counter(
'/err/ipip4-input/packets decapsulated')
self.assertEqual(err, 10)
rx = self.send_and_expect(self.pg1, p6 * 10, self.pg0)
for p in rx:
self.validate(p[1], p6_reply)
+ self.assert_packet_checksums_valid(p)
- err = self.statistics.get_counter(
+ err = self.statistics.get_err_counter(
'/err/ipip4-input/packets decapsulated')
self.assertEqual(err, 20)
for p in rx:
self.validate(p[1], p4_reply)
- err = self.statistics.get_counter(
+ err = self.statistics.get_err_counter(
'/err/ipip4-input/packets decapsulated')
self.assertEqual(err, 1020)
self.pg_start()
rx = self.pg0.get_capture(6)
reass_pkt = reassemble4(rx)
- p4_reply.ttl -= 1
p4_reply.id = 256
self.validate(reass_pkt, p4_reply)
self.pg_start()
rx = self.pg0.get_capture(2)
reass_pkt = reassemble4(rx)
- p4_reply.ttl -= 1
p4_reply.id = 512
self.validate(reass_pkt, p4_reply)
+ # send large packets through the tunnel, expect them to be fragmented
+ self.vapi.sw_interface_set_mtu(sw_if_index, [600, 0, 0, 0])
+
+ p4 = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
+ IP(src="1.2.3.4", dst="130.67.0.1", tos=42) /
+ UDP(sport=1234, dport=1234) / Raw(b'Q' * 1000))
+ rx = self.send_and_expect(self.pg0, p4 * 15, self.pg1, 30)
+ inners = []
+ for p in rx:
+ inners.append(p[IP].payload)
+ reass_pkt = reassemble4(inners)
+ for p in reass_pkt:
+ self.assert_packet_checksums_valid(p)
+ self.assertEqual(p[IP].ttl, 63)
+
def test_ipip_create(self):
""" ipip create / delete interface test """
- rv = self.vapi.ipip_add_tunnel(
- src_address=inet_pton(AF_INET, '1.2.3.4'),
- dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0)
+ rv = ipip_add_tunnel(self, '1.2.3.4', '2.3.4.5')
sw_if_index = rv.sw_if_index
self.vapi.ipip_del_tunnel(sw_if_index)
t = VppIpTable(self, 20)
t.add_vpp_config()
- rv = self.vapi.ipip_add_tunnel(
- src_address=inet_pton(AF_INET, '1.2.3.4'),
- dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0,
- table_id=20)
+ rv = ipip_add_tunnel(self, '1.2.3.4', '2.3.4.5', table_id=20)
sw_if_index = rv.sw_if_index
self.vapi.ipip_del_tunnel(sw_if_index)
def setup_tunnel(self):
# IPv6 transport
- rv = self.vapi.ipip_add_tunnel(
- src_address=self.pg0.local_ip6n,
- dst_address=self.pg1.remote_ip6n, tc_tos=255)
+ rv = ipip_add_tunnel(self,
+ self.pg0.local_ip6,
+ self.pg1.remote_ip6,
+ tc_tos=255)
sw_if_index = rv.sw_if_index
self.tunnel_if_index = sw_if_index
self, "130.67.0.0", 16,
[VppRoutePath("0.0.0.0",
sw_if_index,
- proto=DpoProto.DPO_PROTO_IP4)], is_ip6=0)
+ proto=FibPathProto.FIB_PATH_NH_PROTO_IP4)])
ip4_via_tunnel.add_vpp_config()
ip6_via_tunnel = VppIpRoute(
self, "dead::", 16,
[VppRoutePath("::",
sw_if_index,
- proto=DpoProto.DPO_PROTO_IP6)], is_ip6=1)
+ proto=FibPathProto.FIB_PATH_NH_PROTO_IP6)])
ip6_via_tunnel.add_vpp_config()
self.tunnel_ip6_via_tunnel = ip6_via_tunnel
is_ip6=1)
# Send lots of fragments, verify reassembled packet
- before_cnt = self.statistics.get_counter(
+ before_cnt = self.statistics.get_err_counter(
'/err/ipip6-input/packets decapsulated')
frags, p6_reply = self.generate_ip6_frags(3131, 1400)
f = []
for p in rx:
self.validate(p[1], p6_reply)
- cnt = self.statistics.get_counter(
+ cnt = self.statistics.get_err_counter(
'/err/ipip6-input/packets decapsulated')
self.assertEqual(cnt, before_cnt + 1000)
def test_ipip_create(self):
""" ipip create / delete interface test """
- rv = self.vapi.ipip_add_tunnel(
- src_address=inet_pton(AF_INET, '1.2.3.4'),
- dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0)
+ rv = ipip_add_tunnel(self, '1.2.3.4', '2.3.4.5')
sw_if_index = rv.sw_if_index
self.vapi.ipip_del_tunnel(sw_if_index)
t = VppIpTable(self, 20)
t.add_vpp_config()
- rv = self.vapi.ipip_add_tunnel(
- src_address=inet_pton(AF_INET, '1.2.3.4'),
- dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0,
- table_id=20)
+ rv = ipip_add_tunnel(self, '1.2.3.4', '2.3.4.5', table_id=20)
sw_if_index = rv.sw_if_index
self.vapi.ipip_del_tunnel(sw_if_index)
Code Review / vpp.git / blobdiff