#!/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)
address and check NAT address pool self.pg7.config_ip4() adresses = self.vapi.nat44_address_dump() self.assertEqual(1, len(adresses)) self.assertEqual(adresses[0].ip_address[0:4], self.pg7.local_ip4n) # remove interface address and check NAT address pool self.pg7.unconfig_ip4() adresses = self.vapi.nat44_address_dump() self.assertEqual(0, len(adresses)) def test_interface_addr_static_mapping(self): """ Static mapping with addresses from interface """ self.vapi.nat44_add_interface_addr(self.pg7.sw_if_index) self.nat44_add_static_mapping( '1.2.3.4', external_sw_if_index=self.pg7.sw_if_index) # static mappings with external interface static_mappings = self.vapi.nat44_static_mapping_dump() self.assertEqual(1, len(static_mappings)) self.assertEqual(self.pg7.sw_if_index, static_mappings[0].external_sw_if_index) # configure interface address and check static mappings self.pg7.config_ip4() static_mappings = self.vapi.nat44_static_mapping_dump() self.assertEqual(1, len(static_mappings)) self.assertEqual(static_mappings[0].external_ip_address[0:4], self.pg7.local_ip4n) self.assertEqual(0xFFFFFFFF, static_mappings[0].external_sw_if_index) # remove interface address and check static mappings self.pg7.unconfig_ip4() static_mappings = self.vapi.nat44_static_mapping_dump() self.assertEqual(0, len(static_mappings)) def test_ipfix_nat44_sess(self): """ IPFIX logging NAT44 session created/delted """ self.ipfix_domain_id = 10 self.ipfix_src_port = 20202 colector_port = 30303 bind_layers(UDP, IPFIX, dport=30303) self.nat44_add_address(self.nat_addr) self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index) self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index, is_inside=0) self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4n, src_address=self.pg3.local_ip4n, path_mtu=512, template_interval=10, collector_port=colector_port) self.vapi.nat_ipfix(domain_id=self.ipfix_domain_id, src_port=self.ipfix_src_port) pkts = self.create_stream_in(self.pg0, self.pg1) self.pg0.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg1.get_capture(len(pkts)) self.verify_capture_out(capture) self.nat44_add_address(self.nat_addr, is_add=0) self.vapi.cli("ipfix flush") # FIXME this should be an API call capture = self.pg3.get_capture(3) ipfix = IPFIXDecoder() # first load template for p in capture: self.assertTrue(p.haslayer(IPFIX)) self.assertEqual(p[IP].src, self.pg3.local_ip4) self.assertEqual(p[IP].dst, self.pg3.remote_ip4) self.assertEqual(p[UDP].sport, self.ipfix_src_port) self.assertEqual(p[UDP].dport, colector_port) self.assertEqual(p[IPFIX].observationDomainID, self.ipfix_domain_id) if p.haslayer(Template): ipfix.add_template(p.getlayer(Template)) # verify events in data set for p in capture: if p.haslayer(Data): data = ipfix.decode_data_set(p.getlayer(Set)) self.verify_ipfix_nat44_ses(data) def test_ipfix_addr_exhausted(self): """ IPFIX logging NAT addresses exhausted """ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index) self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index, is_inside=0) self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4n, src_address=self.pg3.local_ip4n, path_mtu=512, template_interval=10) self.vapi.nat_ipfix(domain_id=self.ipfix_domain_id, src_port=self.ipfix_src_port) p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / TCP(sport=3025)) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg1.get_capture(0) self.vapi.cli("ipfix flush") # FIXME this should be an API call capture = self.pg3.get_capture(3) ipfix = IPFIXDecoder() # first load template for p in capture: self.assertTrue(p.haslayer(IPFIX)) self.assertEqual(p[IP].src, self.pg3.local_ip4) self.assertEqual(p[IP].dst, self.pg3.remote_ip4) self.assertEqual(p[UDP].sport, self.ipfix_src_port) self.assertEqual(p[UDP].dport, 4739) self.assertEqual(p[IPFIX].observationDomainID, self.ipfix_domain_id) if p.haslayer(Template): ipfix.add_template(p.getlayer(Template)) # verify events in data set for p in capture: if p.haslayer(Data): data = ipfix.decode_data_set(p.getlayer(Set)) self.verify_ipfix_addr_exhausted(data) def test_pool_addr_fib(self): """ NAT44 add pool addresses to FIB """ static_addr = '10.0.0.10' self.nat44_add_address(self.nat_addr) self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index) self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index, is_inside=0) self.nat44_add_static_mapping(self.pg0.remote_ip4, static_addr) # NAT44 address p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') / ARP(op=ARP.who_has, pdst=self.nat_addr, psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac)) self.pg1.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg1.get_capture(1) self.assertTrue(capture[0].haslayer(ARP)) self.assertTrue(capture[0][ARP].op, ARP.is_at) # 1:1 NAT address p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') / ARP(op=ARP.who_has, pdst=static_addr, psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac)) self.pg1.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg1.get_capture(1) self.assertTrue(capture[0].haslayer(ARP)) self.assertTrue(capture[0][ARP].op, ARP.is_at) # send ARP to non-NAT44 interface p = (Ether(src=self.pg2.remote_mac, dst='ff:ff:ff:ff:ff:ff') / ARP(op=ARP.who_has, pdst=self.nat_addr, psrc=self.pg2.remote_ip4, hwsrc=self.pg2.remote_mac)) self.pg2.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg1.get_capture(0) # remove addresses and verify self.nat44_add_address(self.nat_addr, is_add=0) self.nat44_add_static_mapping(self.pg0.remote_ip4, static_addr, is_add=0) p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') / ARP(op=ARP.who_has, pdst=self.nat_addr, psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac)) self.pg1.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg1.get_capture(0) p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') / ARP(op=ARP.who_has, pdst=static_addr, psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac)) self.pg1.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg1.get_capture(0) def test_vrf_mode(self): """ NAT44 tenant VRF aware address pool mode """ vrf_id1 = 1 vrf_id2 = 2 nat_ip1 = "10.0.0.10" nat_ip2 = "10.0.0.11" self.pg0.unconfig_ip4() self.pg1.unconfig_ip4() self.pg0.set_table_ip4(vrf_id1) self.pg1.set_table_ip4(vrf_id2) self.pg0.config_ip4() self.pg1.config_ip4() self.nat44_add_address(nat_ip1, vrf_id=vrf_id1) self.nat44_add_address(nat_ip2, vrf_id=vrf_id2) self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index) self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index) self.vapi.nat44_interface_add_del_feature(self.pg2.sw_if_index, is_inside=0) # first VRF pkts = self.create_stream_in(self.pg0, self.pg2) self.pg0.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg2.get_capture(len(pkts)) self.verify_capture_out(capture, nat_ip1) # second VRF pkts = self.create_stream_in(self.pg1, self.pg2) self.pg1.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg2.get_capture(len(pkts)) self.verify_capture_out(capture, nat_ip2) def test_vrf_feature_independent(self): """ NAT44 tenant VRF independent address pool mode """ nat_ip1 = "10.0.0.10" nat_ip2 = "10.0.0.11" self.nat44_add_address(nat_ip1) self.nat44_add_address(nat_ip2) self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index) self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index) self.vapi.nat44_interface_add_del_feature(self.pg2.sw_if_index, is_inside=0) # first VRF pkts = self.create_stream_in(self.pg0, self.pg2) self.pg0.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg2.get_capture(len(pkts)) self.verify_capture_out(capture, nat_ip1) # second VRF pkts = self.create_stream_in(self.pg1, self.pg2) self.pg1.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg2.get_capture(len(pkts)) self.verify_capture_out(capture, nat_ip1) def test_dynamic_ipless_interfaces(self): """ NAT44 interfaces without configured IP address """ self.vapi.ip_neighbor_add_del(self.pg7.sw_if_index, self.pg7.remote_mac, self.pg7.remote_ip4n, is_static=1) self.vapi.ip_neighbor_add_del(self.pg8.sw_if_index, self.pg8.remote_mac, self.pg8.remote_ip4n, is_static=1) self.vapi.ip_add_del_route(dst_address=self.pg7.remote_ip4n, dst_address_length=32, next_hop_address=self.pg7.remote_ip4n, next_hop_sw_if_index=self.pg7.sw_if_index) self.vapi.ip_add_del_route(dst_address=self.pg8.remote_ip4n, dst_address_length=32, next_hop_address=self.pg8.remote_ip4n, next_hop_sw_if_index=self.pg8.sw_if_index) self.nat44_add_address(self.nat_addr) self.vapi.nat44_interface_add_del_feature(self.pg7.sw_if_index) self.vapi.nat44_interface_add_del_feature(self.pg8.sw_if_index, is_inside=0) # in2out pkts = self.create_stream_in(self.pg7, self.pg8) self.pg7.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg8.get_capture(len(pkts)) self.verify_capture_out(capture) # out2in pkts = self.create_stream_out(self.pg8, self.nat_addr) self.pg8.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg7.get_capture(len(pkts)) self.verify_capture_in(capture, self.pg7) def test_static_ipless_interfaces(self): """ NAT44 interfaces without configured IP address - 1:1 NAT """ self.vapi.ip_neighbor_add_del(self.pg7.sw_if_index, self.pg7.remote_mac, self.pg7.remote_ip4n, is_static=1) self.vapi.ip_neighbor_add_del(self.pg8.sw_if_index, self.pg8.remote_mac, self.pg8.remote_ip4n, is_static=1) self.vapi.ip_add_del_route(dst_address=self.pg7.remote_ip4n, dst_address_length=32, next_hop_address=self.pg7.remote_ip4n, next_hop_sw_if_index=self.pg7.sw_if_index) self.vapi.ip_add_del_route(dst_address=self.pg8.remote_ip4n, dst_address_length=32, next_hop_address=self.pg8.remote_ip4n, next_hop_sw_if_index=self.pg8.sw_if_index) self.nat44_add_static_mapping(self.pg7.remote_ip4, self.nat_addr) self.vapi.nat44_interface_add_del_feature(self.pg7.sw_if_index) self.vapi.nat44_interface_add_del_feature(self.pg8.sw_if_index, is_inside=0) # out2in pkts = self.create_stream_out(self.pg8) self.pg8.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg7.get_capture(len(pkts)) self.verify_capture_in(capture, self.pg7) # in2out pkts = self.create_stream_in(self.pg7, self.pg8) self.pg7.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg8.get_capture(len(pkts)) self.verify_capture_out(capture, self.nat_addr, True) def test_static_with_port_ipless_interfaces(self): """ NAT44 interfaces without configured IP address - 1:1 NAPT """ self.tcp_port_out = 30606 self.udp_port_out = 30607 self.icmp_id_out = 30608 self.vapi.ip_neighbor_add_del(self.pg7.sw_if_index, self.pg7.remote_mac, self.pg7.remote_ip4n, is_static=1) self.vapi.ip_neighbor_add_del(self.pg8.sw_if_index, self.pg8.remote_mac, self.pg8.remote_ip4n, is_static=1) self.vapi.ip_add_del_route(dst_address=self.pg7.remote_ip4n, dst_address_length=32, next_hop_address=self.pg7.remote_ip4n, next_hop_sw_if_index=self.pg7.sw_if_index) self.vapi.ip_add_del_route(dst_address=self.pg8.remote_ip4n, dst_address_length=32, next_hop_address=self.pg8.remote_ip4n, next_hop_sw_if_index=self.pg8.sw_if_index) self.nat44_add_address(self.nat_addr) self.nat44_add_static_mapping(self.pg7.remote_ip4, self.nat_addr, self.tcp_port_in, self.tcp_port_out, proto=IP_PROTOS.tcp) self.nat44_add_static_mapping(self.pg7.remote_ip4, self.nat_addr, self.udp_port_in, self.udp_port_out, proto=IP_PROTOS.udp) self.nat44_add_static_mapping(self.pg7.remote_ip4, self.nat_addr, self.icmp_id_in, self.icmp_id_out, proto=IP_PROTOS.icmp) self.vapi.nat44_interface_add_del_feature(self.pg7.sw_if_index) self.vapi.nat44_interface_add_del_feature(self.pg8.sw_if_index, is_inside=0) # out2in pkts = self.create_stream_out(self.pg8) self.pg8.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg7.get_capture(len(pkts)) self.verify_capture_in(capture, self.pg7) # in2out pkts = self.create_stream_in(self.pg7, self.pg8) self.pg7.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg8.get_capture(len(pkts)) self.verify_capture_out(capture) def test_static_unknown_proto(self): """ 1:1 NAT translate packet with unknown protocol """ nat_ip = "10.0.0.10" self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip) self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index) self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index, is_inside=0) # in2out p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / GRE() / IP(src=self.pg2.remote_ip4, dst=self.pg3.remote_ip4) / TCP(sport=1234, dport=1234)) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() p = self.pg1.get_capture(1) packet = p[0] try: self.assertEqual(packet[IP].src, nat_ip) self.assertEqual(packet[IP].dst, self.pg1.remote_ip4) self.assertTrue(packet.haslayer(GRE)) self.check_ip_checksum(packet) except: self.logger.error(ppp("Unexpected or invalid packet:", packet)) raise # out2in p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / IP(src=self.pg1.remote_ip4, dst=nat_ip) / GRE() / IP(src=self.pg3.remote_ip4, dst=self.pg2.remote_ip4) / TCP(sport=1234, dport=1234)) self.pg1.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() p = self.pg0.get_capture(1) packet = p[0] try: self.assertEqual(packet[IP].src, self.pg1.remote_ip4) self.assertEqual(packet[IP].dst, self.pg0.remote_ip4) self.assertTrue(packet.haslayer(GRE)) self.check_ip_checksum(packet) except: self.logger.error(ppp("Unexpected or invalid packet:", packet)) raise def test_hairpinning_static_unknown_proto(self): """ 1:1 NAT translate packet with unknown protocol - hairpinning """ host = self.pg0.remote_hosts[0] server = self.pg0.remote_hosts[1] host_nat_ip = "10.0.0.10" server_nat_ip = "10.0.0.11" self.nat44_add_static_mapping(host.ip4, host_nat_ip) self.nat44_add_static_mapping(server.ip4, server_nat_ip) self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index) self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index, is_inside=0) # host to server p = (Ether(dst=self.pg0.local_mac, src=host.mac) / IP(src=host.ip4, dst=server_nat_ip) / GRE() / IP(src=self.pg2.remote_ip4, dst=self.pg3.remote_ip4) / TCP(sport=1234, dport=1234)) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() p = self.pg0.get_capture(1) packet = p[0] try: self.assertEqual(packet[IP].src, host_nat_ip) self.assertEqual(packet[IP].dst, server.ip4) self.assertTrue(packet.haslayer(GRE)) self.check_ip_checksum(packet) except: self.logger.error(ppp("Unexpected or invalid packet:", packet)) raise # server to host p = (Ether(dst=self.pg0.local_mac, src=server.mac) / IP(src=server.ip4, dst=host_nat_ip) / GRE() / IP(src=self.pg3.remote_ip4, dst=self.pg2.remote_ip4) / TCP(sport=1234, dport=1234)) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() p = self.pg0.get_capture(1) packet = p[0] try: self.assertEqual(packet[IP].src, server_nat_ip) self.assertEqual(packet[IP].dst, host.ip4) self.assertTrue(packet.haslayer(GRE)) self.check_ip_checksum(packet) except: self.logger.error(ppp("Unexpected or invalid packet:", packet)) raise def test_unknown_proto(self): """ NAT44 translate packet with unknown protocol """ self.nat44_add_address(self.nat_addr) self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index) self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index, is_inside=0) # in2out p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / TCP(sport=self.tcp_port_in, dport=20)) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() p = self.pg1.get_capture(1) p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / GRE() / IP(src=self.pg2.remote_ip4, dst=self.pg3.remote_ip4) / TCP(sport=1234, dport=1234)) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() p = self.pg1.get_capture(1) packet = p[0] try: self.assertEqual(packet[IP].src, self.nat_addr) self.assertEqual(packet[IP].dst, self.pg1.remote_ip4) self.assertTrue(packet.haslayer(GRE)) self.check_ip_checksum(packet) except: self.logger.error(ppp("Unexpected or invalid packet:", packet)) raise # out2in p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / GRE() / IP(src=self.pg3.remote_ip4, dst=self.pg2.remote_ip4) / TCP(sport=1234, dport=1234)) self.pg1.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() p = self.pg0.get_capture(1) packet = p[0] try: self.assertEqual(packet[IP].src, self.pg1.remote_ip4) self.assertEqual(packet[IP].dst, self.pg0.remote_ip4) self.assertTrue(packet.haslayer(GRE)) self.check_ip_checksum(packet) except: self.logger.error(ppp("Unexpected or invalid packet:", packet)) raise def test_hairpinning_unknown_proto(self): """ NAT44 translate packet with unknown protocol - hairpinning """ host = self.pg0.remote_hosts[0] server = self.pg0.remote_hosts[1] host_in_port = 1234 host_out_port = 0 server_in_port = 5678 server_out_port = 8765 server_nat_ip = "10.0.0.11" self.nat44_add_address(self.nat_addr) self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index) self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index, is_inside=0) # add static mapping for server self.nat44_add_static_mapping(server.ip4, server_nat_ip) # host to server p = (Ether(src=host.mac, dst=self.pg0.local_mac) / IP(src=host.ip4, dst=server_nat_ip) / TCP(sport=host_in_port, dport=server_out_port)) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg0.get_capture(1) p = (Ether(dst=self.pg0.local_mac, src=host.mac) / IP(src=host.ip4, dst=server_nat_ip) / GRE() / IP(src=self.pg2.remote_ip4, dst=self.pg3.remote_ip4) / TCP(sport=1234, dport=1234)) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() p = self.pg0.get_capture(1) packet = p[0] try: self.assertEqual(packet[IP].src, self.nat_addr) self.assertEqual(packet[IP].dst, server.ip4) self.assertTrue(packet.haslayer(GRE)) self.check_ip_checksum(packet) except: self.logger.error(ppp("Unexpected or invalid packet:", packet)) raise # server to host p = (Ether(dst=self.pg0.local_mac, src=server.mac) / IP(src=server.ip4, dst=self.nat_addr) / GRE() / IP(src=self.pg3.remote_ip4, dst=self.pg2.remote_ip4) / TCP(sport=1234, dport=1234)) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() p = self.pg0.get_capture(1) packet = p[0] try: self.assertEqual(packet[IP].src, server_nat_ip) self.assertEqual(packet[IP].dst, host.ip4) self.assertTrue(packet.haslayer(GRE)) self.check_ip_checksum(packet) except: self.logger.error(ppp("Unexpected or invalid packet:", packet)) raise def test_output_feature(self): """ NAT44 interface output feature (in2out postrouting) """ self.nat44_add_address(self.nat_addr) self.vapi.nat44_interface_add_del_output_feature(self.pg0.sw_if_index) self.vapi.nat44_interface_add_del_output_feature(self.pg1.sw_if_index, is_inside=0) # in2out pkts = self.create_stream_in(self.pg0, self.pg1) self.pg0.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg1.get_capture(len(pkts)) self.verify_capture_out(capture) # out2in pkts = self.create_stream_out(self.pg1) self.pg1.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg0.get_capture(len(pkts)) self.verify_capture_in(capture, self.pg0) def test_output_feature_vrf_aware(self): """ NAT44 interface output feature VRF aware (in2out postrouting) """ nat_ip_vrf10 = "10.0.0.10" nat_ip_vrf20 = "10.0.0.20" self.vapi.ip_add_del_route(dst_address=self.pg3.remote_ip4n, dst_address_length=32, next_hop_address=self.pg3.remote_ip4n, next_hop_sw_if_index=self.pg3.sw_if_index, table_id=10) self.vapi.ip_add_del_route(dst_address=self.pg3.remote_ip4n, dst_address_length=32, next_hop_address=self.pg3.remote_ip4n, next_hop_sw_if_index=self.pg3.sw_if_index, table_id=20) self.nat44_add_address(nat_ip_vrf10, vrf_id=10) self.nat44_add_address(nat_ip_vrf20, vrf_id=20) self.vapi.nat44_interface_add_del_output_feature(self.pg4.sw_if_index) self.vapi.nat44_interface_add_del_output_feature(self.pg6.sw_if_index) self.vapi.nat44_interface_add_del_output_feature(self.pg3.sw_if_index, is_inside=0) # in2out VRF 10 pkts = self.create_stream_in(self.pg4, self.pg3) self.pg4.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg3.get_capture(len(pkts)) self.verify_capture_out(capture, nat_ip=nat_ip_vrf10) # out2in VRF 10 pkts = self.create_stream_out(self.pg3, dst_ip=nat_ip_vrf10) self.pg3.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg4.get_capture(len(pkts)) self.verify_capture_in(capture, self.pg4) # in2out VRF 20 pkts = self.create_stream_in(self.pg6, self.pg3) self.pg6.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg3.get_capture(len(pkts)) self.verify_capture_out(capture, nat_ip=nat_ip_vrf20) # out2in VRF 20 pkts = self.create_stream_out(self.pg3, dst_ip=nat_ip_vrf20) self.pg3.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg6.get_capture(len(pkts)) self.verify_capture_in(capture, self.pg6) def test_output_feature_hairpinning(self): """ NAT44 interface output feature hairpinning (in2out postrouting) """ host = self.pg0.remote_hosts[0] server = self.pg0.remote_hosts[1] host_in_port = 1234 host_out_port = 0 server_in_port = 5678 server_out_port = 8765 self.nat44_add_address(self.nat_addr) self.vapi.nat44_interface_add_del_output_feature(self.pg0.sw_if_index) self.vapi.nat44_interface_add_del_output_feature(self.pg1.sw_if_index, is_inside=0) # add static mapping for server self.nat44_add_static_mapping(server.ip4, self.nat_addr, server_in_port, server_out_port, proto=IP_PROTOS.tcp) # send packet from host to server p = (Ether(src=host.mac, dst=self.pg0.local_mac) / IP(src=host.ip4, dst=self.nat_addr) / TCP(sport=host_in_port, dport=server_out_port)) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg0.get_capture(1) p = capture[0] try: ip = p[IP] tcp = p[TCP] self.assertEqual(ip.src, self.nat_addr) self.assertEqual(ip.dst, server.ip4) self.assertNotEqual(tcp.sport, host_in_port) self.assertEqual(tcp.dport, server_in_port) self.check_tcp_checksum(p) host_out_port = tcp.sport except: self.logger.error(ppp("Unexpected or invalid packet:", p)) raise # send reply from server to host p = (Ether(src=server.mac, dst=self.pg0.local_mac) / IP(src=server.ip4, dst=self.nat_addr) / TCP(sport=server_in_port, dport=host_out_port)) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg0.get_capture(1) p = capture[0] try: ip = p[IP] tcp = p[TCP] self.assertEqual(ip.src, self.nat_addr) self.assertEqual(ip.dst, host.ip4) self.assertEqual(tcp.sport, server_out_port) self.assertEqual(tcp.dport, host_in_port) self.check_tcp_checksum(p) except: self.logger.error(ppp("Unexpected or invalid packet:"), p) raise def tearDown(self): super(TestNAT44, self).tearDown() if not self.vpp_dead: self.logger.info(self.vapi.cli("show nat44 verbose")) self.clear_nat44() class TestDeterministicNAT(MethodHolder): """ Deterministic NAT Test Cases """ @classmethod def setUpConstants(cls): super(TestDeterministicNAT, cls).setUpConstants() cls.vpp_cmdline.extend(["nat", "{", "deterministic", "}"]) @classmethod def setUpClass(cls): super(TestDeterministicNAT, cls).setUpClass() try: cls.tcp_port_in = 6303 cls.tcp_external_port = 6303 cls.udp_port_in = 6304 cls.udp_external_port = 6304 cls.icmp_id_in = 6305 cls.nat_addr = '10.0.0.3' cls.create_pg_interfaces(range(3)) cls.interfaces = list(cls.pg_interfaces) for i in cls.interfaces: i.admin_up() i.config_ip4() i.resolve_arp() cls.pg0.generate_remote_hosts(2) cls.pg0.configure_ipv4_neighbors() except Exception: super(TestDeterministicNAT, cls).tearDownClass() raise 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=self.tcp_external_port)) 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=self.udp_external_port)) 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 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=self.tcp_external_port)) 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=self.udp_external_port)) 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_external_id, type='echo-reply')) pkts.append(p) return pkts def verify_capture_out(self, capture, nat_ip=None, packet_num=3): """ 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) """ if nat_ip is None: nat_ip = self.nat_addr self.assertEqual(packet_num, len(capture)) for packet in capture: try: self.assertEqual(packet[IP].src, nat_ip) if packet.haslayer(TCP): self.tcp_port_out = packet[TCP].sport elif packet.haslayer(UDP): self.udp_port_out = packet[UDP].sport else: self.icmp_external_id = packet[ICMP].id except: self.logger.error(ppp("Unexpected or invalid packet " "(outside network):", packet)) raise def initiate_tcp_session(self, in_if, out_if): """ Initiates TCP session :param in_if: Inside interface :param out_if: Outside interface """ try: # SYN packet in->out p = (Ether(src=in_if.remote_mac, dst=in_if.local_mac) / IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) / TCP(sport=self.tcp_port_in, dport=self.tcp_external_port, flags="S")) in_if.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = out_if.get_capture(1) p = capture[0] self.tcp_port_out = p[TCP].sport # SYN + ACK packet out->in p = (Ether(src=out_if.remote_mac, dst=out_if.local_mac) / IP(src=out_if.remote_ip4, dst=self.nat_addr) / TCP(sport=self.tcp_external_port, dport=self.tcp_port_out, flags="SA")) out_if.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() in_if.get_capture(1) # ACK packet in->out p = (Ether(src=in_if.remote_mac, dst=in_if.local_mac) / IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) / TCP(sport=self.tcp_port_in, dport=self.tcp_external_port, flags="A")) in_if.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() out_if.get_capture(1) except: self.logger.error("TCP 3 way handshake failed") raise def verify_ipfix_max_entries_per_user(self, data): """ Verify IPFIX maximum entries per user exceeded event :param data: Decoded IPFIX data records """ self.assertEqual(1, len(data)) record = data[0] # natEvent self.assertEqual(ord(record[230]), 13) # natQuotaExceededEvent self.assertEqual('\x03\x00\x00\x00', record[466]) # sourceIPv4Address self.assertEqual(self.pg0.remote_ip4n, record[8]) def test_deterministic_mode(self): """ NAT plugin run deterministic mode """ in_addr = '172.16.255.0' out_addr = '172.17.255.50' in_addr_t = '172.16.255.20' in_addr_n = socket.inet_aton(in_addr) out_addr_n = socket.inet_aton(out_addr) in_addr_t_n = socket.inet_aton(in_addr_t) in_plen = 24 out_plen = 32 nat_config = self.vapi.nat_show_config() self.assertEqual(1, nat_config.deterministic) self.vapi.nat_det_add_del_map(in_addr_n, in_plen, out_addr_n, out_plen) rep1 = self.vapi.nat_det_forward(in_addr_t_n) self.assertEqual(rep1.out_addr[:4], out_addr_n) rep2 = self.vapi.nat_det_reverse(out_addr_n, rep1.out_port_hi) self.assertEqual(rep2.in_addr[:4], in_addr_t_n) deterministic_mappings = self.vapi.nat_det_map_dump() self.assertEqual(len(deterministic_mappings), 1) dsm = deterministic_mappings[0] self.assertEqual(in_addr_n, dsm.in_addr[:4]) self.assertEqual(in_plen, dsm.in_plen) self.assertEqual(out_addr_n, dsm.out_addr[:4]) self.assertEqual(out_plen, dsm.out_plen) self.clear_nat_det() deterministic_mappings = self.vapi.nat_det_map_dump() self.assertEqual(len(deterministic_mappings), 0) def test_set_timeouts(self): """ Set deterministic NAT timeouts """ timeouts_before = self.vapi.nat_det_get_timeouts() self.vapi.nat_det_set_timeouts(timeouts_before.udp + 10, timeouts_before.tcp_established + 10, timeouts_before.tcp_transitory + 10, timeouts_before.icmp + 10) timeouts_after = self.vapi.nat_det_get_timeouts() self.assertNotEqual(timeouts_before.udp, timeouts_after.udp) self.assertNotEqual(timeouts_before.icmp, timeouts_after.icmp) self.assertNotEqual(timeouts_before.tcp_established, timeouts_after.tcp_established) self.assertNotEqual(timeouts_before.tcp_transitory, timeouts_after.tcp_transitory) def test_det_in(self): """ Deterministic NAT translation test (TCP, UDP, ICMP) """ nat_ip = "10.0.0.10" self.vapi.nat_det_add_del_map(self.pg0.remote_ip4n, 32, socket.inet_aton(nat_ip), 32) self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index) self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index, is_inside=0) # in2out pkts = self.create_stream_in(self.pg0, self.pg1) self.pg0.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg1.get_capture(len(pkts)) self.verify_capture_out(capture, nat_ip) # out2in pkts = self.create_stream_out(self.pg1, nat_ip) self.pg1.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg0.get_capture(len(pkts)) self.verify_capture_in(capture, self.pg0) # session dump test sessions = self.vapi.nat_det_session_dump(self.pg0.remote_ip4n) self.assertEqual(len(sessions), 3) # TCP session s = sessions[0] self.assertEqual(s.ext_addr[:4], self.pg1.remote_ip4n) self.assertEqual(s.in_port, self.tcp_port_in) self.assertEqual(s.out_port, self.tcp_port_out) self.assertEqual(s.ext_port, self.tcp_external_port) # UDP session s = sessions[1] self.assertEqual(s.ext_addr[:4], self.pg1.remote_ip4n) self.assertEqual(s.in_port, self.udp_port_in) self.assertEqual(s.out_port, self.udp_port_out) self.assertEqual(s.ext_port, self.udp_external_port) # ICMP session s = sessions[2] self.assertEqual(s.ext_addr[:4], self.pg1.remote_ip4n) self.assertEqual(s.in_port, self.icmp_id_in) self.assertEqual(s.out_port, self.icmp_external_id) def test_multiple_users(self): """ Deterministic NAT multiple users """ nat_ip = "10.0.0.10" port_in = 80 external_port = 6303 host0 = self.pg0.remote_hosts[0] host1 = self.pg0.remote_hosts[1] self.vapi.nat_det_add_del_map(host0.ip4n, 24, socket.inet_aton(nat_ip), 32) self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index) self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index, is_inside=0) # host0 to out p = (Ether(src=host0.mac, dst=self.pg0.local_mac) / IP(src=host0.ip4, dst=self.pg1.remote_ip4) / TCP(sport=port_in, dport=external_port)) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg1.get_capture(1) p = capture[0] try: ip = p[IP] tcp = p[TCP] self.assertEqual(ip.src, nat_ip) self.assertEqual(ip.dst, self.pg1.remote_ip4) self.assertEqual(tcp.dport, external_port) port_out0 = tcp.sport except: self.logger.error(ppp("Unexpected or invalid packet:", p)) raise # host1 to out p = (Ether(src=host1.mac, dst=self.pg0.local_mac) / IP(src=host1.ip4, dst=self.pg1.remote_ip4) / TCP(sport=port_in, dport=external_port)) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg1.get_capture(1) p = capture[0] try: ip = p[IP] tcp = p[TCP] self.assertEqual(ip.src, nat_ip) self.assertEqual(ip.dst, self.pg1.remote_ip4) self.assertEqual(tcp.dport, external_port) port_out1 = tcp.sport except: self.logger.error(ppp("Unexpected or invalid packet:", p)) raise dms = self.vapi.nat_det_map_dump() self.assertEqual(1, len(dms)) self.assertEqual(2, dms[0].ses_num) # out to host0 p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / IP(src=self.pg1.remote_ip4, dst=nat_ip) / TCP(sport=external_port, dport=port_out0)) self.pg1.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg0.get_capture(1) p = capture[0] try: ip = p[IP] tcp = p[TCP] self.assertEqual(ip.src, self.pg1.remote_ip4) self.assertEqual(ip.dst, host0.ip4) self.assertEqual(tcp.dport, port_in) self.assertEqual(tcp.sport, external_port) except: self.logger.error(ppp("Unexpected or invalid packet:", p)) raise # out to host1 p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / IP(src=self.pg1.remote_ip4, dst=nat_ip) / TCP(sport=external_port, dport=port_out1)) self.pg1.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg0.get_capture(1) p = capture[0] try: ip = p[IP] tcp = p[TCP] self.assertEqual(ip.src, self.pg1.remote_ip4) self.assertEqual(ip.dst, host1.ip4) self.assertEqual(tcp.dport, port_in) self.assertEqual(tcp.sport, external_port) except: self.logger.error(ppp("Unexpected or invalid packet", p)) raise # session close api test self.vapi.nat_det_close_session_out(socket.inet_aton(nat_ip), port_out1, self.pg1.remote_ip4n, external_port) dms = self.vapi.nat_det_map_dump() self.assertEqual(dms[0].ses_num, 1) self.vapi.nat_det_close_session_in(host0.ip4n, port_in, self.pg1.remote_ip4n, external_port) dms = self.vapi.nat_det_map_dump() self.assertEqual(dms[0].ses_num, 0) def test_tcp_session_close_detection_in(self): """ Deterministic NAT TCP session close from inside network """ self.vapi.nat_det_add_del_map(self.pg0.remote_ip4n, 32, socket.inet_aton(self.nat_addr), 32) self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index) self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index, is_inside=0) self.initiate_tcp_session(self.pg0, self.pg1) # close the session from inside try: # FIN packet in -> out p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / TCP(sport=self.tcp_port_in, dport=self.tcp_external_port, flags="F")) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() self.pg1.get_capture(1) pkts = [] # ACK packet out -> in p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / TCP(sport=self.tcp_external_port, dport=self.tcp_port_out, flags="A")) pkts.append(p) # FIN packet out -> in p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / TCP(sport=self.tcp_external_port, dport=self.tcp_port_out, flags="F")) pkts.append(p) self.pg1.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() self.pg0.get_capture(2) # ACK packet in -> out p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / TCP(sport=self.tcp_port_in, dport=self.tcp_external_port, flags="A")) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() self.pg1.get_capture(1) # Check if deterministic NAT44 closed the session dms = self.vapi.nat_det_map_dump() self.assertEqual(0, dms[0].ses_num) except: self.logger.error("TCP session termination failed") raise def test_tcp_session_close_detection_out(self): """ Deterministic NAT TCP session close from outside network """ self.vapi.nat_det_add_del_map(self.pg0.remote_ip4n, 32, socket.inet_aton(self.nat_addr), 32) self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index) self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index, is_inside=0) self.initiate_tcp_session(self.pg0, self.pg1) # close the session from outside try: # FIN packet out -> in p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / TCP(sport=self.tcp_external_port, dport=self.tcp_port_out, flags="F")) self.pg1.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() self.pg0.get_capture(1) pkts = [] # ACK packet in -> out p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / TCP(sport=self.tcp_port_in, dport=self.tcp_external_port, flags="A")) pkts.append(p) # ACK packet in -> out p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / TCP(sport=self.tcp_port_in, dport=self.tcp_external_port, flags="F")) pkts.append(p) self.pg0.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() self.pg1.get_capture(2) # ACK packet out -> in p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / TCP(sport=self.tcp_external_port, dport=self.tcp_port_out, flags="A")) self.pg1.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() self.pg0.get_capture(1) # Check if deterministic NAT44 closed the session dms = self.vapi.nat_det_map_dump() self.assertEqual(0, dms[0].ses_num) except: self.logger.error("TCP session termination failed") raise @unittest.skipUnless(running_extended_tests(), "part of extended tests") def test_session_timeout(self): """ Deterministic NAT session timeouts """ self.vapi.nat_det_add_del_map(self.pg0.remote_ip4n, 32, socket.inet_aton(self.nat_addr), 32) self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index) self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index, is_inside=0) self.initiate_tcp_session(self.pg0, self.pg1) self.vapi.nat_det_set_timeouts(5, 5, 5, 5) pkts = self.create_stream_in(self.pg0, self.pg1) self.pg0.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg1.get_capture(len(pkts)) sleep(15) dms = self.vapi.nat_det_map_dump() self.assertEqual(0, dms[0].ses_num) @unittest.skipUnless(running_extended_tests(), "part of extended tests") def test_session_limit_per_user(self): """ Deterministic NAT maximum sessions per user limit """ self.vapi.nat_det_add_del_map(self.pg0.remote_ip4n, 32, socket.inet_aton(self.nat_addr), 32) self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index) self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index, is_inside=0) self.vapi.set_ipfix_exporter(collector_address=self.pg2.remote_ip4n, src_address=self.pg2.local_ip4n, path_mtu=512, template_interval=10) self.vapi.nat_ipfix() pkts = [] for port in range(1025, 2025): p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / UDP(sport=port, dport=port)) pkts.append(p) self.pg0.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg1.get_capture(len(pkts)) p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / UDP(sport=3001, dport=3002)) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg1.assert_nothing_captured() # verify ICMP error packet capture = self.pg0.get_capture(1) p = capture[0] self.assertTrue(p.haslayer(ICMP)) icmp = p[ICMP] self.assertEqual(icmp.type, 3) self.assertEqual(icmp.code, 1) self.assertTrue(icmp.haslayer(IPerror)) inner_ip = icmp[IPerror] self.assertEqual(inner_ip[UDPerror].sport, 3001) self.assertEqual(inner_ip[UDPerror].dport, 3002) dms = self.vapi.nat_det_map_dump() self.assertEqual(1000, dms[0].ses_num) # verify IPFIX logging self.vapi.cli("ipfix flush") # FIXME this should be an API call sleep(1) capture = self.pg2.get_capture(2) ipfix = IPFIXDecoder() # first load template for p in capture: self.assertTrue(p.haslayer(IPFIX)) if p.haslayer(Template): ipfix.add_template(p.getlayer(Template)) # verify events in data set for p in capture: if p.haslayer(Data): data = ipfix.decode_data_set(p.getlayer(Set)) self.verify_ipfix_max_entries_per_user(data) def clear_nat_det(self): """ Clear deterministic NAT configuration. """ self.vapi.nat_ipfix(enable=0) self.vapi.nat_det_set_timeouts() deterministic_mappings = self.vapi.nat_det_map_dump() for dsm in deterministic_mappings: self.vapi.nat_det_add_del_map(dsm.in_addr, dsm.in_plen, dsm.out_addr, dsm.out_plen, is_add=0) interfaces = self.vapi.nat44_interface_dump() for intf in interfaces: self.vapi.nat44_interface_add_del_feature(intf.sw_if_index, intf.is_inside, is_add=0) def tearDown(self): super(TestDeterministicNAT, self).tearDown() if not self.vpp_dead: self.logger.info(self.vapi.cli("show nat44 detail")) self.clear_nat_det() class TestNAT64(MethodHolder): """ NAT64 Test Cases """ @classmethod def setUpClass(cls): super(TestNAT64, 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.nat_addr_n = socket.inet_pton(socket.AF_INET, cls.nat_addr) cls.vrf1_id = 10 cls.vrf1_nat_addr = '10.0.10.3' cls.vrf1_nat_addr_n = socket.inet_pton(socket.AF_INET, cls.vrf1_nat_addr) cls.create_pg_interfaces(range(3)) cls.ip6_interfaces = list(cls.pg_interfaces[0:1]) cls.ip6_interfaces.append(cls.pg_interfaces[2]) cls.ip4_interfaces = list(cls.pg_interfaces[1:2]) cls.pg_interfaces[2].set_table_ip6(cls.vrf1_id) cls.pg0.generate_remote_hosts(2) for i in cls.ip6_interfaces: i.admin_up() i.config_ip6() i.configure_ipv6_neighbors() for i in cls.ip4_interfaces: i.admin_up() i.config_ip4() i.resolve_arp() except Exception: super(TestNAT64, cls).tearDownClass() raise def test_pool(self): """ Add/delete address to NAT64 pool """ nat_addr = socket.inet_pton(socket.AF_INET, '1.2.3.4') self.vapi.nat64_add_del_pool_addr_range(nat_addr, nat_addr) addresses = self.vapi.nat64_pool_addr_dump() self.assertEqual(len(addresses), 1) self.assertEqual(addresses[0].address, nat_addr) self.vapi.nat64_add_del_pool_addr_range(nat_addr, nat_addr, is_add=0) addresses = self.vapi.nat64_pool_addr_dump() self.assertEqual(len(addresses), 0) def test_interface(self): """ Enable/disable NAT64 feature on the interface """ self.vapi.nat64_add_del_interface(self.pg0.sw_if_index) self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0) interfaces = self.vapi.nat64_interface_dump() self.assertEqual(len(interfaces), 2) pg0_found = False pg1_found = False for intf in interfaces: if intf.sw_if_index == self.pg0.sw_if_index: self.assertEqual(intf.is_inside, 1) pg0_found = True elif intf.sw_if_index == self.pg1.sw_if_index: self.assertEqual(intf.is_inside, 0) pg1_found = True self.assertTrue(pg0_found) self.assertTrue(pg1_found) features = self.vapi.cli("show interface features pg0") self.assertNotEqual(features.find('nat64-in2out'), -1) features = self.vapi.cli("show interface features pg1") self.assertNotEqual(features.find('nat64-out2in'), -1) self.vapi.nat64_add_del_interface(self.pg0.sw_if_index, is_add=0) self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_add=0) interfaces = self.vapi.nat64_interface_dump() self.assertEqual(len(interfaces), 0) def test_static_bib(self): """ Add/delete static BIB entry """ in_addr = socket.inet_pton(socket.AF_INET6, '2001:db8:85a3::8a2e:370:7334') out_addr = socket.inet_pton(socket.AF_INET, '10.1.1.3') in_port = 1234 out_port = 5678 proto = IP_PROTOS.tcp self.vapi.nat64_add_del_static_bib(in_addr, out_addr, in_port, out_port, proto) bib = self.vapi.nat64_bib_dump(IP_PROTOS.tcp) static_bib_num = 0 for bibe in bib: if bibe.is_static: static_bib_num += 1 self.assertEqual(bibe.i_addr, in_addr) self.assertEqual(bibe.o_addr, out_addr) self.assertEqual(bibe.i_port, in_port) self.assertEqual(bibe.o_port, out_port) self.assertEqual(static_bib_num, 1) self.vapi.nat64_add_del_static_bib(in_addr, out_addr, in_port, out_port, proto, is_add=0) bib = self.vapi.nat64_bib_dump(IP_PROTOS.tcp) static_bib_num = 0 for bibe in bib: if bibe.is_static: static_bib_num += 1 self.assertEqual(static_bib_num, 0) def test_set_timeouts(self): """ Set NAT64 timeouts """ # verify default values timeouts = self.vapi.nat64_get_timeouts() self.assertEqual(timeouts.udp, 300) self.assertEqual(timeouts.icmp, 60) self.assertEqual(timeouts.tcp_trans, 240) self.assertEqual(timeouts.tcp_est, 7440) self.assertEqual(timeouts.tcp_incoming_syn, 6) # set and verify custom values self.vapi.nat64_set_timeouts(udp=200, icmp=30, tcp_trans=250, tcp_est=7450, tcp_incoming_syn=10) timeouts = self.vapi.nat64_get_timeouts() self.assertEqual(timeouts.udp, 200) self.assertEqual(timeouts.icmp, 30) self.assertEqual(timeouts.tcp_trans, 250) self.assertEqual(timeouts.tcp_est, 7450) self.assertEqual(timeouts.tcp_incoming_syn, 10) def test_dynamic(self): """ NAT64 dynamic translation test """ self.tcp_port_in = 6303 self.udp_port_in = 6304 self.icmp_id_in = 6305 ses_num_start = self.nat64_get_ses_num() self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n, self.nat_addr_n) self.vapi.nat64_add_del_interface(self.pg0.sw_if_index) self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0) # in2out pkts = self.create_stream_in_ip6(self.pg0, self.pg1) self.pg0.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg1.get_capture(len(pkts)) self.verify_capture_out(capture, nat_ip=self.nat_addr, dst_ip=self.pg1.remote_ip4) # out2in pkts = self.create_stream_out(self.pg1, dst_ip=self.nat_addr) self.pg1.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg0.get_capture(len(pkts)) ip = IPv6(src=''.join(['64:ff9b::', self.pg1.remote_ip4])) self.verify_capture_in_ip6(capture, ip[IPv6].src, self.pg0.remote_ip6) # in2out pkts = self.create_stream_in_ip6(self.pg0, self.pg1) self.pg0.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg1.get_capture(len(pkts)) self.verify_capture_out(capture, nat_ip=self.nat_addr, dst_ip=self.pg1.remote_ip4) # out2in pkts = self.create_stream_out(self.pg1, dst_ip=self.nat_addr) self.pg1.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg0.get_capture(len(pkts)) self.verify_capture_in_ip6(capture, ip[IPv6].src, self.pg0.remote_ip6) ses_num_end = self.nat64_get_ses_num() self.assertEqual(ses_num_end - ses_num_start, 3) # tenant with specific VRF self.vapi.nat64_add_del_pool_addr_range(self.vrf1_nat_addr_n, self.vrf1_nat_addr_n, vrf_id=self.vrf1_id) self.vapi.nat64_add_del_interface(self.pg2.sw_if_index) pkts = self.create_stream_in_ip6(self.pg2, self.pg1) self.pg2.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg1.get_capture(len(pkts)) self.verify_capture_out(capture, nat_ip=self.vrf1_nat_addr, dst_ip=self.pg1.remote_ip4) pkts = self.create_stream_out(self.pg1, dst_ip=self.vrf1_nat_addr) self.pg1.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg2.get_capture(len(pkts)) self.verify_capture_in_ip6(capture, ip[IPv6].src, self.pg2.remote_ip6) def test_static(self): """ NAT64 static translation test """ self.tcp_port_in = 60303 self.udp_port_in = 60304 self.icmp_id_in = 60305 self.tcp_port_out = 60303 self.udp_port_out = 60304 self.icmp_id_out = 60305 ses_num_start = self.nat64_get_ses_num() self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n, self.nat_addr_n) self.vapi.nat64_add_del_interface(self.pg0.sw_if_index) self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0) self.vapi.nat64_add_del_static_bib(self.pg0.remote_ip6n, self.nat_addr_n, self.tcp_port_in, self.tcp_port_out, IP_PROTOS.tcp) self.vapi.nat64_add_del_static_bib(self.pg0.remote_ip6n, self.nat_addr_n, self.udp_port_in, self.udp_port_out, IP_PROTOS.udp) self.vapi.nat64_add_del_static_bib(self.pg0.remote_ip6n, self.nat_addr_n, self.icmp_id_in, self.icmp_id_out, IP_PROTOS.icmp) # in2out pkts = self.create_stream_in_ip6(self.pg0, self.pg1) self.pg0.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg1.get_capture(len(pkts)) self.verify_capture_out(capture, nat_ip=self.nat_addr, dst_ip=self.pg1.remote_ip4, same_port=True) # out2in pkts = self.create_stream_out(self.pg1, dst_ip=self.nat_addr) self.pg1.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg0.get_capture(len(pkts)) ip = IPv6(src=''.join(['64:ff9b::', self.pg1.remote_ip4])) self.verify_capture_in_ip6(capture, ip[IPv6].src, self.pg0.remote_ip6) ses_num_end = self.nat64_get_ses_num() self.assertEqual(ses_num_end - ses_num_start, 3) @unittest.skipUnless(running_extended_tests(), "part of extended tests") def test_session_timeout(self): """ NAT64 session timeout """ self.icmp_id_in = 1234 self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n, self.nat_addr_n) self.vapi.nat64_add_del_interface(self.pg0.sw_if_index) self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0) self.vapi.nat64_set_timeouts(icmp=5) pkts = self.create_stream_in_ip6(self.pg0, self.pg1) self.pg0.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg1.get_capture(len(pkts)) ses_num_before_timeout = self.nat64_get_ses_num() sleep(15) # ICMP session after timeout ses_num_after_timeout = self.nat64_get_ses_num() self.assertNotEqual(ses_num_before_timeout, ses_num_after_timeout) def test_icmp_error(self): """ NAT64 ICMP Error message translation """ self.tcp_port_in = 6303 self.udp_port_in = 6304 self.icmp_id_in = 6305 ses_num_start = self.nat64_get_ses_num() self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n, self.nat_addr_n) self.vapi.nat64_add_del_interface(self.pg0.sw_if_index) self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0) # send some packets to create sessions pkts = self.create_stream_in_ip6(self.pg0, self.pg1) self.pg0.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture_ip4 = self.pg1.get_capture(len(pkts)) self.verify_capture_out(capture_ip4, nat_ip=self.nat_addr, dst_ip=self.pg1.remote_ip4) pkts = self.create_stream_out(self.pg1, dst_ip=self.nat_addr) self.pg1.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture_ip6 = self.pg0.get_capture(len(pkts)) ip = IPv6(src=''.join(['64:ff9b::', self.pg1.remote_ip4])) self.verify_capture_in_ip6(capture_ip6, ip[IPv6].src, self.pg0.remote_ip6) # in2out pkts = [Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / IPv6(src=self.pg0.remote_ip6, dst=ip[IPv6].src) / ICMPv6DestUnreach(code=1) / packet[IPv6] for packet in capture_ip6] self.pg0.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg1.get_capture(len(pkts)) for packet in capture: try: self.assertEqual(packet[IP].src, self.nat_addr) self.assertEqual(packet[IP].dst, self.pg1.remote_ip4) self.assertEqual(packet[ICMP].type, 3) self.assertEqual(packet[ICMP].code, 13) inner = packet[IPerror] self.assertEqual(inner.src, self.pg1.remote_ip4) self.assertEqual(inner.dst, self.nat_addr) self.check_icmp_checksum(packet) if inner.haslayer(TCPerror): self.assertEqual(inner[TCPerror].dport, self.tcp_port_out) elif inner.haslayer(UDPerror): self.assertEqual(inner[UDPerror].dport, self.udp_port_out) else: self.assertEqual(inner[ICMPerror].id, self.icmp_id_out) except: self.logger.error(ppp("Unexpected or invalid packet:", packet)) raise # out2in pkts = [Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / ICMP(type=3, code=13) / packet[IP] for packet in capture_ip4] self.pg1.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg0.get_capture(len(pkts)) for packet in capture: try: self.assertEqual(packet[IPv6].src, ip.src) self.assertEqual(packet[IPv6].dst, self.pg0.remote_ip6) icmp = packet[ICMPv6DestUnreach] self.assertEqual(icmp.code, 1) inner = icmp[IPerror6] self.assertEqual(inner.src, self.pg0.remote_ip6) self.assertEqual(inner.dst, ip.src) self.check_icmpv6_checksum(packet) if inner.haslayer(TCPerror): self.assertEqual(inner[TCPerror].sport, self.tcp_port_in) elif inner.haslayer(UDPerror): self.assertEqual(inner[UDPerror].sport, self.udp_port_in) else: self.assertEqual(inner[ICMPv6EchoRequest].id, self.icmp_id_in) except: self.logger.error(ppp("Unexpected or invalid packet:", packet)) raise def test_hairpinning(self): """ NAT64 hairpinning """ client = self.pg0.remote_hosts[0] server = self.pg0.remote_hosts[1] server_tcp_in_port = 22 server_tcp_out_port = 4022 server_udp_in_port = 23 server_udp_out_port = 4023 client_tcp_in_port = 1234 client_udp_in_port = 1235 client_tcp_out_port = 0 client_udp_out_port = 0 ip = IPv6(src=''.join(['64:ff9b::', self.nat_addr])) nat_addr_ip6 = ip.src self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n, self.nat_addr_n) self.vapi.nat64_add_del_interface(self.pg0.sw_if_index) self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0) self.vapi.nat64_add_del_static_bib(server.ip6n, self.nat_addr_n, server_tcp_in_port, server_tcp_out_port, IP_PROTOS.tcp) self.vapi.nat64_add_del_static_bib(server.ip6n, self.nat_addr_n, server_udp_in_port, server_udp_out_port, IP_PROTOS.udp) # client to server pkts = [] p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / IPv6(src=client.ip6, dst=nat_addr_ip6) / TCP(sport=client_tcp_in_port, dport=server_tcp_out_port)) pkts.append(p) p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / IPv6(src=client.ip6, dst=nat_addr_ip6) / UDP(sport=client_udp_in_port, dport=server_udp_out_port)) pkts.append(p) self.pg0.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg0.get_capture(len(pkts)) for packet in capture: try: self.assertEqual(packet[IPv6].src, nat_addr_ip6) self.assertEqual(packet[IPv6].dst, server.ip6) if packet.haslayer(TCP): self.assertNotEqual(packet[TCP].sport, client_tcp_in_port) self.assertEqual(packet[TCP].dport, server_tcp_in_port) self.check_tcp_checksum(packet) client_tcp_out_port = packet[TCP].sport else: self.assertNotEqual(packet[UDP].sport, client_udp_in_port) self.assertEqual(packet[UDP].dport, server_udp_in_port) self.check_udp_checksum(packet) client_udp_out_port = packet[UDP].sport except: self.logger.error(ppp("Unexpected or invalid packet:", packet)) raise # server to client pkts = [] p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / IPv6(src=server.ip6, dst=nat_addr_ip6) / TCP(sport=server_tcp_in_port, dport=client_tcp_out_port)) pkts.append(p) p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / IPv6(src=server.ip6, dst=nat_addr_ip6) / UDP(sport=server_udp_in_port, dport=client_udp_out_port)) pkts.append(p) self.pg0.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg0.get_capture(len(pkts)) for packet in capture: try: self.assertEqual(packet[IPv6].src, nat_addr_ip6) self.assertEqual(packet[IPv6].dst, client.ip6) if packet.haslayer(TCP): self.assertEqual(packet[TCP].sport, server_tcp_out_port) self.assertEqual(packet[TCP].dport, client_tcp_in_port) self.check_tcp_checksum(packet) else: self.assertEqual(packet[UDP].sport, server_udp_out_port) self.assertEqual(packet[UDP].dport, client_udp_in_port) self.check_udp_checksum(packet) except: self.logger.error(ppp("Unexpected or invalid packet:", packet)) raise # ICMP error pkts = [] pkts = [Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / IPv6(src=client.ip6, dst=nat_addr_ip6) / ICMPv6DestUnreach(code=1) / packet[IPv6] for packet in capture] self.pg0.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg0.get_capture(len(pkts)) for packet in capture: try: self.assertEqual(packet[IPv6].src, nat_addr_ip6) self.assertEqual(packet[IPv6].dst, server.ip6) icmp = packet[ICMPv6DestUnreach] self.assertEqual(icmp.code, 1) inner = icmp[IPerror6] self.assertEqual(inner.src, server.ip6) self.assertEqual(inner.dst, nat_addr_ip6) self.check_icmpv6_checksum(packet) if inner.haslayer(TCPerror): self.assertEqual(inner[TCPerror].sport, server_tcp_in_port) self.assertEqual(inner[TCPerror].dport, client_tcp_out_port) else: self.assertEqual(inner[UDPerror].sport, server_udp_in_port) self.assertEqual(inner[UDPerror].dport, client_udp_out_port) except: self.logger.error(ppp("Unexpected or invalid packet:", packet)) raise def test_prefix(self): """ NAT64 Network-Specific Prefix """ self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n, self.nat_addr_n) self.vapi.nat64_add_del_interface(self.pg0.sw_if_index) self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0) self.vapi.nat64_add_del_pool_addr_range(self.vrf1_nat_addr_n, self.vrf1_nat_addr_n, vrf_id=self.vrf1_id) self.vapi.nat64_add_del_interface(self.pg2.sw_if_index) # Add global prefix global_pref64 = "2001:db8::" global_pref64_n = socket.inet_pton(socket.AF_INET6, global_pref64) global_pref64_len = 32 self.vapi.nat64_add_del_prefix(global_pref64_n, global_pref64_len) prefix = self.vapi.nat64_prefix_dump() self.assertEqual(len(prefix), 1) self.assertEqual(prefix[0].prefix, global_pref64_n) self.assertEqual(prefix[0].prefix_len, global_pref64_len) self.assertEqual(prefix[0].vrf_id, 0) # Add tenant specific prefix vrf1_pref64 = "2001:db8:122:300::" vrf1_pref64_n = socket.inet_pton(socket.AF_INET6, vrf1_pref64) vrf1_pref64_len = 56 self.vapi.nat64_add_del_prefix(vrf1_pref64_n, vrf1_pref64_len, vrf_id=self.vrf1_id) prefix = self.vapi.nat64_prefix_dump() self.assertEqual(len(prefix), 2) # Global prefix pkts = self.create_stream_in_ip6(self.pg0, self.pg1, pref=global_pref64, plen=global_pref64_len) self.pg0.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg1.get_capture(len(pkts)) self.verify_capture_out(capture, nat_ip=self.nat_addr, dst_ip=self.pg1.remote_ip4) pkts = self.create_stream_out(self.pg1, dst_ip=self.nat_addr) self.pg1.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg0.get_capture(len(pkts)) dst_ip = self.compose_ip6(self.pg1.remote_ip4, global_pref64, global_pref64_len) self.verify_capture_in_ip6(capture, dst_ip, self.pg0.remote_ip6) # Tenant specific prefix pkts = self.create_stream_in_ip6(self.pg2, self.pg1, pref=vrf1_pref64, plen=vrf1_pref64_len) self.pg2.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg1.get_capture(len(pkts)) self.verify_capture_out(capture, nat_ip=self.vrf1_nat_addr, dst_ip=self.pg1.remote_ip4) pkts = self.create_stream_out(self.pg1, dst_ip=self.vrf1_nat_addr) self.pg1.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg2.get_capture(len(pkts)) dst_ip = self.compose_ip6(self.pg1.remote_ip4, vrf1_pref64, vrf1_pref64_len) self.verify_capture_in_ip6(capture, dst_ip, self.pg2.remote_ip6) def test_unknown_proto(self): """ NAT64 translate packet with unknown protocol """ self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n, self.nat_addr_n) self.vapi.nat64_add_del_interface(self.pg0.sw_if_index) self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0) remote_ip6 = self.compose_ip6(self.pg1.remote_ip4, '64:ff9b::', 96) # in2out p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / IPv6(src=self.pg0.remote_ip6, dst=remote_ip6) / TCP(sport=self.tcp_port_in, dport=20)) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() p = self.pg1.get_capture(1) p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / IPv6(src=self.pg0.remote_ip6, dst=remote_ip6, nh=47) / GRE() / IP(src=self.pg2.local_ip4, dst=self.pg2.remote_ip4) / TCP(sport=1234, dport=1234)) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() p = self.pg1.get_capture(1) packet = p[0] try: self.assertEqual(packet[IP].src, self.nat_addr) self.assertEqual(packet[IP].dst, self.pg1.remote_ip4) self.assertTrue(packet.haslayer(GRE)) self.check_ip_checksum(packet) except: self.logger.error(ppp("Unexpected or invalid packet:", packet)) raise # out2in p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / GRE() / IP(src=self.pg2.remote_ip4, dst=self.pg2.local_ip4) / TCP(sport=1234, dport=1234)) self.pg1.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() p = self.pg0.get_capture(1) packet = p[0] try: self.assertEqual(packet[IPv6].src, remote_ip6) self.assertEqual(packet[IPv6].dst, self.pg0.remote_ip6) self.assertEqual(packet[IPv6].nh, 47) except: self.logger.error(ppp("Unexpected or invalid packet:", packet)) raise def test_hairpinning_unknown_proto(self): """ NAT64 translate packet with unknown protocol - hairpinning """ client = self.pg0.remote_hosts[0] server = self.pg0.remote_hosts[1] server_tcp_in_port = 22 server_tcp_out_port = 4022 client_tcp_in_port = 1234 client_tcp_out_port = 1235 server_nat_ip = "10.0.0.100" client_nat_ip = "10.0.0.110" server_nat_ip_n = socket.inet_pton(socket.AF_INET, server_nat_ip) client_nat_ip_n = socket.inet_pton(socket.AF_INET, client_nat_ip) server_nat_ip6 = self.compose_ip6(server_nat_ip, '64:ff9b::', 96) client_nat_ip6 = self.compose_ip6(client_nat_ip, '64:ff9b::', 96) self.vapi.nat64_add_del_pool_addr_range(server_nat_ip_n, client_nat_ip_n) self.vapi.nat64_add_del_interface(self.pg0.sw_if_index) self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0) self.vapi.nat64_add_del_static_bib(server.ip6n, server_nat_ip_n, server_tcp_in_port, server_tcp_out_port, IP_PROTOS.tcp) self.vapi.nat64_add_del_static_bib(server.ip6n, server_nat_ip_n, 0, 0, IP_PROTOS.gre) self.vapi.nat64_add_del_static_bib(client.ip6n, client_nat_ip_n, client_tcp_in_port, client_tcp_out_port, IP_PROTOS.tcp) # client to server p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / IPv6(src=client.ip6, dst=server_nat_ip6) / TCP(sport=client_tcp_in_port, dport=server_tcp_out_port)) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() p = self.pg0.get_capture(1) p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / IPv6(src=client.ip6, dst=server_nat_ip6, nh=IP_PROTOS.gre) / GRE() / IP(src=self.pg2.local_ip4, dst=self.pg2.remote_ip4) / TCP(sport=1234, dport=1234)) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() p = self.pg0.get_capture(1) packet = p[0] try: self.assertEqual(packet[IPv6].src, client_nat_ip6) self.assertEqual(packet[IPv6].dst, server.ip6) self.assertEqual(packet[IPv6].nh, IP_PROTOS.gre) except: self.logger.error(ppp("Unexpected or invalid packet:", packet)) raise # server to client p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / IPv6(src=server.ip6, dst=client_nat_ip6, nh=IP_PROTOS.gre) / GRE() / IP(src=self.pg2.remote_ip4, dst=self.pg2.local_ip4) / TCP(sport=1234, dport=1234)) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() p = self.pg0.get_capture(1) packet = p[0] try: self.assertEqual(packet[IPv6].src, server_nat_ip6) self.assertEqual(packet[IPv6].dst, client.ip6) self.assertEqual(packet[IPv6].nh, IP_PROTOS.gre) except: self.logger.error(ppp("Unexpected or invalid packet:", packet)) raise def nat64_get_ses_num(self): """ Return number of active NAT64 sessions. """ st = self.vapi.nat64_st_dump() return len(st) def clear_nat64(self): """ Clear NAT64 configuration. """ self.vapi.nat64_set_timeouts() interfaces = self.vapi.nat64_interface_dump() for intf in interfaces: self.vapi.nat64_add_del_interface(intf.sw_if_index, intf.is_inside, is_add=0) bib = self.vapi.nat64_bib_dump(IP_PROTOS.tcp) for bibe in bib: if bibe.is_static: self.vapi.nat64_add_del_static_bib(bibe.i_addr, bibe.o_addr, bibe.i_port, bibe.o_port, bibe.proto, bibe.vrf_id, is_add=0) bib = self.vapi.nat64_bib_dump(IP_PROTOS.udp) for bibe in bib: if bibe.is_static: self.vapi.nat64_add_del_static_bib(bibe.i_addr, bibe.o_addr, bibe.i_port, bibe.o_port, bibe.proto, bibe.vrf_id, is_add=0) bib = self.vapi.nat64_bib_dump(IP_PROTOS.icmp) for bibe in bib: if bibe.is_static: self.vapi.nat64_add_del_static_bib(bibe.i_addr, bibe.o_addr, bibe.i_port, bibe.o_port, bibe.proto, bibe.vrf_id, is_add=0) adresses = self.vapi.nat64_pool_addr_dump() for addr in adresses: self.vapi.nat64_add_del_pool_addr_range(addr.address, addr.address, vrf_id=addr.vrf_id, is_add=0) prefixes = self.vapi.nat64_prefix_dump() for prefix in prefixes: self.vapi.nat64_add_del_prefix(prefix.prefix, prefix.prefix_len, vrf_id=prefix.vrf_id, is_add=0) def tearDown(self): super(TestNAT64, self).tearDown() if not self.vpp_dead: self.logger.info(self.vapi.cli("show nat64 pool")) self.logger.info(self.vapi.cli("show nat64 interfaces")) self.logger.info(self.vapi.cli("show nat64 prefix")) self.logger.info(self.vapi.cli("show nat64 bib all")) self.logger.info(self.vapi.cli("show nat64 session table all")) self.clear_nat64() if __name__ == '__main__': unittest.main(testRunner=VppTestRunner)