#!/usr/bin/env python3 import ipaddress import random import socket import struct import unittest from io import BytesIO import scapy.compat from config import config from framework import tag_fixme_vpp_workers, tag_fixme_ubuntu2204, is_distro_ubuntu2204 from framework import VppTestCase, VppTestRunner from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder from scapy.data import IP_PROTOS from scapy.layers.inet import IP, TCP, UDP, ICMP from scapy.layers.inet import IPerror, TCPerror, UDPerror, ICMPerror from scapy.layers.inet6 import ICMPv6DestUnreach, IPerror6, IPv6ExtHdrFragment from scapy.layers.inet6 import ( IPv6, ICMPv6EchoRequest, ICMPv6EchoReply, ICMPv6ND_NS, ICMPv6ND_NA, ICMPv6NDOptDstLLAddr, fragment6, ) from scapy.layers.l2 import Ether, GRE from scapy.packet import Raw from syslog_rfc5424_parser import SyslogMessage, ParseError from syslog_rfc5424_parser.constants import SyslogSeverity from util import ppc, ppp from vpp_papi import VppEnum @tag_fixme_vpp_workers @tag_fixme_ubuntu2204 class TestNAT64(VppTestCase): """NAT64 Test Cases""" @property def SYSLOG_SEVERITY(self): return VppEnum.vl_api_syslog_severity_t @property def config_flags(self): return VppEnum.vl_api_nat_config_flags_t @classmethod def setUpClass(cls): super(TestNAT64, cls).setUpClass() if is_distro_ubuntu2204 == True and not hasattr(cls, "vpp"): return 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.tcp_external_port = 80 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.ipfix_src_port = 4739 cls.ipfix_domain_id = 1 cls.create_pg_interfaces(range(6)) 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.vapi.ip_table_add_del( is_add=1, table={"table_id": cls.vrf1_id, "is_ip6": 1} ) 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() cls.pg3.admin_up() cls.pg3.config_ip4() cls.pg3.resolve_arp() cls.pg3.config_ip6() cls.pg3.configure_ipv6_neighbors() cls.pg5.admin_up() cls.pg5.config_ip6() @classmethod def tearDownClass(cls): super(TestNAT64, cls).tearDownClass() def setUp(self): super(TestNAT64, self).setUp() self.vapi.nat64_plugin_enable_disable(enable=1, bib_buckets=128, st_buckets=256) def tearDown(self): super(TestNAT64, self).tearDown() if not self.vpp_dead: self.vapi.nat64_plugin_enable_disable(enable=0) def show_commands_at_teardown(self): 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")) 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, use_inside_ports=False): """ 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 :param use_inside_ports: Use inside NAT ports as destination ports instead of outside ports """ if dst_ip is None: dst_ip = self.nat_addr if not use_inside_ports: tcp_port = self.tcp_port_out udp_port = self.udp_port_out icmp_id = self.icmp_id_out else: tcp_port = self.tcp_port_in udp_port = self.udp_port_in icmp_id = self.icmp_id_in 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=tcp_port, sport=20) ) pkts.extend([p, 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=udp_port, 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=icmp_id, type="echo-reply") ) pkts.append(p) return pkts def verify_capture_out( self, capture, nat_ip=None, same_port=False, dst_ip=None, is_ip6=False, ignore_port=False, ): """ Verify captured packets on outside network :param capture: Captured packets :param nat_ip: Translated IP address (Default use global NAT address) :param same_port: Source port number is not translated (Default False) :param dst_ip: Destination IP address (Default do not verify) :param is_ip6: If L3 protocol is IPv6 (Default False) """ if is_ip6: IP46 = IPv6 ICMP46 = ICMPv6EchoRequest else: IP46 = IP ICMP46 = ICMP if nat_ip is None: nat_ip = self.na
"""
  Neighbour Entries

  object abstractions for ARP and ND
"""

from ipaddress import ip_address
from vpp_object import *
from vpp_papi import mac_pton, VppEnum


def find_nbr(test, sw_if_index, nbr_addr, is_static=0, mac=None):
    ip_addr = ip_address(unicode(nbr_addr))
    e = VppEnum.vl_api_ip_neighbor_flags_t
    nbrs = test.vapi.ip_neighbor_dump(sw_if_index,
                                      is_ipv6=(6 is ip_addr.version))

    for n in nbrs:
        if ip_addr == n.neighbor.ip_address and \
           is_static == (n.neighbor.flags & e.IP_API_NEIGHBOR_FLAG_STATIC):
            if mac:
                if mac == str(n.neighbor.mac_address):
                    return True
            else:
                return True
    return False


class VppNeighbor(VppObject):
    """
    ARP Entry
    """

    def __init__(self, test, sw_if_index, mac_addr, nbr_addr,
                 is_static=False, is_no_fib_entry=False):
        self._test = test
        self.sw_if_index = sw_if_index
        self.mac_addr = mac_addr
        self.nbr_addr = nbr_addr

        e = VppEnum.vl_api_ip_neighbor_flags_t
        self.flags = e.IP_API_NEIGHBOR_FLAG_NONE
        if is_static:
            self.flags |= e.IP_API_NEIGHBOR_FLAG_STATIC
        if is_no_fib_entry:
            self.flags |= e.IP_API_NEIGHBOR_FLAG_NO_FIB_ENTRY

    def add_vpp_config(self):
        r = self._test.vapi.ip_neighbor_add_del(
            self.sw_if_index,
            self.mac_addr,
            self.nbr_addr,
            is_add=1,
            flags=self.flags)
        self.stats_index = r.stats_index
        self._test.registry.register(self, self._test.logger)

    def remove_vpp_config(self):
        self._test.vapi.ip_neighbor_add_del(
            self.sw_if_index,
            self.mac_addr,
            self.nbr_addr,
            is_add=0,
            flags=self.flags)

    def is_static(self):
        e = VppEnum.vl_api_ip_neighbor_flags_t
        return (self.flags & e.IP_API_NEIGHBOR_FLAG_STATIC)

    def query_vpp_config(self):
        return find_nbr(self._test,
                        self.sw_if_index,
                        self.nbr_addr,
                        self.is_static())

    def __str__(self):
        return self.object_id()

    def object_id(self):
        return ("%d:%s" % (self.sw_if_index, self.nbr_addr))

    def get_stats(self):
        c = self._test.statistics.get_counter("/net/adjacency")
        return c[0][self.stats_index]
address to NAT64 pool""" nat_addr = "1.2.3.4" self.vapi.nat64_add_del_pool_addr_range( start_addr=nat_addr, end_addr=nat_addr, vrf_id=0xFFFFFFFF, is_add=1 ) addresses = self.vapi.nat64_pool_addr_dump() self.assertEqual(len(addresses), 1) self.assertEqual(str(addresses[0].address), nat_addr) self.vapi.nat64_add_del_pool_addr_range( start_addr=nat_addr, end_addr=nat_addr, vrf_id=0xFFFFFFFF, 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""" flags = self.config_flags.NAT_IS_INSIDE self.vapi.nat64_add_del_interface( is_add=1, flags=flags, sw_if_index=self.pg0.sw_if_index ) self.vapi.nat64_add_del_interface( is_add=1, flags=0, sw_if_index=self.pg1.sw_if_index ) 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.flags, self.config_flags.NAT_IS_INSIDE) pg0_found = True elif intf.sw_if_index == self.pg1.sw_if_index: self.assertEqual(intf.flags, self.config_flags.NAT_IS_OUTSIDE) pg1_found = True self.assertTrue(pg0_found) self.assertTrue(pg1_found) features = self.vapi.cli("show interface features pg0") self.assertIn("nat64-in2out", features) features = self.vapi.cli("show interface features pg1") self.assertIn("nat64-out2in", features) self.vapi.nat64_add_del_interface( is_add=0, flags=flags, sw_if_index=self.pg0.sw_if_index ) self.vapi.nat64_add_del_interface( is_add=0, flags=flags, sw_if_index=self.pg1.sw_if_index ) interfaces = self.vapi.nat64_interface_dump() self.assertEqual(len(interfaces), 0) def test_static_bib(self): """Add/delete static BIB entry""" in_addr = "2001:db8:85a3::8a2e:370:7334" out_addr = "10.1.1.3" in_port = 1234 out_port = 5678 proto = IP_PROTOS.tcp self.vapi.nat64_add_del_static_bib( i_addr=in_addr, o_addr=out_addr, i_port=in_port, o_port=out_port, proto=proto, vrf_id=0, is_add=1, ) bib = self.vapi.nat64_bib_dump(proto=IP_PROTOS.tcp) static_bib_num = 0 for bibe in bib: if bibe.flags & self.config_flags.NAT_IS_STATIC: static_bib_num += 1 self.assertEqual(str(bibe.i_addr), in_addr) self.assertEqual(str(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) bibs = self.statistics.get_counter("/nat64/total-bibs") self.assertEqual(bibs[0][0], 1) self.vapi.nat64_add_del_static_bib( i_addr=in_addr, o_addr=out_addr, i_port=in_port, o_port=out_port, proto=proto, vrf_id=0, is_add=0, ) bib = self.vapi.nat64_bib_dump(proto=IP_PROTOS.tcp) static_bib_num = 0 for bibe in bib: if bibe.flags & self.config_flags.NAT_IS_STATIC: static_bib_num += 1 self.assertEqual(static_bib_num, 0) bibs = self.statistics.get_counter("/nat64/total-bibs") self.assertEqual(bibs[0][0], 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_transitory, 240) self.assertEqual(timeouts.tcp_established, 7440) # set and verify custom values self.vapi.nat64_set_timeouts( udp=200, tcp_established=7450, tcp_transitory=250, icmp=30 ) timeouts = self.vapi.nat64_get_timeouts() self.assertEqual(timeouts.udp, 200) self.assertEqual(timeouts.icmp, 30) self.assertEqual(timeouts.tcp_transitory, 250) self.assertEqual(timeouts.tcp_established, 7450) 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( start_addr=self.nat_addr, end_addr=self.nat_addr, vrf_id=0xFFFFFFFF, is_add=1, ) flags = self.config_flags.NAT_IS_INSIDE self.vapi.nat64_add_del_interface( is_add=1, flags=flags, sw_if_index=self.pg0.sw_if_index ) self.vapi.nat64_add_del_interface( is_add=1, flags=0, sw_if_index=self.pg1.sw_if_index ) # in2out tcpn = self.statistics.get_counter("/nat64/in2out/tcp")[0] udpn = self.statistics.get_counter("/nat64/in2out/udp")[0] icmpn = self.statistics.get_counter("/nat64/in2out/icmp")[0] drops = self.statistics.get_counter("/nat64/in2out/drops")[0] 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 ) if_idx = self.pg0.sw_if_index cnt = self.statistics.get_counter("/nat64/in2out/tcp")[0] self.assertEqual(cnt[if_idx] - tcpn[if_idx], 1) cnt = self.statistics.get_counter("/nat64/in2out/udp")[0] self.assertEqual(cnt[if_idx] - udpn[if_idx], 1) cnt = self.statistics.get_counter("/nat64/in2out/icmp")[0] self.assertEqual(cnt[if_idx] - icmpn[if_idx], 1) cnt = self.statistics.get_counter("/nat64/in2out/drops")[0] self.assertEqual(cnt[if_idx] - drops[if_idx], 0) # out2in tcpn = self.statistics.get_counter("/nat64/out2in/tcp")[0] udpn = self.statistics.get_counter("/nat64/out2in/udp")[0] icmpn = self.statistics.get_counter("/nat64/out2in/icmp")[0] drops = self.statistics.get_counter("/nat64/out2in/drops")[0] 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) if_idx = self.pg1.sw_if_index cnt = self.statistics.get_counter("/nat64/out2in/tcp")[0] self.assertEqual(cnt[if_idx] - tcpn[if_idx], 2) cnt = self.statistics.get_counter("/nat64/out2in/udp")[0] self.assertEqual(cnt[if_idx] - udpn[if_idx], 1) cnt = self.statistics.get_counter("/nat64/out2in/icmp")[0] self.assertEqual(cnt[if_idx] - icmpn[if_idx], 1) cnt = self.statistics.get_counter("/nat64/out2in/drops")[0] self.assertEqual(cnt[if_idx] - drops[if_idx], 0) bibs = self.statistics.get_counter("/nat64/total-bibs") self.assertEqual(bibs[0][0], 3) sessions = self.statistics.get_counter("/nat64/total-sessions") self.assertEqual(sessions[0][0], 3) # 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( start_addr=self.vrf1_nat_addr, end_addr=self.vrf1_nat_addr, vrf_id=self.vrf1_id, is_add=1, ) flags = self.config_flags.NAT_IS_INSIDE self.vapi.nat64_add_del_interface( is_add=1, flags=flags, sw_if_index=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( start_addr=self.nat_addr, end_addr=self.nat_addr, vrf_id=0xFFFFFFFF, is_add=1, ) flags = self.config_flags.NAT_IS_INSIDE self.vapi.nat64_add_del_interface( is_add=1, flags=flags, sw_if_index=self.pg0.sw_if_index ) self.vapi.nat64_add_del_interface( is_add=1, flags=0, sw_if_index=self.pg1.sw_if_index ) self.vapi.nat64_add_del_static_bib( i_addr=self.pg0.remote_ip6, o_addr=self.nat_addr, i_port=self.tcp_port_in, o_port=self.tcp_port_out, proto=IP_PROTOS.tcp, vrf_id=0, is_add=1, ) self.vapi.nat64_add_del_static_bib( i_addr=self.pg0.remote_ip6, o_addr=self.nat_addr, i_port=self.udp_port_in, o_port=self.udp_port_out, proto=IP_PROTOS.udp, vrf_id=0, is_add=1, ) self.vapi.nat64_add_del_static_bib( i_addr=self.pg0.remote_ip6, o_addr=self.nat_addr, i_port=self.icmp_id_in, o_port=self.icmp_id_out, proto=IP_PROTOS.icmp, vrf_id=0, is_add=1, ) # 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) def test_session_timeout(self): """NAT64 session timeout""" self.icmp_id_in = 1234 self.vapi.nat64_add_del_pool_addr_range( start_addr=self.nat_addr, end_addr=self.nat_addr, vrf_id=0xFFFFFFFF, is_add=1, ) flags = self.config_flags.NAT_IS_INSIDE self.vapi.nat64_add_del_interface( is_add=1, flags=flags, sw_if_index=self.pg0.sw_if_index ) self.vapi.nat64_add_del_interface( is_add=1, flags=0, sw_if_index=self.pg1.sw_if_index ) self.vapi.nat64_set_timeouts( udp=300, tcp_established=5, tcp_transitory=5, 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() self.virtual_sleep(15) # ICMP and TCP session after timeout ses_num_after_timeout = self.nat64_get_ses_num() self.assertEqual(ses_num_before_timeout - ses_num_after_timeout, 2) 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 self.vapi.nat64_add_del_pool_addr_range( start_addr=self.nat_addr, end_addr=self.nat_addr, vrf_id=0xFFFFFFFF, is_add=1, ) flags = self.config_flags.NAT_IS_INSIDE self.vapi.nat64_add_del_interface( is_add=1, flags=flags, sw_if_index=self.pg0.sw_if_index ) self.vapi.nat64_add_del_interface( is_add=1, flags=0, sw_if_index=self.pg1.sw_if_index ) # 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.assert_packet_checksums_valid(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.assert_icmpv6_checksum_valid(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( start_addr=self.nat_addr, end_addr=self.nat_addr, vrf_id=0xFFFFFFFF, is_add=1, ) flags = self.config_flags.NAT_IS_INSIDE self.vapi.nat64_add_del_interface( is_add=1, flags=flags, sw_if_index=self.pg0.sw_if_index ) self.vapi.nat64_add_del_interface( is_add=1, flags=0, sw_if_index=self.pg1.sw_if_index ) self.vapi.nat64_add_del_static_bib( i_addr=server.ip6n, o_addr=self.nat_addr, i_port=server_tcp_in_port, o_port=server_tcp_out_port, proto=IP_PROTOS.tcp, vrf_id=0, is_add=1, ) self.vapi.nat64_add_del_static_bib( i_addr=server.ip6n, o_addr=self.nat_addr, i_port=server_udp_in_port, o_port=server_udp_out_port, proto=IP_PROTOS.udp, vrf_id=0, is_add=1, ) # 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) self.assert_packet_checksums_valid(packet) if packet.haslayer(TCP): self.assertNotEqual(packet[TCP].sport, client_tcp_in_port) self.assertEqual(packet[TCP].dport, server_tcp_in_port) 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) 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) self.assert_packet_checksums_valid(packet) if packet.haslayer(TCP): self.assertEqual(packet[TCP].sport, server_tcp_out_port) self.assertEqual(packet[TCP].dport, client_tcp_in_port) else: self.assertEqual(packet[UDP].sport, server_udp_out_port) self.assertEqual(packet[UDP].dport, client_udp_in_port) 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.assert_packet_checksums_valid(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( start_addr=self.nat_addr, end_addr=self.nat_addr, vrf_id=0xFFFFFFFF, is_add=1, ) flags = self.config_flags.NAT_IS_INSIDE self.vapi.nat64_add_del_interface( is_add=1, flags=flags, sw_if_index=self.pg0.sw_if_index ) self.vapi.nat64_add_del_interface( is_add=1, flags=0, sw_if_index=self.pg1.sw_if_index ) self.vapi.nat64_add_del_pool_addr_range( start_addr=self.vrf1_nat_addr, end_addr=self.vrf1_nat_addr, vrf_id=self.vrf1_id, is_add=1, ) self.vapi.nat64_add_del_interface( is_add=1, flags=flags, sw_if_index=self.pg2.sw_if_index ) # Add global prefix global_pref64 = "2001:db8::" global_pref64_len = 32 global_pref64_str = "{}/{}".format(global_pref64, global_pref64_len) self.vapi.nat64_add_del_prefix(prefix=global_pref64_str, vrf_id=0, is_add=1) prefix = self.vapi.nat64_prefix_dump() self.assertEqual(len(prefix), 1) self.assertEqual(str(prefix[0].prefix), global_pref64_str) self.assertEqual(prefix[0].vrf_id, 0) # Add tenant specific prefix vrf1_pref64 = "2001:db8:122:300::" vrf1_pref64_len = 56 vrf1_pref64_str = "{}/{}".format(vrf1_pref64, vrf1_pref64_len) self.vapi.nat64_add_del_prefix( prefix=vrf1_pref64_str, vrf_id=self.vrf1_id, is_add=1 ) 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( start_addr=self.nat_addr, end_addr=self.nat_addr, vrf_id=0xFFFFFFFF, is_add=1, ) flags = self.config_flags.NAT_IS_INSIDE self.vapi.nat64_add_del_interface( is_add=1, flags=flags, sw_if_index=self.pg0.sw_if_index ) self.vapi.nat64_add_del_interface( is_add=1, flags=0, sw_if_index=self.pg1.sw_if_index ) 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.assertEqual(packet.haslayer(GRE), 1) self.assert_packet_checksums_valid(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_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( start_addr=server_nat_ip, end_addr=client_nat_ip, vrf_id=0xFFFFFFFF, is_add=1, ) flags = self.config_flags.NAT_IS_INSIDE self.vapi.nat64_add_del_interface( is_add=1, flags=flags, sw_if_index=self.pg0.sw_if_index ) self.vapi.nat64_add_del_interface( is_add=1, flags=0, sw_if_index=self.pg1.sw_if_index ) self.vapi.nat64_add_del_static_bib( i_addr=server.ip6n, o_addr=server_nat_ip, i_port=server_tcp_in_port, o_port=server_tcp_out_port, proto=IP_PROTOS.tcp, vrf_id=0, is_add=1, ) self.vapi.nat64_add_del_static_bib( i_addr=server.ip6n, o_addr=server_nat_ip, i_port=0, o_port=0, proto=IP_PROTOS.gre, vrf_id=0, is_add=1, ) self.vapi.nat64_add_del_static_bib( i_addr=client.ip6n, o_addr=client_nat_ip, i_port=client_tcp_in_port, o_port=client_tcp_out_port, proto=IP_PROTOS.tcp, vrf_id=0, is_add=1, ) # 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 test_one_armed_nat64(self): """One armed NAT64""" external_port = 0 remote_host_ip6 = self.compose_ip6(self.pg3.remote_ip4, "64:ff9b::", 96) self.vapi.nat64_add_del_pool_addr_range( start_addr=self.nat_addr, end_addr=self.nat_addr, vrf_id=0xFFFFFFFF, is_add=1, ) flags = self.config_flags.NAT_IS_INSIDE self.vapi.nat64_add_del_interface( is_add=1, flags=flags, sw_if_index=self.pg3.sw_if_index ) self.vapi.nat64_add_del_interface( is_add=1, flags=0, sw_if_index=self.pg3.sw_if_index ) # in2out p = ( Ether(src=self.pg3.remote_mac, dst=self.pg3.local_mac) / IPv6(src=self.pg3.remote_ip6, dst=remote_host_ip6) / TCP(sport=12345, dport=80) ) self.pg3.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg3.get_capture(1) p = capture[0] try: ip = p[IP] tcp = p[TCP] self.assertEqual(ip.src, self.nat_addr) self.assertEqual(ip.dst, self.pg3.remote_ip4) self.assertNotEqual(tcp.sport, 12345) external_port = tcp.sport self.assertEqual(tcp.dport, 80) self.assert_packet_checksums_valid(p) except: self.logger.error(ppp("Unexpected or invalid packet:", p)) raise # out2in p = ( Ether(src=self.pg3.remote_mac, dst=self.pg3.local_mac) / IP(src=self.pg3.remote_ip4, dst=self.nat_addr) / TCP(sport=80, dport=external_port) ) self.pg3.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg3.get_capture(1) p = capture[0] try: ip = p[IPv6] tcp = p[TCP] self.assertEqual(ip.src, remote_host_ip6) self.assertEqual(ip.dst, self.pg3.remote_ip6) self.assertEqual(tcp.sport, 80) self.assertEqual(tcp.dport, 12345) self.assert_packet_checksums_valid(p) except: self.logger.error(ppp("Unexpected or invalid packet:", p)) raise def test_frag_in_order(self): """NAT64 translate fragments arriving in order""" self.tcp_port_in = random.randint(1025, 65535) self.vapi.nat64_add_del_pool_addr_range( start_addr=self.nat_addr, end_addr=self.nat_addr, vrf_id=0xFFFFFFFF, is_add=1, ) flags = self.config_flags.NAT_IS_INSIDE self.vapi.nat64_add_del_interface( is_add=1, flags=flags, sw_if_index=self.pg0.sw_if_index ) self.vapi.nat64_add_del_interface( is_add=1, flags=0, sw_if_index=self.pg1.sw_if_index ) # in2out data = b"a" * 200 pkts = self.create_stream_frag_ip6( self.pg0, self.pg1.remote_ip4, self.tcp_port_in, 20, data ) self.pg0.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() frags = self.pg1.get_capture(len(pkts)) p = self.reass_frags_and_verify(frags, self.nat_addr, self.pg1.remote_ip4) self.assertEqual(p[TCP].dport, 20) self.assertNotEqual(p[TCP].sport, self.tcp_port_in) self.tcp_port_out = p[TCP].sport self.assertEqual(data, p[Raw].load) # out2in data = b"A" * 4 + b"b" * 16 + b"C" * 3 pkts = self.create_stream_frag( self.pg1, self.nat_addr, 20, self.tcp_port_out, data ) self.pg1.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() frags = self.pg0.get_capture(len(pkts)) self.logger.debug(ppc("Captured:", frags)) src = self.compose_ip6(self.pg1.remote_ip4, "64:ff9b::", 96) p = self.reass_frags_and_verify_ip6(frags, src, self.pg0.remote_ip6) self.assertEqual(p[TCP].sport, 20) self.assertEqual(p[TCP].dport, self.tcp_port_in) self.assertEqual(data, p[Raw].load) def test_reass_hairpinning(self): """NAT64 fragments hairpinning""" data = b"a" * 200 server = self.pg0.remote_hosts[1] server_in_port = random.randint(1025, 65535) server_out_port = random.randint(1025, 65535) client_in_port = random.randint(1025, 65535) ip = IPv6(src="".join(["64:ff9b::", self.nat_addr])) nat_addr_ip6 = ip.src self.vapi.nat64_add_del_pool_addr_range( start_addr=self.nat_addr, end_addr=self.nat_addr, vrf_id=0xFFFFFFFF, is_add=1, ) flags = self.config_flags.NAT_IS_INSIDE self.vapi.nat64_add_del_interface( is_add=1, flags=flags, sw_if_index=self.pg0.sw_if_index ) self.vapi.nat64_add_del_interface( is_add=1, flags=0, sw_if_index=self.pg1.sw_if_index ) # add static BIB entry for server self.vapi.nat64_add_del_static_bib( i_addr=server.ip6n, o_addr=self.nat_addr, i_port=server_in_port, o_port=server_out_port, proto=IP_PROTOS.tcp, vrf_id=0, is_add=1, ) # send packet from host to server pkts = self.create_stream_frag_ip6( self.pg0, self.nat_addr, client_in_port, server_out_port, data ) self.pg0.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() frags = self.pg0.get_capture(len(pkts)) self.logger.debug(ppc("Captured:", frags)) p = self.reass_frags_and_verify_ip6(frags, nat_addr_ip6, server.ip6) self.assertNotEqual(p[TCP].sport, client_in_port) self.assertEqual(p[TCP].dport, server_in_port) self.assertEqual(data, p[Raw].load) def test_frag_out_of_order(self): """NAT64 translate fragments arriving out of order""" self.tcp_port_in = random.randint(1025, 65535) self.vapi.nat64_add_del_pool_addr_range( start_addr=self.nat_addr, end_addr=self.nat_addr, vrf_id=0xFFFFFFFF, is_add=1, ) flags = self.config_flags.NAT_IS_INSIDE self.vapi.nat64_add_del_interface( is_add=1, flags=flags, sw_if_index=self.pg0.sw_if_index ) self.vapi.nat64_add_del_interface( is_add=1, flags=0, sw_if_index=self.pg1.sw_if_index ) # in2out data = b"a" * 200 pkts = self.create_stream_frag_ip6( self.pg0, self.pg1.remote_ip4, self.tcp_port_in, 20, data ) pkts.reverse() self.pg0.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() frags = self.pg1.get_capture(len(pkts)) p = self.reass_frags_and_verify(frags, self.nat_addr, self.pg1.remote_ip4) self.assertEqual(p[TCP].dport, 20) self.assertNotEqual(p[TCP].sport, self.tcp_port_in) self.tcp_port_out = p[TCP].sport self.assertEqual(data, p[Raw].load) # out2in data = b"A" * 4 + b"B" * 16 + b"C" * 3 pkts = self.create_stream_frag( self.pg1, self.nat_addr, 20, self.tcp_port_out, data ) pkts.reverse() self.pg1.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() frags = self.pg0.get_capture(len(pkts)) src = self.compose_ip6(self.pg1.remote_ip4, "64:ff9b::", 96) p = self.reass_frags_and_verify_ip6(frags, src, self.pg0.remote_ip6) self.assertEqual(p[TCP].sport, 20) self.assertEqual(p[TCP].dport, self.tcp_port_in) self.assertEqual(data, p[Raw].load) def test_interface_addr(self): """Acquire NAT64 pool addresses from interface""" self.vapi.nat64_add_del_interface_addr( is_add=1, sw_if_index=self.pg4.sw_if_index ) # no address in NAT64 pool addresses = self.vapi.nat44_address_dump() self.assertEqual(0, len(addresses)) # configure interface address and check NAT64 address pool self.pg4.config_ip4() addresses = self.vapi.nat64_pool_addr_dump() self.assertEqual(len(addresses), 1) self.assertEqual(str(addresses[0].address), self.pg4.local_ip4) # remove interface address and check NAT64 address pool self.pg4.unconfig_ip4() addresses = self.vapi.nat64_pool_addr_dump() self.assertEqual(0, len(addresses)) @unittest.skipUnless(config.extended, "part of extended tests") def test_ipfix_max_bibs_sessions(self): """IPFIX logging maximum session and BIB entries exceeded""" max_bibs = 1280 max_sessions = 2560 remote_host_ip6 = self.compose_ip6(self.pg1.remote_ip4, "64:ff9b::", 96) self.vapi.nat64_add_del_pool_addr_range( start_addr=self.nat_addr, end_addr=self.nat_addr, vrf_id=0xFFFFFFFF, is_add=1, ) flags = self.config_flags.NAT_IS_INSIDE self.vapi.nat64_add_del_interface( is_add=1, flags=flags, sw_if_index=self.pg0.sw_if_index ) self.vapi.nat64_add_del_interface( is_add=1, flags=0, sw_if_index=self.pg1.sw_if_index ) pkts = [] src = "" for i in range(0, max_bibs): src = "fd01:aa::%x" % (i) p = ( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IPv6(src=src, dst=remote_host_ip6) / TCP(sport=12345, dport=80) ) pkts.append(p) p = ( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IPv6(src=src, dst=remote_host_ip6) / TCP(sport=12345, dport=22) ) pkts.append(p) self.pg0.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() self.pg1.get_capture(max_sessions) self.vapi.set_ipfix_exporter( collector_address=self.pg3.remote_ip4, src_address=self.pg3.local_ip4, path_mtu=512, template_interval=10, ) self.vapi.nat_ipfix_enable_disable( domain_id=self.ipfix_domain_id, src_port=self.ipfix_src_port, enable=1 ) p = ( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IPv6(src=src, dst=remote_host_ip6) / TCP(sport=12345, dport=25) ) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() self.pg1.assert_nothing_captured() self.vapi.ipfix_flush() capture = self.pg3.get_capture(7) 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_max_sessions(data, max_sessions) p = ( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IPv6(src=self.pg0.remote_ip6, dst=remote_host_ip6) / TCP(sport=12345, dport=80) ) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() self.pg1.assert_nothing_captured() self.vapi.ipfix_flush() capture = self.pg3.get_capture(1) # verify events in data set 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(Data): data = ipfix.decode_data_set(p.getlayer(Set)) self.verify_ipfix_max_bibs(data, max_bibs) def test_ipfix_bib_ses(self): """IPFIX logging NAT64 BIB/session create and delete events""" self.tcp_port_in = random.randint(1025, 65535) remote_host_ip6 = self.compose_ip6(self.pg1.remote_ip4, "64:ff9b::", 96) self.vapi.nat64_add_del_pool_addr_range( start_addr=self.nat_addr, end_addr=self.nat_addr, vrf_id=0xFFFFFFFF, is_add=1, ) flags = self.config_flags.NAT_IS_INSIDE self.vapi.nat64_add_del_interface( is_add=1, flags=flags, sw_if_index=self.pg0.sw_if_index ) self.vapi.nat64_add_del_interface( is_add=1, flags=0, sw_if_index=self.pg1.sw_if_index ) self.vapi.set_ipfix_exporter( collector_address=self.pg3.remote_ip4, src_address=self.pg3.local_ip4, path_mtu=512, template_interval=10, ) self.vapi.nat_ipfix_enable_disable( domain_id=self.ipfix_domain_id, src_port=self.ipfix_src_port, enable=1 ) # Create p = ( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IPv6(src=self.pg0.remote_ip6, dst=remote_host_ip6) / TCP(sport=self.tcp_port_in, dport=25) ) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() p = self.pg1.get_capture(1) self.tcp_port_out = p[0][TCP].sport self.vapi.ipfix_flush() capture = self.pg3.get_capture(8) 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)) if scapy.compat.orb(data[0][230]) == 10: self.verify_ipfix_bib(data, 1, self.pg0.remote_ip6) elif scapy.compat.orb(data[0][230]) == 6: self.verify_ipfix_nat64_ses( data, 1, self.pg0.remote_ip6, self.pg1.remote_ip4, 25 ) else: self.logger.error(ppp("Unexpected or invalid packet: ", p)) # Delete self.pg_enable_capture(self.pg_interfaces) self.vapi.nat64_add_del_pool_addr_range( start_addr=self.nat_addr, end_addr=self.nat_addr, vrf_id=0xFFFFFFFF, is_add=0, ) self.vapi.ipfix_flush() capture = self.pg3.get_capture(2) # verify events in data set 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(Data): data = ipfix.decode_data_set(p.getlayer(Set)) if scapy.compat.orb(data[0][230]) == 11: self.verify_ipfix_bib(data, 0, self.pg0.remote_ip6) elif scapy.compat.orb(data[0][230]) == 7: self.verify_ipfix_nat64_ses( data, 0, self.pg0.remote_ip6, self.pg1.remote_ip4, 25 ) else: self.logger.error(ppp("Unexpected or invalid packet: ", p)) def test_syslog_sess(self): """Test syslog session creation and deletion""" self.tcp_port_in = random.randint(1025, 65535) remote_host_ip6 = self.compose_ip6(self.pg1.remote_ip4, "64:ff9b::", 96) self.vapi.nat64_add_del_pool_addr_range( start_addr=self.nat_addr, end_addr=self.nat_addr, vrf_id=0xFFFFFFFF, is_add=1, ) flags = self.config_flags.NAT_IS_INSIDE self.vapi.nat64_add_del_interface( is_add=1, flags=flags, sw_if_index=self.pg0.sw_if_index ) self.vapi.nat64_add_del_interface( is_add=1, flags=0, sw_if_index=self.pg1.sw_if_index ) self.vapi.syslog_set_filter(self.SYSLOG_SEVERITY.SYSLOG_API_SEVERITY_INFO) self.vapi.syslog_set_sender(self.pg3.local_ip4, self.pg3.remote_ip4) p = ( Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IPv6(src=self.pg0.remote_ip6, dst=remote_host_ip6) / TCP(sport=self.tcp_port_in, dport=self.tcp_external_port) ) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() p = self.pg1.get_capture(1) self.tcp_port_out = p[0][TCP].sport capture = self.pg3.get_capture(1) self.verify_syslog_sess(capture[0][Raw].load, is_ip6=True) self.pg_enable_capture(self.pg_interfaces) self.pg_start() self.vapi.nat64_add_del_pool_addr_range( start_addr=self.nat_addr, end_addr=self.nat_addr, vrf_id=0xFFFFFFFF, is_add=0, ) capture = self.pg3.get_capture(1) self.verify_syslog_sess(capture[0][Raw].load, False, True) def nat64_get_ses_num(self): """ Return number of active NAT64 sessions. """ st = self.vapi.nat64_st_dump(proto=255) return len(st) def clear_nat64(self): """ Clear NAT64 configuration. """ self.vapi.nat_ipfix_enable_disable( domain_id=self.ipfix_domain_id, src_port=self.ipfix_src_port, enable=0 ) self.ipfix_src_port = 4739 self.ipfix_domain_id = 1 self.vapi.syslog_set_filter(self.SYSLOG_SEVERITY.SYSLOG_API_SEVERITY_EMERG) self.vapi.nat64_set_timeouts( udp=300, tcp_established=7440, tcp_transitory=240, icmp=60 ) interfaces = self.vapi.nat64_interface_dump() for intf in interfaces: self.vapi.nat64_add_del_interface( is_add=0, flags=intf.flags, sw_if_index=intf.sw_if_index ) bib = self.vapi.nat64_bib_dump(proto=255) for bibe in bib: if bibe.flags & self.config_flags.NAT_IS_STATIC: self.vapi.nat64_add_del_static_bib( i_addr=bibe.i_addr, o_addr=bibe.o_addr, i_port=bibe.i_port, o_port=bibe.o_port, proto=bibe.proto, vrf_id=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( start_addr=addr.address, end_addr=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=str(prefix.prefix), vrf_id=prefix.vrf_id, is_add=0 ) bibs = self.statistics.get_counter("/nat64/total-bibs") self.assertEqual(bibs[0][0], 0) sessions = self.statistics.get_counter("/nat64/total-sessions") self.assertEqual(sessions[0][0], 0) if __name__ == "__main__": unittest.main(testRunner=VppTestRunner)