#!/usr/bin/env python import socket import unittest import struct from framework import VppTestCase, VppTestRunner, running_extended_tests from scapy.layers.inet import IP, TCP, UDP, ICMP from scapy.layers.inet import IPerror, TCPerror, UDPerror, ICMPerror from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest, ICMPv6EchoReply from scapy.layers.inet6 import ICMPv6DestUnreach, IPerror6 from scapy.layers.l2 import Ether, ARP, GRE from scapy.data import IP_PROTOS from scapy.packet import bind_layers from util import ppp from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder from time import sleep from util import ip4_range class MethodHolder(VppTestCase): """ NAT create capture and verify method holder """ @classmethod def setUpClass(cls): super(MethodHolder, cls).setUpClass() def tearDown(self): super(MethodHolder, self).tearDown() def check_ip_checksum(self, pkt): """ Check IP checksum of the packet :param pkt: Packet to check IP checksum """ new = pkt.__class__(str(pkt)) del new['IP'].chksum new = new.__class__(str(new)) self.assertEqual(new['IP'].chksum, pkt['IP'].chksum) def check_tcp_checksum(self, pkt): """ Check TCP checksum in IP packet :param pkt: Packet to check TCP checksum """ new = pkt.__class__(str(pkt)) del new['TCP'].chksum new = new.__class__(str(new)) self.assertEqual(new['TCP'].chksum, pkt['TCP'].chksum) def check_udp_checksum(self, pkt): """ Check UDP checksum in IP packet :param pkt: Packet to check UDP checksum """ new = pkt.__class__(str(pkt)) del new['UDP'].chksum new = new.__class__(str(new)) self.assertEqual(new['UDP'].chksum, pkt['UDP'].chksum) def check_icmp_errror_embedded(self, pkt): """ Check ICMP error embeded packet checksum :param pkt: Packet to check ICMP error embeded packet checksum """ if pkt.haslayer(IPerror): new = pkt.__class__(str(pkt)) del new['IPerror'].chksum new = new.__class__(str(new)) self.assertEqual(new['IPerror'].chksum, pkt['IPerror'].chksum) if pkt.haslayer(TCPerror): new = pkt.__class__(str(pkt)) del new['TCPerror'].chksum new = new.__class__(str(new)) self.assertEqual(new['TCPerror'].chksum, pkt['TCPerror'].chksum) if pkt.haslayer(UDPerror): if pkt['UDPerror'].chksum != 0: new = pkt.__class__(str(pkt)) del new['UDPerror'].chksum new = new.__class__(str(new)) self.assertEqual(new['UDPerror'].chksum, pkt['UDPerror'].chksum) if pkt.haslayer(ICMPerror): del new['ICMPerror'].chksum new = new.__class__(str(new)) self.assertEqual(new['ICMPerror'].chksum, pkt['ICMPerror'].chksum) def check_icmp_checksum(self, pkt): """ Check ICMP checksum in IPv4 packet :param pkt: Packet to check ICMP checksum """ new = pkt.__class__(str(pkt)) del new['ICMP'].chksum new = new.__class__(str(new)) self.assertEqual(new['ICMP'].chksum, pkt['ICMP'].chksum) if pkt.haslayer(IPerror): self.check_icmp_errror_embedded(pkt) def check_icmpv6_checksum(self, pkt): """ Check ICMPv6 checksum in IPv4 packet :param pkt: Packet to check ICMPv6 checksum """ new = pkt.__class__(str(pkt)) if pkt.haslayer(ICMPv6DestUnreach): del new['ICMPv6DestUnreach'].cksum new = new.__class__(str(new)) self.assertEqual(new['ICMPv6DestUnreach'].cksum, pkt['ICMPv6DestUnreach'].cksum) self.check_icmp_errror_embedded(pkt) if pkt.haslayer(ICMPv6EchoRequest): del new['ICMPv6EchoRequest'].cksum new = new.__class__(str(new)) self.assertEqual(new['ICMPv6EchoRequest'].cksum, pkt['ICMPv6EchoRequest'].cksum) if pkt.haslayer(ICMPv6EchoReply): del new['ICMPv6EchoReply'].cksum new = new.__class__(str(new)) self.assertEqual(new['ICMPv6EchoReply'].cksum, pkt['ICMPv6EchoReply'].cksum) def create_stream_in(self, in_if, out_if, ttl=64): """ Create packet stream for inside network :param in_if: Inside interface :param out_if: Outside interface :param ttl: TTL of generated packets """ pkts = [] # TCP p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) / IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) / TCP(sport=self.tcp_port_in, dport=20)) pkts.append(p) # UDP p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) / IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) / UDP(sport=self.udp_port_in, dport=20)) pkts.append(p) # ICMP p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) / IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) / ICMP(id=self.icmp_id_in, type='echo-request')) pkts.append(p) return pkts def compose_ip6(self, ip4, pref, plen): """ Compose IPv4-embedded IPv6 addresses :param ip4: IPv4 address :param pref: IPv6 prefix :param plen: IPv6 prefix length :returns: IPv4-embedded IPv6 addresses """ pref_n = list(socket.inet_pton(socket.AF_INET6, pref)) ip4_n = list(socket.inet_pton(socket.AF_INET, ip4)) if plen == 32: pref_n[4] = ip4_n[0] pref_n[5] = ip4_n[1] pref_n[6] = ip4_n[2] pref_n[7] = ip4_n[3] elif plen == 40: pref_n[5] = ip4_n[0] pref_n[6] = ip4_n[1] pref_n[7] = ip4_n[2] pref_n[9] = ip4_n[3] elif plen == 48: pref_n[6] = ip4_n[0] pref_n[7] = ip4_n[1] pref_n[9] = ip4_n[2] pref_n[10] = ip4_n[3] elif plen == 56: pref_n[7] = ip4_n[0] pref_n[9] = ip4_n[1] pref_n[10] = ip4_n[2] pref_n[11] = ip4_n[3] elif plen == 64: pref_n[9] = ip4_n[0] pref_n[10] = ip4_n[1] pref_n[11] = ip4_n[2] pref_n[12] = ip4_n[3] elif plen == 96: pref_n[12] = ip4_n[0] pref_n[13] = ip4_n[1] pref_n[14] = ip4_n[2] pref_n[15] = ip4_n[3] return socket.inet_ntop(socket.AF_INET6, ''.join(pref_n)) def create_stream_in_ip6(self, in_if, out_if, hlim=64, pref=None, plen=0): """ Create IPv6 packet stream for inside network :param in_if: Inside interface :param out_if: Outside interface :param ttl: Hop Limit of generated packets :param pref: NAT64 prefix :param plen: NAT64 prefix length """ pkts = [] if pref is None: dst = ''.join(['64:ff9b::', out_if.remote_ip4]) else: dst = self.compose_ip6(out_if.remote_ip4, pref, plen) # TCP p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) / IPv6(src=in_if.remote_ip6, dst=dst, hlim=hlim) / TCP(sport=self.tcp_port_in, dport=20)) pkts.append(p) # UDP p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) / IPv6(src=in_if.remote_ip6, dst=dst, hlim=hlim) / UDP(sport=self.udp_port_in, dport=20)) pkts.append(p) # ICMP p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) / IPv6(src=in_if.remote_ip6, dst=dst, hlim=hlim) / ICMPv6EchoRequest(id=self.icmp_id_in)) pkts.append(p) return pkts def create_stream_out(self, out_if, dst_ip=None, ttl=64): """ Create packet stream for outside network :param out_if: Outside interface :param dst_ip: Destination IP address (Default use global NAT address) :param ttl: TTL of generated packets """ if dst_ip is None: dst_ip = self.nat_addr pkts = [] # TCP p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) / IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) / TCP(dport=self.tcp_port_out, sport=20)) pkts.append(p) # UDP p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) / IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) / UDP(dport=self.udp_port_out, sport=20)) pkts.append(p) # ICMP p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) / IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) / ICMP(id=self.icmp_id_out, type='echo-reply')) pkts.append(p) return pkts def verify_capture_out(self, capture, nat_ip=None, same_port=False, packet_num=3, dst_ip=None): """ Verify captured packets on outside network :param capture: Captured packets :param nat_ip: Translated IP address (Default use global NAT address) :param same_port: Sorce port number is not translated (Default False) :param packet_num: Expected number of packets (Default 3) :param dst_ip: Destination IP address (Default do not verify) """ if nat_ip is None: nat_ip = self.nat_addr self.assertEqual(packet_num, len(capture)) for packet in capture: try: self.check_ip_checksum(packet) self.assertEqual(packet[IP].src, nat_ip) if dst_ip is not None: self.assertEqual(packet[IP].dst, dst_ip) if packet.haslayer(TCP): if same_port: self.assertEqual(packet[TCP].sport, self.tcp_port_in) else: self.assertNotEqual( packet[TCP].sport, self.tcp_port_in) self.tcp_port_out = packet[TCP].sport self.check_tcp_checksum(packet) elif packet.haslayer(UDP): if same_port: self.assertEqual(packet[UDP].sport, self.udp_port_in) else: self.assertNotEqual( packet[UDP].sport, self.udp_port_in) self.udp_port_out = packet[UDP].sport else: if same_port: self.assertEqual(packet[ICMP].id, self.icmp_id_in) else: self.assertNotEqual(packet[ICMP].id, self.icmp_id_in) self.icmp_id_out = packet[ICMP].id self.check_icmp_checksum(packet) except: self.logger.error(ppp("Unexpected or invalid packet " "(outside network):", packet)) raise def verify_capture_in(self, capture, in_if, packet_num=3): """ Verify captured packets on inside network :param capture: Captured packets :param in_if: Inside interface :param packet_num: Expected number of packets (Default 3) """ self.assertEqual(packet_num, len(capture)) for packet in capture: try: self.check_ip_checksum(packet) self.assertEqual(packet[IP].dst, in_if.remote_ip4) if packet.haslayer(TCP): self.assertEqual(packet[TCP].dport, self.tcp_port_in) self.check_tcp_checksum(packet) elif packet.haslayer(UDP): self.assertEqual(packet[UDP].dport, self.udp_port_in) else: self.assertEqual(packet[ICMP].id, self.icmp_id_in) self.check_icmp_checksum(packet) except: self.logger.error(ppp("Unexpected or invalid packet " "(inside network):", packet)) raise def verify_capture_in_ip6(self, capture, src_ip, dst_ip, packet_num=3): """ Verify captured IPv6 packets on inside network :param capture: Captured packets :param src_ip: Source IP :param dst_ip: Destination IP address :param packet_num: Expected number of packets (Default 3) """ self.assertEqual(packet_num, len(capture)) for packet in capture: try: self.assertEqual(packet[IPv6].src, src_ip) self.assertEqual(packet[IPv6].dst, dst_ip) if packet.haslayer(TCP): self.assertEqual(packet[TCP].dport, self.tcp_port_in) self.check_tcp_checksum(packet) elif packet.haslayer(UDP): self.assertEqual(packet[UDP].dport, self.udp_port_in) self.check_udp_checksum(packet) else: self.assertEqual(packet[ICMPv6EchoReply].id, self.icmp_id_in) self.check_icmpv6_checksum(packet) except: self.logger.error(ppp("Unexpected or invalid packet " "(inside network):", packet)) raise def verify_capture_no_translation(self, capture, ingress_if, egress_if): """ Verify captured packet that don't have to be translated :param capture: Captured packets :param ingress_if: Ingress interface :param egress_if: Egress interface """ for packet in capture: try: self.assertEqual(packet[IP].src, ingress_if.remote_ip4) self.assertEqual(packet[IP].dst, egress_if.remote_ip4) if packet.haslayer(TCP): self.assertEqual(packet[TCP].sport, self.tcp_port_in) elif packet.haslayer(UDP): self.assertEqual(packet[UDP].sport, self.udp_port_in) else: self.assertEqual(packet[ICMP].id, self.icmp_id_in) except: self.logger.error(ppp("Unexpected or invalid packet " "(inside network):", packet)) raise def verify_capture_out_with_icmp_errors(self, capture, src_ip=None, packet_num=3, icmp_type=11): """ Verify captured packets with ICMP errors on outside network :param capture: Captured packets :param src_ip: Translated IP address or IP address of VPP (Default use global NAT address) :param packet_num: Expected number of packets (Default 3) :param icmp_type: Type of error ICMP packet we are expecting (Default 11) """ if src_ip is None: src_ip = self.nat_addr self.assertEqual(packet_num, len(capture)) for packet in capture: try: self.assertEqual(packet[IP].src, src_ip) self.assertTrue(packet.haslayer(ICMP)) icmp = packet[ICMP] self.assertEqual(icmp.type, icmp_type) self.assertTrue(icmp.haslayer(IPerror)) inner_ip = icmp[IPerror] if inner_ip.haslayer(TCPerror): self.assertEqual(inner_ip[TCPerror].dport, self.tcp_port_out) elif inner_ip.haslayer(UDPerror): self.assertEqual(inner_ip[UDPerror].dport, self.udp_port_out) else: self.assertEqual(inner_ip[ICMPerror].id, self.icmp_id_out) except: self.logger.error(ppp("Unexpected or invalid packet " "(outside network):", packet)) raise def verify_capture_in_with_icmp_errors(self, capture, in_if, packet_num=3, icmp_type=11): """ Verify captured packets with ICMP errors on inside network :param capture: Captured packets :param in_if: Inside interface :param packet_num: Expected number of packets (Default 3) :param icmp_type: Type of error ICMP packet we are expecting (Default 11) """ self.assertEqual(packet_num, len(capture)) for packet in capture: try: self.assertEqual(packet[IP].dst, in_if.remote_ip4) self.assertTrue(packet.haslayer(ICMP)) icmp = packet[ICMP] self.assertEqual(icmp.type, icmp_type) self.assertTrue(icmp.haslayer(IPerror)) inner_ip = icmp[IPerror] if inner_ip.haslayer(TCPerror): self.assertEqual(inner_ip[TCPerror].sport, self.tcp_port_in) elif inner_ip.haslayer(UDPerror): self.assertEqual(inner_ip[UDPerror].sport, self.udp_port_in) else: self.assertEqual(inner_ip[ICMPerror].id, self.icmp_id_in) except: self.logger.error(ppp("Unexpected or invalid packet " "(inside network):", packet)) raise def verify_ipfix_nat44_ses(self, data): """ Verify IPFIX NAT44 session create/delete event :param data: Decoded IPFIX data records """ nat44_ses_create_num = 0 nat44_ses_delete_num = 0 self.assertEqual(6, len(data)) for record in data: # natEvent self.assertIn(ord(record[230]), [4, 5]) if ord(record[230]) == 4: nat44_ses_create_num += 1 else: nat44_ses_delete_num += 1 # sourceIPv4Address self.assertEqual(self.pg0.remote_ip4n, record[8]) # postNATSourceIPv4Address self.assertEqual(socket.inet_pton(socket.AF_INET, self.nat_addr), record[225]) # ingressVRFID self.assertEqual(struct.pack("!I", 0), record[234]) # protocolIdentifier/sourceTransportPort/postNAPTSourceTransportPort if IP_PROTOS.icmp == ord(record[4]): self.assertEqual(struct.pack("!H", self.icmp_id_in), record[7]) self.assertEqual(struct.pack("!H", self.icmp_id_out), record[227]) elif IP_PROTOS.tcp == ord(record[4]): self.assertEqual(struct.pack("!H", self.tcp_port_in), record[7]) self.assertEqual(struct.pack("!H", self.tcp_port_out), record[227]) elif IP_PROTOS.udp == ord(record[4]): self.assertEqual(struct.pack("!H", self.udp_port_in), record[7]) self.assertEqual(struct.pack("!H", self.udp_port_out), record[227]) else: self.fail("Invalid protocol") self.assertEqual(3, nat44_ses_create_num) self.assertEqual(3, nat44_ses_delete_num) def verify_ipfix_addr_exhausted(self, data): """ Verify IPFIX NAT addresses event :param data: Decoded IPFIX data records """ self.assertEqual(1, len(data)) record = data[0] # natEvent self.assertEqual(ord(record[230]), 3) # natPoolID self.assertEqual(struct.pack("!I", 0), record[283]) class TestNAT44(MethodHolder): """ NAT44 Test Cases """ @classmethod def setUpClass(cls): super(TestNAT44, cls).setUpClass() try: cls.tcp_port_in = 6303 cls.tcp_port_out = 6303 cls.udp_port_in = 6304 cls.udp_port_out = 6304 cls.icmp_id_in = 6305 cls.icmp_id_out = 6305 cls.nat_addr = '10.0.0.3' cls.ipfix_src_port = 4739 cls.ipfix_domain_id = 1 cls.create_pg_interfaces(range(9)) cls.interfaces = list(cls.pg_interfaces[0:4]) for i in cls.interfaces: i.admin_up() i.config_ip4() i.resolve_arp() cls.pg0.generate_remote_hosts(3) cls.pg0.configure_ipv4_neighbors() cls.overlapping_interfaces = list(list(cls.pg_interfaces[4:7])) cls.pg4._local_ip4 = "172.16.255.1" cls.pg4._local_ip4n = socket.inet_pton(socket.AF_INET, i.local_ip4) cls.pg4._remote_hosts[0]._ip4 = "172.16.255.2" cls.pg4.set_table_ip4(10) cls.pg5._local_ip4 = "172.17.255.3" cls.pg5._local_ip4n = socket.inet_pton(socket.AF_INET, i.local_ip4) cls.pg5._remote_hosts[0]._ip4 = "172.17.255.4" cls.pg5.set_table_ip4(10) cls.pg6._local_ip4 = "172.16.255.1" }
#!/usr/bin/env python
from socket import AF_INET, AF_INET6, inet_pton
from framework import VppTestCase, VppTestRunner
from vpp_interface import VppInterface
from vpp_ip_route import VppIpTable, VppIpRoute, VppRoutePath
from scapy.packet import Raw
from scapy.layers.l2 import Ether
from scapy.layers.inet import IP, UDP
class VppPipe(VppInterface):
"""
VPP Pipe
"""
@property
def east(self):
return self.result.pipe_sw_if_index[1]
@property
def west(self):
return self.result.pipe_sw_if_index[0]
def __init__(self, test, instance=0xffffffff):
super(VppPipe, self).__init__(test)
self._test = test
self.instance = instance
def add_vpp_config(self):
self.result = self._test.vapi.pipe_create(
0 if self.instance == 0xffffffff else 1,
self.instance)
self.set_sw_if_index(self.result.sw_if_index)
def remove_vpp_config(self):
self._test.vapi.pipe_delete(
self.result.sw_if_index)
def __str__(self):
return self.object_id()
def object_id(self):
return "pipe-%d" % (self._sw_if_index)
def query_vpp_config(self):
pipes = self._test.vapi.pipe_dump()
for p in pipes:
if p.sw_if_index == self.result.sw_if_index:
return True
return False
def set_unnumbered(self, ip_sw_if_index, is_add=True):
res = self._test.vapi.sw_interface_set_unnumbered(
self.east, ip_sw_if_index, is_add)
res = self._test.vapi.sw_interface_set_unnumbered(
self.west, ip_sw_if_index, is_add)
class TestPipe(VppTestCase):
""" Pipes """
def setUp(self):
super(TestPipe, self).setUp()
self.create_pg_interfaces(range(4))
for i in self.pg_interfaces:
i.admin_up()
def tearDown(self):
for i in self.pg_interfaces:
i.admin_down()
super(TestPipe, self).tearDown()
def test_pipe(self):
""" Pipes """
pipes = [VppPipe(self), VppPipe(self, 10)]
for p in pipes:
p.add_vpp_config()
p.admin_up()
#
# L2 cross-connect pipe0 east with pg0 and west with pg1
#
self.vapi.sw_interface_set_l2_xconnect(self.pg0.sw_if_index,
pipes[0].east,
enable=1)
self.vapi.sw_interface_set_l2_xconnect(pipes[0].east,
self.pg0.sw_if_index,
enable=1)
self.vapi.sw_interface_set_l2_xconnect(self.pg1.sw_if_index,
pipes[0].west,
enable=1)
self.vapi.sw_interface_set_l2_xconnect(pipes[0].west,
self.pg1.sw_if_index,
enable=1)
# test bi-drectional L2 flow pg0<->pg1
p = (Ether(src=self.pg0.remote_mac,
dst=self.pg1.remote_mac) /
IP(src="1.1.1.1",
dst="1.1.1.2") /
UDP(sport=1234, dport=1234) /
Raw('\xa5' * 100))
self.send_and_expect(self.pg0, p * 65, self.pg1)
self.send_and_expect(self.pg1, p * 65, self.pg0)
#
# Attach ACL to ensure features are run on the pipe
#
rule_1 = ({'is_permit': 0,
'is_ipv6': 0,
'proto': 17,
'srcport_or_icmptype_first': 1234,
'srcport_or_icmptype_last': 1234,
'src_ip_prefix_len': 32,
'src_ip_addr': inet_pton(AF_INET, "1.1.1.1"),
'dstport_or_icmpcode_first': 1234,
'dstport_or_icmpcode_last': 1234,
'dst_ip_prefix_len': 32,
'dst_ip_addr': inet_pton(AF_INET, "1.1.1.2")})
acl = self.vapi.acl_add_replace(acl_index=4294967295,
r=[rule_1])
# Apply the ACL on the pipe on output
self.vapi.acl_interface_set_acl_list(pipes[0].east,
0,
[acl.acl_index])
self.send_and_assert_no_replies(self.pg0, p * 65)
self.send_and_expect(self.pg1, p * 65, self.pg0)
# remove from output and apply on input
self.vapi.acl_interface_set_acl_list(pipes[0].east,
0,
[])
self.vapi.acl_interface_set_acl_list(pipes[0].west,
1,
[acl.acl_index])
self.send_and_assert_no_replies(self.pg0, p * 65)
self.send_and_expect(self.pg1, p * 65, self.pg0)
self.vapi.acl_interface_set_acl_list(pipes[0].west,
0,
[])
self.send_and_expect(self.pg0, p * 65, self.pg1)
self.send_and_expect(self.pg1, p * 65, self.pg0)
#
# L3 routes in two separate tables so a pipe can be used to L3
# x-connect
#
tables = []
tables.append(VppIpTable(self, 1))
tables.append(VppIpTable(self, 2))
for t in tables:
t.add_vpp_config()
self.pg2.set_table_ip4(1)
self.pg2.config_ip4()
self.pg2.resolve_arp()
self.pg3.set_table_ip4(2)
self.pg3.config_ip4()
self.pg3.resolve_arp()
routes = []
routes.append(VppIpRoute(self, "1.1.1.1", 32,
[VppRoutePath(self.pg3.remote_ip4,
self.pg3.sw_if_index)],
table_id=2))
routes.append(VppIpRoute(self, "1.1.1.1", 32,
[VppRoutePath("0.0.0.0", pipes[1].east)],
table_id=1))
routes.append(VppIpRoute(self, "1.1.1.2", 32,
[VppRoutePath("0.0.0.0", pipes[1].west)],
table_id=2))
routes.append(VppIpRoute(self, "1.1.1.2", 32,
[VppRoutePath(self.pg2.remote_ip4,
self.pg2.sw_if_index)],
table_id=1))
for r in routes:
r.add_vpp_config()
p_east = (Ether(src=self.pg2.remote_mac,
dst=self.pg2.local_mac) /
IP(src="1.1.1.2",
dst="1.1.1.1") /
UDP(sport=1234, dport=1234) /
Raw('\xa5' * 100))
# bind the pipe ends to the correct tables
self.vapi.sw_interface_set_table(pipes[1].west, 0, 2)
self.vapi.sw_interface_set_table(pipes[1].east, 0, 1)
# IP is not enabled on the pipes at this point
self.send_and_assert_no_replies(self.pg2, p_east * 65)
# IP enable the Pipes by making them unnumbered
pipes[0].set_unnumbered(self.pg2.sw_if_index)
pipes[1].set_unnumbered(self.pg3.sw_if_index)
self.send_and_expect(self.pg2, p_east * 65, self.pg3)
# and the return path
p_west = (Ether(src=self.pg3.remote_mac,
dst=self.pg3.local_mac) /
IP(src="1.1.1.1",
dst="1.1.1.2") /
UDP(sport=1234, dport=1234) /
Raw('\xa5' * 100))
self.send_and_expect(self.pg3, p_west * 65, self.pg2)
#
# Use ACLs to test features run on the Pipes
#
self.vapi.acl_interface_set_acl_list(pipes[1].east,
0,
[acl.acl_index])
self.send_and_assert_no_replies(self.pg2, p_east * 65)
self.send_and_expect(self.pg3, p_west * 65, self.pg2)
# remove from output and apply on input
self.vapi.acl_interface_set_acl_list(pipes[1].east,
0,
[])
self.vapi.acl_interface_set_acl_list(pipes[1].west,
1,
[acl.acl_index])
self.send_and_assert_no_replies(self.pg2, p_east * 65)
self.send_and_expect(self.pg3, p_west * 65, self.pg2)
self.vapi.acl_interface_set_acl_list(pipes[1].west,
0,
[])
self.send_and_expect(self.pg2, p_east * 65, self.pg3)
self.send_and_expect(self.pg3, p_west * 65, self.pg2)
# cleanup (so the tables delete)
self.pg2.unconfig_ip4()
self.pg2.set_table_ip4(0)
self.pg3.unconfig_ip4()
self.pg3.set_table_ip4(0)
self.vapi.sw_interface_set_table(pipes[1].west, 0, 0)
self.vapi.sw_interface_set_table(pipes[1].east, 0, 0)
if __name__ == '__main__':
unittest.main(testRunner=VppTestRunner)