#!/usr/bin/env python3
import unittest
from framework import tag_fixme_vpp_workers
from framework import VppTestCase, VppTestRunner
from vpp_udp_encap import find_udp_encap, VppUdpEncap
from vpp_udp_decap import VppUdpDecap
from vpp_ip_route import (
VppIpRoute,
VppRoutePath,
VppIpTable,
VppMplsLabel,
VppMplsTable,
VppMplsRoute,
FibPathType,
FibPathProto,
)
from vpp_neighbor import VppNeighbor
from vpp_papi import VppEnum
from scapy.packet import Raw
from scapy.layers.l2 import Ether
from scapy.layers.inet import IP, UDP, ICMP
from scapy.layers.inet6 import IPv6
from scapy.contrib.mpls import MPLS
NUM_PKTS = 67
ENTROPY_PORT_MIN = 0x3 << 14
ENTROPY_PORT_MAX = 0xFFFF
@tag_fixme_vpp_workers
class TestUdpEncap(VppTestCase):
"""UDP Encap Test Case"""
@classmethod
def setUpClass(cls):
super(TestUdpEncap, cls).setUpClass()
@classmethod
def tearDownClass(cls):
super(TestUdpEncap, cls).tearDownClass()
def setUp(self):
super(TestUdpEncap, self).setUp()
# create 2 pg interfaces
self.create_pg_interfaces(range(4))
# setup interfaces
# assign them different tables.
table_id = 0
self.tables = []
for i in self.pg_interfaces:
i.admin_up()
if table_id != 0:
tbl = VppIpTable(self, table_id)
tbl.add_vpp_config()
self.tables.append(tbl)
tbl = VppIpTable(self, table_id, is_ip6=1)
tbl.add_vpp_config()
self.tables.append(tbl)
i.set_table_ip4(table_id)
i.set_table_ip6(table_id)
i.config_ip4()
i.resolve_arp()
i.config_ip6()
i.resolve_ndp()
table_id += 1
def tearDown(self):
for i in self.pg_interfaces:
i.unconfig_ip4()
i.unconfig_ip6()
i.set_table_ip4(0)
i.set_table_ip6(0)
i.admin_down()
super(TestUdpEncap, self).tearDown()
def validate_outer4(self, rx, encap_obj, sport_entropy=False):
self.assertEqual(rx[IP].src, encap_obj.src_ip_s)
self.assertEqual(rx[IP].dst, encap_obj.dst_ip_s)
if sport_entropy:
self.assert_in_range(rx[UDP].sport, ENTROPY_PORT_MIN, ENTROPY_PORT_MAX)
else:
self.assertEqual(rx[UDP].sport, encap_obj.src_port)
self.assertEqual(rx[UDP].dport, encap_obj.dst_port)
def validate_outer6(self, rx, encap_obj, sport_entropy=False):
self.assertEqual(rx[IPv6].src, encap_obj.src_ip_s)
self.assertEqual(rx[IPv6].dst, encap_obj.dst_ip_s)
if sport_entropy:
self.assert_in_range(rx[UDP].sport, ENTROPY_PORT_MIN, ENTROPY_PORT_MAX)
else:
self.assertEqual(rx[UDP].sport, encap_obj.src_port)
self.assertEqual(rx[UDP].dport, encap_obj.dst_port)
def validate_inner4(self, rx, tx, ttl=None):
self.assertEqual(rx[IP].src, tx[IP].src)
self.assertEqual(rx[IP].dst, tx[IP].dst)
if ttl:
self.assertEqual(rx[IP].ttl, ttl)
else:
self.assertEqual(rx[IP].ttl, tx[IP].ttl)
def validate_inner6(self, rx, tx, hlim=None):
self.assertEqual(rx.src, tx[IPv6].src)
self.assertEqual(rx.dst, tx[IPv6].dst)
if hlim:
self.assertEqual(rx.hlim, hlim)
else:
self.assertEqual(rx.hlim, tx[IPv6].hlim)
def test_udp_encap(self):
"""UDP Encap test"""
#
# construct a UDP encap object through each of the peers
# v4 through the first two peers, v6 through the second.
# The last encap is v4 and is used to check the codepath
# where 2 different udp encap objects are processed at the
# same time
#
udp_encap_0 = VppUdpEncap(
self, self.pg0.local_ip4, self.pg0.remote_ip4, 330, 440
)
udp_encap_1 = VppUdpEncap(
self, self.pg1.local_ip4, self.pg1.remote_ip4, 331, 441, table_id=1
)
udp_encap_2 = VppUdpEncap(
self, self.pg2.local_ip6, self.pg2.remote_ip6, 332, 442, table_id=2
)
udp_encap_3 = VppUdpEncap(
self, self.pg3.local_ip6, self.pg3.remote_ip6, 333, 443, table_id=3
)
udp_encap_4 = VppUdpEncap(
self, self.pg0.local_ip4, self.pg0.remote_ip4, 334, 444
)
udp_encap_0.add_vpp_config()
udp_encap_1.add_vpp_config()
udp_encap_2.add_vpp_config()
udp_encap_3.add_vpp_config()
udp_encap_4.add_vpp_config()
self.logger.info(self.vapi.cli("sh udp encap"))
self.assertTrue(find_udp_encap(self, udp_encap_2))
self.assertTrue(find_udp_encap(self, udp_encap_3))
self.assertTrue(find_udp_encap(self, udp_encap_0))
self.assertTrue(find_udp_encap(self, udp_encap_1))
se
echo \n\n====== gdb_cmdfile.vpp ======\n
# Here are some interesting vpp-api breakpoints...
# Uncomment them out to set during gdb init.
#
#b vl_api_connect_sock_t_handler
#b vl_api_connect_sock_reply_t_handler