#!/usr/bin/env python import unittest import binascii from socket import AF_INET6 from framework import VppTestCase, VppTestRunner from vpp_ip import DpoProto from vpp_ip_route import VppIpRoute, VppRoutePath, VppIpTable from vpp_srv6 import SRv6LocalSIDBehaviors, VppSRv6LocalSID, VppSRv6Policy, \ SRv6PolicyType, VppSRv6Steering, SRv6PolicySteeringTypes from scapy.packet import Raw from scapy.layers.l2 import Ether, Dot1Q from scapy.layers.inet6 import IPv6, UDP, IPv6ExtHdrSegmentRouting from scapy.layers.inet import IP, UDP from scapy.utils import inet_pton, inet_ntop from util import ppp class TestSRv6(VppTestCase): """ SRv6 Dynamic Proxy plugin Test Case """ @classmethod def setUpClass(self): super(TestSRv6, self).setUpClass() def setUp(self): """ Perform test setup before each test case. """ super(TestSRv6, self).setUp() # packet sizes, inclusive L2 overhead self.pg_packet_sizes = [64, 512, 1518, 9018] # reset packet_infos self.reset_packet_infos() def tearDown(self): """ Clean up test setup after each test case. """ self.teardown_interfaces() super(TestSRv6, self).tearDown() def configure_interface(self, interface, ipv6=False, ipv4=False, ipv6_table_id=0, ipv4_table_id=0): """ Configure interface. :param ipv6: configure IPv6 on interface :param ipv4: configure IPv4 on interface :param ipv6_table_id: FIB table_id for IPv6 :param ipv4_table_id: FIB table_id for IPv4 """ self.logger.debug("Configuring interface %s" % (interface.name)) if ipv6: self.logger.debug("Configuring IPv6") interface.set_table_ip6(ipv6_table_id) interface.config_ip6() interface.resolve_ndp(timeout=5) if ipv4: self.logger.debug("Configuring IPv4") interface.set_table_ip4(ipv4_table_id) interface.config_ip4() interface.resolve_arp() interface.admin_up() def setup_interfaces(self, ipv6=[], ipv4=[], ipv6_table_id=[], ipv4_table_id=[]): """ Create and configure interfaces. :param ipv6: list of interface IPv6 capabilities :param ipv4: list of interface IPv4 capabilities :param ipv6_table_id: list of intf IPv6 FIB table_ids :param ipv4_table_id: list of intf IPv4 FIB table_ids :returns: List of created interfaces. """ # how many interfaces? if len(ipv6): count = len(ipv6) else: count = len(ipv4) self.logger.debug("Creating and configuring %d interfaces" % (count)) # fill up ipv6 and ipv4 lists if needed # not enabled (False) is the default if len(ipv6) < count: ipv6 += (count - len(ipv6)) * [False] if len(ipv4) < count: ipv4 += (count - len(ipv4)) * [False] # fill up table_id lists if needed # table_id 0 (global) is the default if len(ipv6_table_id) < count: ipv6_table_id += (count - len(ipv6_table_id)) * [0] if len(ipv4_table_id) < count: ipv4_table_id += (count - len(ipv4_table_id)) * [0] # create 'count' pg interfaces self.create_pg_interfaces(range(count)) # setup all interfaces for i in range(count): intf = self.pg_interfaces[i] self.configure_interface(intf, ipv6[i], ipv4[i], ipv6_table_id[i], ipv4_table_id[i]) if any(ipv6): self.logger.debug(self.vapi.cli("show ip6 neighbors")) if any(ipv4): self.logger.debug(self.vapi.cli("show ip arp")) self.logger.debug(self.vapi.cli("show interface")) self.logger.debug(self.vapi.cli("show hardware")) return self.pg_interfaces def teardown_interfaces(self): """ Unconfigure and bring down interface. """ self.logger.debug("Tearing down interfaces") # tear down all interfaces # AFAIK they cannot be deleted for i in self.pg_interfaces: self.logger.debug("Tear down interface %s" % (i.name)) i.admin_down() i.unconfig() i.set_table_ip4(0) i.set_table_ip6(0) def test_SRv6_End_AD_IPv6(self): """ Test SRv6 End.AD behavior with IPv6 traffic. """ self.src_addr = 'a0::' self.sid_list = ['a1::', 'a2::a6', 'a3::'] self.test_sid_index = 1 # send traffic to one destination interface # source and destination interfaces are IPv6 only self.setup_interfaces(ipv6=[True, True]) # configure route to next segment route = VppIpRoute(self, self.sid_list[self.test_sid_index + 1], 128, [VppRoutePath(self.pg0.remote_ip6, self.pg0.sw_if_index, proto=DpoProto.DPO_PROTO_IP6)], is_ip6=1) route.add_vpp_config() # configure SRv6 localSID behavior cli_str = "sr localsid address " + \ self.sid_list[self.test_sid_index] + \ " behavior end.ad" + \ " nh " + self.pg1.remote_ip6 + \ " oif " + self.pg1.name + \ " iif " + self.pg1.name self.vapi.cli(cli_str) # log the localsids self.logger.debug(self.vapi.cli("show sr localsid")) # send one packet per packet size count = len(self.pg_packet_sizes) # prepare IPv6 in SRv6 headers packet_header1 = self.create_packet_header_IPv6_SRH_IPv6( srcaddr=self.src_addr, sidlist=self.sid_list[::-1], segleft=len(self.sid_list) - self.test_sid_index - 1) # generate packets (pg0->pg1) pkts1 = self.create_stream(self.pg0, self.pg1, packet_header1, self.pg_packet_sizes, count) # send packets and verify received packets self.send_and_verify_pkts(self.pg0, pkts1, self.pg1, self.compare_rx_tx_packet_End_AD_IPv6_out) # log the localsid counters self.logger.info(self.vapi.cli("show sr localsid")) # prepare IPv6 header for returning packets packet_header2 = self.create_packet_header_IPv6() # generate returning packets (pg1->pg0) pkts2 = self.create_stream(self.pg1, self.pg0, packet_header2, self.pg_packet_sizes, count) # send packets and verify received packets self.send_and_verify_pkts(self.pg1, pkts2, self.pg0, self.compare_rx_tx_packet_End_AD_IPv6_in) # log the localsid counters self.logger.info(self.vapi.cli("show sr localsid")) # remove SRv6 localSIDs cli_str = "sr localsid del address " + \ self.sid_list[self.test_sid_index] self.vapi.cli(cli_str) # cleanup interfaces self.teardown_interfaces() def compare_rx_tx_packet_End_AD_IPv6_out(self, tx_pkt, rx_pkt): """ Compare input and output packet after passing End.AD with IPv6 :param tx_pkt: transmitted packet :param rx_pkt: received packet """ # get first (outer) IPv6 header of rx'ed packet rx_ip = rx_pkt.getlayer(IPv6) tx_ip = tx_pkt.getlayer(IPv6) tx_ip2 = tx_pkt.getlayer(IPv6, 2) # verify if rx'ed packet has no SRH self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting)) # the whole rx_ip pkt should be equal to tx_ip2 # except for the hlim field # -> adjust tx'ed hlim to expected hlim tx_ip2.hlim = tx_ip2.hlim - 1 self.assertEqual(rx_ip, tx_ip2) self.logger.debug("packet verification: SUCCESS") def compare_rx_tx_packet_End_AD_IPv6_in(self, tx_pkt, rx_pkt): """ Compare input and output packet after passing End.AD :param tx_pkt: transmitted packet :param rx_pkt: received packet """ # get first (outer) IPv6 header of rx'ed packet rx_ip = rx_pkt.getlayer(IPv6) # received ip.src should be equal to SR Policy source self.assertEqual(rx_ip.src, self.src_addr) # received ip.dst should be equal to expected sidlist next segment self.assertEqual(rx_ip.dst, self.sid_list[self.test_sid_index + 1]) # rx'ed packet should have SRH self.assertTrue(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting)) # get SRH rx_srh = rx_pkt.getlayer(IPv6ExtHdrSegmentRouting) # rx'ed seglist should be equal to SID-list in reversed order self.assertEqual(rx_srh.addresses, self.sid_list[::-1]) # segleft should be equal to previous segleft value minus 1 self.assertEqual(rx_srh.segleft, len(self.sid_list) - self.test_sid_index - 2) # lastentry should be equal to the SID-list length minus 1 self.assertEqual(rx_srh.lastentry, len(self.sid_list) - 1) # the whole rx'ed pkt beyond SRH should be equal to tx'ed pkt # except for the hop-limit field tx_ip = tx_pkt.getlayer(IPv6) # -> update tx'ed hlim to the expected hlim tx_ip.hlim -= 1 # -> check payload self.assertEqual(rx_srh.payload, tx_ip) self.logger.debug("packet verification: SUCCESS") def test_SRv6_End_AD_IPv4(self): """ Test SRv6 End.AD behavior with IPv4 traffic. """ self.src_addr = 'a0::' self.sid_list = ['a1::', 'a2::a4', 'a3::'] self.test_sid_index = 1 # send traffic to one destination interface # source and destination interfaces are IPv6 only self.setup_interfaces(ipv6=[True, False], ipv4=[False, True]) # configure route to next segment route = VppIpRoute(self, self.sid_list[self.test_sid_index + 1], 128, [VppRoutePath(self.pg0.remote_ip6, self.pg0.sw_if_index, proto=DpoProto.DPO_PROTO_IP6)], is_ip6=1) route.add_vpp_config() # configure SRv6 localSID behavior cli_str = "sr localsid address " + \
/*
* Copyright (c) 2017 Cisco and/or its affiliates.
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <stdio.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netdb.h>
#include <vppinfra/format.h>
#include <sys/time.h>
int
main (int argc, char *argv[])
{
int sockfd, portno, n;
struct sockaddr_in serv_addr;
struct hostent *server;
u8 *rx_buffer = 0, *tx_buffer = 0, no_echo = 0, test_bytes = 0;
u32 offset;
long bytes = 1 << 20, to_send;
int i;
struct timeval start, end;
double deltat;
if (argc >= 3)
{
portno = atoi (argv[2]);
server = gethostbyname (argv[1]);
if (server == NULL)
{
clib_unix_warning ("gethostbyname");
exit (1);
}
argc -= 3;
argv += 3;
if (argc)
{
bytes = ((long) atoi (argv[0])) << 20;
argc--;
argv++;
}
if (argc)
{
no_echo = atoi (argv[0]);
argc--;
argv++;
}
if (argc)
{
test_bytes = atoi (argv[0]);
argc--;
argv++;
}
}
else
{
portno = 1234; // atoi(argv[2]);
server = gethostbyname ("6.0.1.1" /* argv[1] */ );
if (server == NULL)
{
clib_unix_warning ("gethostbyname");
exit (1);
}
}
to_send = bytes;
sockfd = socket (AF_INET, SOCK_STREAM, 0);
if (sockfd < 0)
{
clib_unix_error ("socket");
exit (1);
}
bzero ((char *) &serv_addr, sizeof (serv_addr));
serv_addr.sin_family = AF_INET;
bcopy ((char *) server->h_addr,
(char *) &serv_addr.sin_addr.s_addr, server->h_length);
serv_addr.sin_port = htons (portno);
if (connect (sockfd, (const void *) &serv_addr, sizeof (serv_addr)) < 0)
{
clib_unix_warning ("connect");
exit (1);
}
vec_validate (rx_buffer, 128 << 10);
vec_validate (tx_buffer, 128 << 10);
for (i = 0; i < vec_len (tx_buffer); i++)
tx_buffer[i] = (i +