#!/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.l2 import Ether, ARP from scapy.data import IP_PROTOS from util import ppp from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder from time import sleep class MethodHolder(VppTestCase): """ SNAT create capture and verify method holder """ @classmethod def setUpClass(cls): super(MethodHolder, cls).setUpClass() def tearDown(self): super(MethodHolder, self).tearDown() 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)) 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)) 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 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 SNAT address) :param ttl: TTL of generated packets """ if dst_ip is None: dst_ip = self.snat_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)) 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)) 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): """ Verify captured packets on outside network :param capture: Captured packets :param nat_ip: Translated IP address (Default use global SNAT address) :param same_port: Sorce port number is not translated (Default False) :param packet_num: Expected number of packets (Default 3) """ if nat_ip is None: nat_ip = self.snat_addr self.assertEqual(packet_num, len(capture)) for packet in capture: try: self.assertEqual(packet[IP].src, nat_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 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 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.assertEqual(packet[IP].dst, in_if.remote_ip4) if packet.haslayer(TCP): self.assertEqual(packet[TCP].dport, self.tcp_port_in) elif packet.haslayer(UDP): self.assertEqual(packet[UDP].dport, 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_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 SNAT 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.snat_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) "
#!/usr/bin/env python
""" Container integration tests """
import unittest
from framework import VppTestCase, VppTestRunner, running_extended_tests
from scapy.layers.l2 import Ether
from scapy.packet import Raw
from scapy.layers.inet import IP, UDP, TCP
from scapy.packet import Packet
from socket import inet_pton, AF_INET, AF_INET6
from scapy.layers.inet6 import IPv6, ICMPv6Unknown, ICMPv6EchoRequest
from scapy.layers.inet6 import ICMPv6EchoReply, IPv6ExtHdrRouting
from scapy.layers.inet6 import IPv6ExtHdrFragment
from pprint import pprint
from random import randint
from util import L4_Conn
class Conn(L4_Conn):
# for now same as L4_Conn
pass
@unittest.skipUnless(running_extended_tests(), "part of extended tests")
class ContainerIntegrationTestCase(VppTestCase):
""" Container integration extended testcases """
@classmethod
def setUpClass(self):
super(ContainerIntegrationTestCase, self).setUpClass()
# create pg0 and pg1
self.create_pg_interfaces(range(2))
for i in self.pg_interfaces:
i.admin_up()
i.config_ip4()
i.config_ip6()
i.resolve_arp()
i.resolve_ndp()
def tearDown(self):
"""Run standard test teardown and log various show commands
"""
super(ContainerIntegrationTestCase, self).tearDown()
if not self.vpp_dead:
self.logger.info(self.vapi.cli("show ip arp"))
self.logger.info(self.vapi.cli("show ip6 neighbors"))
def run_basic_conn_test(self, af, acl_side):
""" Basic connectivity test """
conn1 = Conn(self, self.pg0, self.pg1, af, UDP, 42001, 4242)
conn1.send_through(0)
# the return packets should pass
conn1.send_through(1)
def run_negative_conn_test(self, af, acl_side):
""" Packets with local spoofed address """
conn1 = Conn(self, self.pg0, self.pg1, af, UDP, 42001, 4242)
try:
p2 = conn1.send_through(0).command()
except:
# If we asserted while waiting, it's good.
# the conn should have timed out.
p2 = None
self.assert_equal(p2, None, ": packet should have been dropped")
def test_0010_basic_conn_test(self):
""" IPv4 basic connectivity test """
self.run_basic_conn_test(AF_INET, 0)
def test_0011_basic_conn_test(self):
""" IPv6 basic connectivity test """
self.run_basic_conn_test(AF_INET6, 0)
def test_0050_loopback_prepare_test(self):
""" Create loopbacks overlapping with remote addresses """
self.create_loopback_interfaces(2)
for i in range(2):
intf = self.lo_interfaces[i]
intf.admin_up()
intf._local_ip4 = self.pg_interfaces[i].remote_ip4
intf._local_ip4_prefix_len = 32
intf.config_ip4()
intf._local_ip6 = self.pg_interfaces[i].remote_ip6
intf._local_ip6_prefix_len = 128
intf.config_ip6()
def test_0110_basic_conn_test(self):
""" IPv4 local-spoof connectivity test """
self.run_negative_conn_test(AF_INET, 0)
def test_0111_basic_conn_test(self):
""" IPv6 local-spoof connectivity test """
self.run_negative_conn_test(AF_INET, 1)
def test_0200_basic_conn_test(self):
""" Configure container commands """
for i in range(2):
for addr in [self.pg_interfaces[i].remote_ip4,
self.pg_interfaces[i].remote_ip6]:
self.vapi.ppcli("ip container " + addr + " " +
self.pg_interfaces[i].name)
self.vapi.ppcli("stn rule address " + addr +
" interface " + self.pg_interfaces[i].name)
def test_0210_basic_conn_test(self):
""" IPv4 test after configuring container """
self.run_basic_conn_test(AF_INET, 0)
def test_0211_basic_conn_test(self):
""" IPv6 test after configuring container """
self.run_basic_conn_test(AF_INET, 1)
def test_0300_unconfigure_commands(self):
""" Unconfigure container commands """
for i in range(2):
for addr in [self.pg_interfaces[i].remote_ip4,
self.pg_interfaces[i].remote_ip6]:
self.vapi.ppcli("ip container " + addr + " " +
self.pg_interfaces[i].name +
" del")
self.vapi.ppcli("stn rule address " + addr +
" interface " + self.pg_interfaces[i].name +
" del")
def test_0410_spoof_test(self):
""" IPv4 local-spoof after unconfig test """
self.run_negative_conn_test(AF_INET, 0)
def test_0411_spoof_test(self):
""" IPv6 local-spoof after unconfig test """
self.run_negative_conn_test(AF_INET, 1)