diff options
Diffstat (limited to 'src/plugins/nat/test/test_nat44_ed.py')
| -rw-r--r-- | src/plugins/nat/test/test_nat44_ed.py | 3662 |
1 files changed, 0 insertions, 3662 deletions
diff --git a/src/plugins/nat/test/test_nat44_ed.py b/src/plugins/nat/test/test_nat44_ed.py deleted file mode 100644 index 2ce7f23dac9..00000000000 --- a/src/plugins/nat/test/test_nat44_ed.py +++ /dev/null @@ -1,3662 +0,0 @@ -#!/usr/bin/env python3 - -import unittest -from io import BytesIO -from random import randint, shuffle, choice - -import scapy.compat -from framework import VppTestCase, VppTestRunner -from scapy.data import IP_PROTOS -from scapy.layers.inet import IP, TCP, UDP, ICMP, GRE -from scapy.layers.inet import IPerror, TCPerror -from scapy.layers.l2 import Ether -from scapy.packet import Raw -from syslog_rfc5424_parser import SyslogMessage, ParseError -from syslog_rfc5424_parser.constants import SyslogSeverity -from util import ppp, ip4_range -from vpp_acl import AclRule, VppAcl, VppAclInterface -from vpp_ip_route import VppIpRoute, VppRoutePath -from vpp_papi import VppEnum - - -class NAT44EDTestCase(VppTestCase): - - nat_addr = '10.0.0.3' - - tcp_port_in = 6303 - tcp_port_out = 6303 - - udp_port_in = 6304 - udp_port_out = 6304 - - icmp_id_in = 6305 - icmp_id_out = 6305 - - tcp_external_port = 80 - - max_sessions = 100 - - def setUp(self): - super(NAT44EDTestCase, self).setUp() - self.plugin_enable() - - def tearDown(self): - super(NAT44EDTestCase, self).tearDown() - if not self.vpp_dead: - self.plugin_disable() - - def plugin_enable(self): - self.vapi.nat44_ed_plugin_enable_disable( - sessions=self.max_sessions, enable=1) - - def plugin_disable(self): - self.vapi.nat44_ed_plugin_enable_disable(enable=0) - - @property - def config_flags(self): - return VppEnum.vl_api_nat_config_flags_t - - @property - def nat44_config_flags(self): - return VppEnum.vl_api_nat44_config_flags_t - - @property - def syslog_severity(self): - return VppEnum.vl_api_syslog_severity_t - - @property - def server_addr(self): - return self.pg1.remote_hosts[0].ip4 - - @staticmethod - def random_port(): - return randint(1025, 65535) - - @staticmethod - def proto2layer(proto): - if proto == IP_PROTOS.tcp: - return TCP - elif proto == IP_PROTOS.udp: - return UDP - elif proto == IP_PROTOS.icmp: - return ICMP - else: - raise Exception("Unsupported protocol") - - @classmethod - def create_and_add_ip4_table(cls, i, table_id=0): - cls.vapi.ip_table_add_del(is_add=1, table={'table_id': table_id}) - i.set_table_ip4(table_id) - - @classmethod - def configure_ip4_interface(cls, i, hosts=0, table_id=None): - if table_id: - cls.create_and_add_ip4_table(i, table_id) - - i.admin_up() - i.config_ip4() - i.resolve_arp() - - if hosts: - i.generate_remote_hosts(hosts) - i.configure_ipv4_neighbors() - - @classmethod - def nat_add_interface_address(cls, i): - cls.vapi.nat44_add_del_interface_addr( - sw_if_index=i.sw_if_index, is_add=1) - - def nat_add_inside_interface(self, i): - self.vapi.nat44_interface_add_del_feature( - flags=self.config_flags.NAT_IS_INSIDE, - sw_if_index=i.sw_if_index, is_add=1) - - def nat_add_outside_interface(self, i): - self.vapi.nat44_interface_add_del_feature( - flags=self.config_flags.NAT_IS_OUTSIDE, - sw_if_index=i.sw_if_index, is_add=1) - - def nat_add_address(self, address, twice_nat=0, - vrf_id=0xFFFFFFFF, is_add=1): - flags = self.config_flags.NAT_IS_TWICE_NAT if twice_nat else 0 - self.vapi.nat44_add_del_address_range(first_ip_address=address, - last_ip_address=address, - vrf_id=vrf_id, - is_add=is_add, - flags=flags) - - def nat_add_static_mapping(self, local_ip, external_ip='0.0.0.0', - local_port=0, external_port=0, vrf_id=0, - is_add=1, external_sw_if_index=0xFFFFFFFF, - proto=0, tag="", flags=0): - - if not (local_port and external_port): - flags |= self.config_flags.NAT_IS_ADDR_ONLY - - self.vapi.nat44_add_del_static_mapping( - is_add=is_add, - local_ip_address=local_ip, - external_ip_address=external_ip, - external_sw_if_index=external_sw_if_index, - local_port=local_port, - external_port=external_port, - vrf_id=vrf_id, protocol=proto, - flags=flags, - tag=tag) - - @classmethod - def setUpClass(cls): - super(NAT44EDTestCase, cls).setUpClass() - - cls.create_pg_interfaces(range(12)) - cls.interfaces = list(cls.pg_interfaces[:4]) - - cls.create_and_add_ip4_table(cls.pg2, 10) - - for i in cls.interfaces: - cls.configure_ip4_interface(i, hosts=3) - - # test specific (test-multiple-vrf) - cls.vapi.ip_table_add_del(is_add=1, table={'table_id': 1}) - - # test specific (test-one-armed-nat44-static) - cls.pg4.generate_remote_hosts(2) - cls.pg4.config_ip4() - cls.vapi.sw_interface_add_del_address( - sw_if_index=cls.pg4.sw_if_index, - prefix="10.0.0.1/24") - cls.pg4.admin_up() - cls.pg4.resolve_arp() - cls.pg4._remote_hosts[1]._ip4 = cls.pg4._remote_hosts[0]._ip4 - cls.pg4.resolve_arp() - - # test specific interface (pg5) - cls.pg5._local_ip4 = "10.1.1.1" - cls.pg5._remote_hosts[0]._ip4 = "10.1.1.2" - cls.pg5.set_table_ip4(1) - cls.pg5.config_ip4() - cls.pg5.admin_up() - cls.pg5.resolve_arp() - - # test specific interface (pg6) - cls.pg6._local_ip4 = "10.1.2.1" - cls.pg6._remote_hosts[0]._ip4 = "10.1.2.2" - cls.pg6.set_table_ip4(1) - cls.pg6.config_ip4() - cls.pg6.admin_up() - cls.pg6.resolve_arp() - - rl = list() - - rl.append(VppIpRoute(cls, "0.0.0.0", 0, - [VppRoutePath("0.0.0.0", 0xffffffff, - nh_table_id=0)], - register=False, table_id=1)) - rl.append(VppIpRoute(cls, "0.0.0.0", 0, - [VppRoutePath(cls.pg1.local_ip4, - cls.pg1.sw_if_index)], - register=False)) - rl.append(VppIpRoute(cls, cls.pg5.remote_ip4, 32, - [VppRoutePath("0.0.0.0", - cls.pg5.sw_if_index)], - register=False, table_id=1)) - rl.append(VppIpRoute(cls, cls.pg6.remote_ip4, 32, - [VppRoutePath("0.0.0.0", - cls.pg6.sw_if_index)], - register=False, table_id=1)) - rl.append(VppIpRoute(cls, cls.pg6.remote_ip4, 16, - [VppRoutePath("0.0.0.0", 0xffffffff, - nh_table_id=1)], - register=False, table_id=0)) - - for r in rl: - r.add_vpp_config() - - def get_err_counter(self, path): - return self.statistics.get_err_counter(path) - - def reass_hairpinning(self, server_addr, server_in_port, server_out_port, - host_in_port, proto=IP_PROTOS.tcp, - ignore_port=False): - layer = self.proto2layer(proto) - - if proto == IP_PROTOS.tcp: - data = b"A" * 4 + b"B" * 16 + b"C" * 3 - else: - data = b"A" * 16 + b"B" * 16 + b"C" * 3 - - # send packet from host to server - pkts = self.create_stream_frag(self.pg0, - self.nat_addr, - host_in_port, - server_out_port, - data, - proto) - self.pg0.add_stream(pkts) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - frags = self.pg0.get_capture(len(pkts)) - p = self.reass_frags_and_verify(frags, - self.nat_addr, - server_addr) - if proto != IP_PROTOS.icmp: - if not ignore_port: - self.assertNotEqual(p[layer].sport, host_in_port) - self.assertEqual(p[layer].dport, server_in_port) - else: - if not ignore_port: - self.assertNotEqual(p[layer].id, host_in_port) - self.assertEqual(data, p[Raw].load) - - def frag_out_of_order(self, proto=IP_PROTOS.tcp, dont_translate=False, - ignore_port=False): - layer = self.proto2layer(proto) - - if proto == IP_PROTOS.tcp: - data = b"A" * 4 + b"B" * 16 + b"C" * 3 - else: - data = b"A" * 16 + b"B" * 16 + b"C" * 3 - self.port_in = self.random_port() - - for i in range(2): - # in2out - pkts = self.create_stream_frag(self.pg0, self.pg1.remote_ip4, - self.port_in, 20, data, proto) - pkts.reverse() - self.pg0.add_stream(pkts) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - frags = self.pg1.get_capture(len(pkts)) - if not dont_translate: - p = self.reass_frags_and_verify(frags, - self.nat_addr, - self.pg1.remote_ip4) - else: - p = self.reass_frags_and_verify(frags, - self.pg0.remote_ip4, - self.pg1.remote_ip4) - if proto != IP_PROTOS.icmp: - if not dont_translate: - self.assertEqual(p[layer].dport, 20) - if not ignore_port: - self.assertNotEqual(p[layer].sport, self.port_in) - else: - self.assertEqual(p[layer].sport, self.port_in) - else: - if not ignore_port: - if not dont_translate: - self.assertNotEqual(p[layer].id, self.port_in) - else: - self.assertEqual(p[layer].id, self.port_in) - self.assertEqual(data, p[Raw].load) - - # out2in - if not dont_translate: - dst_addr = self.nat_addr - else: - dst_addr = self.pg0.remote_ip4 - if proto != IP_PROTOS.icmp: - sport = 20 - dport = p[layer].sport - else: - sport = p[layer].id - dport = 0 - pkts = self.create_stream_frag(self.pg1, dst_addr, sport, dport, - data, proto, echo_reply=True) - pkts.reverse() - self.pg1.add_stream(pkts) - self.pg_enable_capture(self.pg_interfaces) - self.logger.info(self.vapi.cli("show trace")) - self.pg_start() - frags = self.pg0.get_capture(len(pkts)) - p = self.reass_frags_and_verify(frags, - self.pg1.remote_ip4, - self.pg0.remote_ip4) - if proto != IP_PROTOS.icmp: - self.assertEqual(p[layer].sport, 20) - self.assertEqual(p[layer].dport, self.port_in) - else: - self.assertEqual(p[layer].id, self.port_in) - self.assertEqual(data, p[Raw].load) - - def reass_frags_and_verify(self, frags, src, dst): - buffer = BytesIO() - for p in frags: - self.assertEqual(p[IP].src, src) - self.assertEqual(p[IP].dst, dst) - self.assert_ip_checksum_valid(p) - buffer.seek(p[IP].frag * 8) - buffer.write(bytes(p[IP].payload)) - ip = IP(src=frags[0][IP].src, dst=frags[0][IP].dst, - proto=frags[0][IP].proto) - if ip.proto == IP_PROTOS.tcp: - p = (ip / TCP(buffer.getvalue())) - self.logger.debug(ppp("Reassembled:", p)) - self.assert_tcp_checksum_valid(p) - elif ip.proto == IP_PROTOS.udp: - p = (ip / UDP(buffer.getvalue()[:8]) / - Raw(buffer.getvalue()[8:])) - elif ip.proto == IP_PROTOS.icmp: - p = (ip / ICMP(buffer.getvalue())) - return p - - def frag_in_order(self, proto=IP_PROTOS.tcp, dont_translate=False, - ignore_port=False): - layer = self.proto2layer(proto) - - if proto == IP_PROTOS.tcp: - data = b"A" * 4 + b"B" * 16 + b"C" * 3 - else: - data = b"A" * 16 + b"B" * 16 + b"C" * 3 - self.port_in = self.random_port() - - # in2out - pkts = self.create_stream_frag(self.pg0, self.pg1.remote_ip4, - self.port_in, 20, data, proto) - self.pg0.add_stream(pkts) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - frags = self.pg1.get_capture(len(pkts)) - if not dont_translate: - p = self.reass_frags_and_verify(frags, - self.nat_addr, - self.pg1.remote_ip4) - else: - p = self.reass_frags_and_verify(frags, - self.pg0.remote_ip4, - self.pg1.remote_ip4) - if proto != IP_PROTOS.icmp: - if not dont_translate: - self.assertEqual(p[layer].dport, 20) - if not ignore_port: - self.assertNotEqual(p[layer].sport, self.port_in) - else: - self.assertEqual(p[layer].sport, self.port_in) - else: - if not ignore_port: - if not dont_translate: - self.assertNotEqual(p[layer].id, self.port_in) - else: - self.assertEqual(p[layer].id, self.port_in) - self.assertEqual(data, p[Raw].load) - - # out2in - if not dont_translate: - dst_addr = self.nat_addr - else: - dst_addr = self.pg0.remote_ip4 - if proto != IP_PROTOS.icmp: - sport = 20 - dport = p[layer].sport - else: - sport = p[layer].id - dport = 0 - pkts = self.create_stream_frag(self.pg1, dst_addr, sport, dport, data, - proto, echo_reply=True) - self.pg1.add_stream(pkts) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - frags = self.pg0.get_capture(len(pkts)) - p = self.reass_frags_and_verify(frags, - self.pg1.remote_ip4, - self.pg0.remote_ip4) - if proto != IP_PROTOS.icmp: - self.assertEqual(p[layer].sport, 20) - self.assertEqual(p[layer].dport, self.port_in) - else: - self.assertEqual(p[layer].id, self.port_in) - self.assertEqual(data, p[Raw].load) - - def verify_capture_out(self, capture, nat_ip=None, same_port=False, - dst_ip=None, ignore_port=False): - if nat_ip is None: - nat_ip = self.nat_addr - for packet in capture: - try: - self.assert_packet_checksums_valid(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 not ignore_port: - 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.assert_packet_checksums_valid(packet) - elif packet.haslayer(UDP): - if not ignore_port: - 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 not ignore_port: - 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.assert_packet_checksums_valid(packet) - except: - self.logger.error(ppp("Unexpected or invalid packet " - "(outside network):", packet)) - raise - - def verify_capture_in(self, capture, in_if): - for packet in capture: - try: - self.assert_packet_checksums_valid(packet) - self.assertEqual(packet[IP].dst, in_if.remote_ip4) - if packet.haslayer(TCP): - self.assertEqual(packet[TCP].dport, self.tcp_port_in) - elif packet.haslayer(UDP): - self.assertEqual(packet[UDP].dport, self.udp_port_in) - else: - self.assertEqual(packet[ICMP].id, self.icmp_id_in) - except: - self.logger.error(ppp("Unexpected or invalid packet " - "(inside network):", packet)) - raise - - def create_stream_in(self, in_if, out_if, dst_ip=None, ttl=64): - if dst_ip is None: - dst_ip = out_if.remote_ip4 - - pkts = [] - # TCP - p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) / - IP(src=in_if.remote_ip4, dst=dst_ip, ttl=ttl) / - TCP(sport=self.tcp_port_in, dport=20)) - pkts.extend([p, p]) - - # UDP - p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) / - IP(src=in_if.remote_ip4, dst=dst_ip, 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=dst_ip, 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, - use_inside_ports=False): - 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 create_tcp_stream(self, in_if, out_if, count): - pkts = [] - port = 6303 - - for i in range(count): - p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) / - IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=64) / - TCP(sport=port + i, dport=20)) - pkts.append(p) - - return pkts - - def create_stream_frag(self, src_if, dst, sport, dport, data, - proto=IP_PROTOS.tcp, echo_reply=False): - if proto == IP_PROTOS.tcp: - p = (IP(src=src_if.remote_ip4, dst=dst) / - TCP(sport=sport, dport=dport) / - Raw(data)) - p = p.__class__(scapy.compat.raw(p)) - chksum = p[TCP].chksum - proto_header = TCP(sport=sport, dport=dport, chksum=chksum) - elif proto == IP_PROTOS.udp: - proto_header = UDP(sport=sport, dport=dport) - elif proto == IP_PROTOS.icmp: - if not echo_reply: - proto_header = ICMP(id=sport, type='echo-request') - else: - proto_header = ICMP(id=sport, type='echo-reply') - else: - raise Exception("Unsupported protocol") - id = self.random_port() - pkts = [] - if proto == IP_PROTOS.tcp: - raw = Raw(data[0:4]) - else: - raw = Raw(data[0:16]) - p = (Ether(src=src_if.remote_mac, dst=src_if.local_mac) / - IP(src=src_if.remote_ip4, dst=dst, flags="MF", frag=0, id=id) / - proto_header / - raw) - pkts.append(p) - if proto == IP_PROTOS.tcp: - raw = Raw(data[4:20]) - else: - raw = Raw(data[16:32]) - p = (Ether(src=src_if.remote_mac, dst=src_if.local_mac) / - IP(src=src_if.remote_ip4, dst=dst, flags="MF", frag=3, id=id, - proto=proto) / - raw) - pkts.append(p) - if proto == IP_PROTOS.tcp: - raw = Raw(data[20:]) - else: - raw = Raw(data[32:]) - p = (Ether(src=src_if.remote_mac, dst=src_if.local_mac) / - IP(src=src_if.remote_ip4, dst=dst, frag=5, proto=proto, - id=id) / - raw) - pkts.append(p) - return pkts - - def frag_in_order_in_plus_out(self, in_addr, out_addr, in_port, out_port, - proto=IP_PROTOS.tcp): - - layer = self.proto2layer(proto) - - if proto == IP_PROTOS.tcp: - data = b"A" * 4 + b"B" * 16 + b"C" * 3 - else: - data = b"A" * 16 + b"B" * 16 + b"C" * 3 - port_in = self.random_port() - - for i in range(2): - # out2in - pkts = self.create_stream_frag(self.pg0, out_addr, - port_in, out_port, - data, proto) - 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.pg0.remote_ip4, - in_addr) - if proto != IP_PROTOS.icmp: - self.assertEqual(p[layer].sport, port_in) - self.assertEqual(p[layer].dport, in_port) - else: - self.assertEqual(p[layer].id, port_in) - self.assertEqual(data, p[Raw].load) - - # in2out - if proto != IP_PROTOS.icmp: - pkts = self.create_stream_frag(self.pg1, self.pg0.remote_ip4, - in_port, - p[layer].sport, data, proto) - else: - pkts = self.create_stream_frag(self.pg1, self.pg0.remote_ip4, - p[layer].id, 0, data, proto, - echo_reply=True) - self.pg1.add_stream(pkts) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - frags = self.pg0.get_capture(len(pkts)) - p = self.reass_frags_and_verify(frags, - out_addr, - self.pg0.remote_ip4) - if proto != IP_PROTOS.icmp: - self.assertEqual(p[layer].sport, out_port) - self.assertEqual(p[layer].dport, port_in) - else: - self.assertEqual(p[layer].id, port_in) - self.assertEqual(data, p[Raw].load) - - def frag_out_of_order_in_plus_out(self, in_addr, out_addr, in_port, - out_port, proto=IP_PROTOS.tcp): - - layer = self.proto2layer(proto) - - if proto == IP_PROTOS.tcp: - data = b"A" * 4 + b"B" * 16 + b"C" * 3 - else: - data = b"A" * 16 + b"B" * 16 + b"C" * 3 - port_in = self.random_port() - - for i in range(2): - # out2in - pkts = self.create_stream_frag(self.pg0, out_addr, - port_in, out_port, - data, proto) - 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.pg0.remote_ip4, - in_addr) - if proto != IP_PROTOS.icmp: - self.assertEqual(p[layer].dport, in_port) - self.assertEqual(p[layer].sport, port_in) - self.assertEqual(p[layer].dport, in_port) - else: - self.assertEqual(p[layer].id, port_in) - self.assertEqual(data, p[Raw].load) - - # in2out - if proto != IP_PROTOS.icmp: - pkts = self.create_stream_frag(self.pg1, self.pg0.remote_ip4, - in_port, - p[layer].sport, data, proto) - else: - pkts = self.create_stream_frag(self.pg1, self.pg0.remote_ip4, - p[layer].id, 0, data, proto, - echo_reply=True) - pkts.reverse() - self.pg1.add_stream(pkts) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - frags = self.pg0.get_capture(len(pkts)) - p = self.reass_frags_and_verify(frags, - out_addr, - self.pg0.remote_ip4) - if proto != IP_PROTOS.icmp: - self.assertEqual(p[layer].sport, out_port) - self.assertEqual(p[layer].dport, port_in) - else: - self.assertEqual(p[layer].id, port_in) - self.assertEqual(data, p[Raw].load) - - def init_tcp_session(self, in_if, out_if, in_port, ext_port): - # 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=in_port, dport=ext_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] - out_port = 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=ext_port, dport=out_port, 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=in_port, dport=ext_port, flags="A")) - in_if.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - out_if.get_capture(1) - - return out_port - - def twice_nat_common(self, self_twice_nat=False, same_pg=False, lb=False, - client_id=None): - twice_nat_addr = '10.0.1.3' - - port_in = 8080 - if lb: - if not same_pg: - port_in1 = port_in - port_in2 = port_in - else: - port_in1 = port_in + 1 - port_in2 = port_in + 2 - - port_out = 80 - eh_port_out = 4567 - - server1 = self.pg0.remote_hosts[0] - server2 = self.pg0.remote_hosts[1] - if lb and same_pg: - server2 = server1 - if not lb: - server = server1 - - pg0 = self.pg0 - if same_pg: - pg1 = self.pg0 - else: - pg1 = self.pg1 - - eh_translate = ((not self_twice_nat) or (not lb and same_pg) or - client_id == 1) - - self.nat_add_address(self.nat_addr) - self.nat_add_address(twice_nat_addr, twice_nat=1) - - flags = 0 - if self_twice_nat: - flags |= self.config_flags.NAT_IS_SELF_TWICE_NAT - else: - flags |= self.config_flags.NAT_IS_TWICE_NAT - - if not lb: - self.nat_add_static_mapping(pg0.remote_ip4, self.nat_addr, - port_in, port_out, - proto=IP_PROTOS.tcp, - flags=flags) - else: - locals = [{'addr': server1.ip4, - 'port': port_in1, - 'probability': 50, - 'vrf_id': 0}, - {'addr': server2.ip4, - 'port': port_in2, - 'probability': 50, - 'vrf_id': 0}] - out_addr = self.nat_addr - - self.vapi.nat44_add_del_lb_static_mapping(is_add=1, flags=flags, - external_addr=out_addr, - external_port=port_out, - protocol=IP_PROTOS.tcp, - local_num=len(locals), - locals=locals) - self.nat_add_inside_interface(pg0) - self.nat_add_outside_interface(pg1) - - if same_pg: - if not lb: - client = server - else: - assert client_id is not None - if client_id == 1: - client = self.pg0.remote_hosts[0] - elif client_id == 2: - client = self.pg0.remote_hosts[1] - else: - client = pg1.remote_hosts[0] - p = (Ether(src=pg1.remote_mac, dst=pg1.local_mac) / - IP(src=client.ip4, dst=self.nat_addr) / - TCP(sport=eh_port_out, dport=port_out)) - pg1.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = pg0.get_capture(1) - p = capture[0] - try: - ip = p[IP] - tcp = p[TCP] - if lb: - if ip.dst == server1.ip4: - server = server1 - port_in = port_in1 - else: - server = server2 - port_in = port_in2 - self.assertEqual(ip.dst, server.ip4) - if lb and same_pg: - self.assertIn(tcp.dport, [port_in1, port_in2]) - else: - self.assertEqual(tcp.dport, port_in) - if eh_translate: - self.assertEqual(ip.src, twice_nat_addr) - self.assertNotEqual(tcp.sport, eh_port_out) - else: - self.assertEqual(ip.src, client.ip4) - self.assertEqual(tcp.sport, eh_port_out) - eh_addr_in = ip.src - eh_port_in = tcp.sport - saved_port_in = tcp.dport - self.assert_packet_checksums_valid(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - p = (Ether(src=server.mac, dst=pg0.local_mac) / - IP(src=server.ip4, dst=eh_addr_in) / - TCP(sport=saved_port_in, dport=eh_port_in)) - pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = pg1.get_capture(1) - p = capture[0] - try: - ip = p[IP] - tcp = p[TCP] - self.assertEqual(ip.dst, client.ip4) - self.assertEqual(ip.src, self.nat_addr) - self.assertEqual(tcp.dport, eh_port_out) - self.assertEqual(tcp.sport, port_out) - self.assert_packet_checksums_valid(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - if eh_translate: - sessions = self.vapi.nat44_user_session_dump(server.ip4, 0) - self.assertEqual(len(sessions), 1) - self.assertTrue(sessions[0].flags & - self.config_flags.NAT_IS_EXT_HOST_VALID) - self.assertTrue(sessions[0].flags & - self.config_flags.NAT_IS_TWICE_NAT) - self.logger.info(self.vapi.cli("show nat44 sessions")) - self.vapi.nat44_del_session( - address=sessions[0].inside_ip_address, - port=sessions[0].inside_port, - protocol=sessions[0].protocol, - flags=(self.config_flags.NAT_IS_INSIDE | - self.config_flags.NAT_IS_EXT_HOST_VALID), - ext_host_address=sessions[0].ext_host_nat_address, - ext_host_port=sessions[0].ext_host_nat_port) - sessions = self.vapi.nat44_user_session_dump(server.ip4, 0) - self.assertEqual(len(sessions), 0) - - def verify_syslog_sess(self, data, is_add=True, is_ip6=False): - message = data.decode('utf-8') - try: - message = SyslogMessage.parse(message) - except ParseError as e: - self.logger.error(e) - raise - else: - self.assertEqual(message.severity, SyslogSeverity.info) - self.assertEqual(message.appname, 'NAT') - self.assertEqual(message.msgid, 'SADD' if is_add else 'SDEL') - sd_params = message.sd.get('nsess') - self.assertTrue(sd_params is not None) - if is_ip6: - self.assertEqual(sd_params.get('IATYP'), 'IPv6') - self.assertEqual(sd_params.get('ISADDR'), self.pg0.remote_ip6) - else: - self.assertEqual(sd_params.get('IATYP'), 'IPv4') - self.assertEqual(sd_params.get('ISADDR'), self.pg0.remote_ip4) - self.assertTrue(sd_params.get('SSUBIX') is not None) - self.assertEqual(sd_params.get('ISPORT'), "%d" % self.tcp_port_in) - self.assertEqual(sd_params.get('XATYP'), 'IPv4') - self.assertEqual(sd_params.get('XSADDR'), self.nat_addr) - self.assertEqual(sd_params.get('XSPORT'), "%d" % self.tcp_port_out) - self.assertEqual(sd_params.get('PROTO'), "%d" % IP_PROTOS.tcp) - self.assertEqual(sd_params.get('SVLAN'), '0') - self.assertEqual(sd_params.get('XDADDR'), self.pg1.remote_ip4) - self.assertEqual(sd_params.get('XDPORT'), - "%d" % self.tcp_external_port) - - -class TestNAT44ED(NAT44EDTestCase): - """ NAT44ED Test Case """ - - def test_users_dump(self): - """ NAT44ED API test - nat44_user_dump """ - - self.nat_add_address(self.nat_addr) - self.nat_add_inside_interface(self.pg0) - self.nat_add_outside_interface(self.pg1) - - self.vapi.nat44_forwarding_enable_disable(enable=1) - - local_ip = self.pg0.remote_ip4 - external_ip = self.nat_addr - self.nat_add_static_mapping(local_ip, external_ip) - - users = self.vapi.nat44_user_dump() - self.assertEqual(len(users), 0) - - # in2out - static mapping match - - 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) - - 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, same_port=True) - - users = self.vapi.nat44_user_dump() - self.assertEqual(len(users), 1) - static_user = users[0] - self.assertEqual(static_user.nstaticsessions, 3) - self.assertEqual(static_user.nsessions, 0) - - # in2out - no static mapping match (forwarding test) - - host0 = self.pg0.remote_hosts[0] - self.pg0.remote_hosts[0] = self.pg0.remote_hosts[1] - try: - pkts = self.create_stream_out(self.pg1, - dst_ip=self.pg0.remote_ip4, - use_inside_ports=True) - 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) - - 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=self.pg0.remote_ip4, - same_port=True) - finally: - self.pg0.remote_hosts[0] = host0 - - users = self.vapi.nat44_user_dump() - self.assertEqual(len(users), 2) - if str(users[0].ip_address) == self.pg0.remote_hosts[0].ip4: - non_static_user = users[1] - static_user = users[0] - else: - non_static_user = users[0] - static_user = users[1] - self.assertEqual(static_user.nstaticsessions, 3) - self.assertEqual(static_user.nsessions, 0) - self.assertEqual(non_static_user.nstaticsessions, 0) - self.assertEqual(non_static_user.nsessions, 3) - - users = self.vapi.nat44_user_dump() - self.assertEqual(len(users), 2) - if str(users[0].ip_address) == self.pg0.remote_hosts[0].ip4: - non_static_user = users[1] - static_user = users[0] - else: - non_static_user = users[0] - static_user = users[1] - self.assertEqual(static_user.nstaticsessions, 3) - self.assertEqual(static_user.nsessions, 0) - self.assertEqual(non_static_user.nstaticsessions, 0) - self.assertEqual(non_static_user.nsessions, 3) - - def test_frag_out_of_order_do_not_translate(self): - """ NAT44ED don't translate fragments arriving out of order """ - self.nat_add_inside_interface(self.pg0) - self.nat_add_outside_interface(self.pg1) - self.vapi.nat44_forwarding_enable_disable(enable=True) - self.frag_out_of_order(proto=IP_PROTOS.tcp, dont_translate=True) - - def test_forwarding(self): - """ NAT44ED forwarding test """ - - self.nat_add_inside_interface(self.pg0) - self.nat_add_outside_interface(self.pg1) - self.vapi.nat44_forwarding_enable_disable(enable=1) - - real_ip = self.pg0.remote_ip4 - alias_ip = self.nat_addr - flags = self.config_flags.NAT_IS_ADDR_ONLY - self.vapi.nat44_add_del_static_mapping(is_add=1, - local_ip_address=real_ip, - external_ip_address=alias_ip, - external_sw_if_index=0xFFFFFFFF, - flags=flags) - - try: - # in2out - static mapping match - - 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) - - 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, same_port=True) - - # in2out - no static mapping match - - host0 = self.pg0.remote_hosts[0] - self.pg0.remote_hosts[0] = self.pg0.remote_hosts[1] - try: - pkts = self.create_stream_out(self.pg1, - dst_ip=self.pg0.remote_ip4, - use_inside_ports=True) - 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) - - 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=self.pg0.remote_ip4, - same_port=True) - finally: - self.pg0.remote_hosts[0] = host0 - - user = self.pg0.remote_hosts[1] - sessions = self.vapi.nat44_user_session_dump(user.ip4, 0) - self.assertEqual(len(sessions), 3) - self.assertTrue(sessions[0].flags & - self.config_flags.NAT_IS_EXT_HOST_VALID) - self.vapi.nat44_del_session( - address=sessions[0].inside_ip_address, - port=sessions[0].inside_port, - protocol=sessions[0].protocol, - flags=(self.config_flags.NAT_IS_INSIDE | - self.config_flags.NAT_IS_EXT_HOST_VALID), - ext_host_address=sessions[0].ext_host_address, - ext_host_port=sessions[0].ext_host_port) - sessions = self.vapi.nat44_user_session_dump(user.ip4, 0) - self.assertEqual(len(sessions), 2) - - finally: - self.vapi.nat44_forwarding_enable_disable(enable=0) - flags = self.config_flags.NAT_IS_ADDR_ONLY - self.vapi.nat44_add_del_static_mapping( - is_add=0, - local_ip_address=real_ip, - external_ip_address=alias_ip, - external_sw_if_index=0xFFFFFFFF, - flags=flags) - - def test_output_feature_and_service2(self): - """ NAT44ED interface output feature and service host direct access """ - self.vapi.nat44_forwarding_enable_disable(enable=1) - self.nat_add_address(self.nat_addr) - - self.vapi.nat44_interface_add_del_output_feature( - sw_if_index=self.pg1.sw_if_index, is_add=1,) - - # session initiated from service host - translate - 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, ignore_port=True) - - 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) - - # session initiated from remote host - do not translate - tcp_port_in = self.tcp_port_in - udp_port_in = self.udp_port_in - icmp_id_in = self.icmp_id_in - - self.tcp_port_in = 60303 - self.udp_port_in = 60304 - self.icmp_id_in = 60305 - - try: - pkts = self.create_stream_out(self.pg1, - self.pg0.remote_ip4, - use_inside_ports=True) - 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) - - 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=self.pg0.remote_ip4, - same_port=True) - finally: - self.tcp_port_in = tcp_port_in - self.udp_port_in = udp_port_in - self.icmp_id_in = icmp_id_in - - def test_twice_nat(self): - """ NAT44ED Twice NAT """ - self.twice_nat_common() - - def test_self_twice_nat_positive(self): - """ NAT44ED Self Twice NAT (positive test) """ - self.twice_nat_common(self_twice_nat=True, same_pg=True) - - def test_self_twice_nat_lb_positive(self): - """ NAT44ED Self Twice NAT local service load balancing (positive test) - """ - self.twice_nat_common(lb=True, self_twice_nat=True, same_pg=True, - client_id=1) - - def test_twice_nat_lb(self): - """ NAT44ED Twice NAT local service load balancing """ - self.twice_nat_common(lb=True) - - def test_output_feature(self): - """ NAT44ED interface output feature (in2out postrouting) """ - self.vapi.nat44_forwarding_enable_disable(enable=1) - self.nat_add_address(self.nat_addr) - - self.nat_add_outside_interface(self.pg0) - self.vapi.nat44_interface_add_del_output_feature( - sw_if_index=self.pg1.sw_if_index, is_add=1) - - # 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, ignore_port=True) - - # 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_static_with_port_out2(self): - """ NAT44ED 1:1 NAPT asymmetrical rule """ - - external_port = 80 - local_port = 8080 - - self.vapi.nat44_forwarding_enable_disable(enable=1) - flags = self.config_flags.NAT_IS_OUT2IN_ONLY - self.nat_add_static_mapping(self.pg0.remote_ip4, self.nat_addr, - local_port, external_port, - proto=IP_PROTOS.tcp, flags=flags) - - self.nat_add_inside_interface(self.pg0) - self.nat_add_outside_interface(self.pg1) - - # from client to service - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - TCP(sport=12345, dport=external_port)) - 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.dst, self.pg0.remote_ip4) - self.assertEqual(tcp.dport, local_port) - self.assert_packet_checksums_valid(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - # ICMP error - p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / - IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / - ICMP(type=11) / capture[0][IP]) - 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: - self.assertEqual(p[IP].src, self.nat_addr) - inner = p[IPerror] - self.assertEqual(inner.dst, self.nat_addr) - self.assertEqual(inner[TCPerror].dport, external_port) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - # from service back to client - 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=local_port, dport=12345)) - 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, self.nat_addr) - self.assertEqual(tcp.sport, external_port) - self.assert_packet_checksums_valid(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - # ICMP error - p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - ICMP(type=11) / capture[0][IP]) - 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: - self.assertEqual(p[IP].dst, self.pg0.remote_ip4) - inner = p[IPerror] - self.assertEqual(inner.src, self.pg0.remote_ip4) - self.assertEqual(inner[TCPerror].sport, local_port) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - # from client to server (no translation) - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=self.pg0.remote_ip4) / - TCP(sport=12346, dport=local_port)) - 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.dst, self.pg0.remote_ip4) - self.assertEqual(tcp.dport, local_port) - self.assert_packet_checksums_valid(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - # from service back to client (no translation) - 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=local_port, dport=12346)) - 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, self.pg0.remote_ip4) - self.assertEqual(tcp.sport, local_port) - self.assert_packet_checksums_valid(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - def test_static_lb(self): - """ NAT44ED local service load balancing """ - external_addr_n = self.nat_addr - external_port = 80 - local_port = 8080 - server1 = self.pg0.remote_hosts[0] - server2 = self.pg0.remote_hosts[1] - - locals = [{'addr': server1.ip4, - 'port': local_port, - 'probability': 70, - 'vrf_id': 0}, - {'addr': server2.ip4, - 'port': local_port, - 'probability': 30, - 'vrf_id': 0}] - - self.nat_add_address(self.nat_addr) - self.vapi.nat44_add_del_lb_static_mapping( - is_add=1, - external_addr=external_addr_n, - external_port=external_port, - protocol=IP_PROTOS.tcp, - local_num=len(locals), - locals=locals) - flags = self.config_flags.NAT_IS_INSIDE - self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg0.sw_if_index, - flags=flags, is_add=1) - self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg1.sw_if_index, - is_add=1) - - # from client to service - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - TCP(sport=12345, dport=external_port)) - self.pg1.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg0.get_capture(1) - p = capture[0] - server = None - try: - ip = p[IP] - tcp = p[TCP] - self.assertIn(ip.dst, [server1.ip4, server2.ip4]) - if ip.dst == server1.ip4: - server = server1 - else: - server = server2 - self.assertEqual(tcp.dport, local_port) - self.assert_packet_checksums_valid(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - # from service back to client - p = (Ether(src=server.mac, dst=self.pg0.local_mac) / - IP(src=server.ip4, dst=self.pg1.remote_ip4) / - TCP(sport=local_port, dport=12345)) - 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, self.nat_addr) - self.assertEqual(tcp.sport, external_port) - self.assert_packet_checksums_valid(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - sessions = self.vapi.nat44_user_session_dump(server.ip4, 0) - self.assertEqual(len(sessions), 1) - self.assertTrue(sessions[0].flags & - self.config_flags.NAT_IS_EXT_HOST_VALID) - self.vapi.nat44_del_session( - address=sessions[0].inside_ip_address, - port=sessions[0].inside_port, - protocol=sessions[0].protocol, - flags=(self.config_flags.NAT_IS_INSIDE | - self.config_flags.NAT_IS_EXT_HOST_VALID), - ext_host_address=sessions[0].ext_host_address, - ext_host_port=sessions[0].ext_host_port) - sessions = self.vapi.nat44_user_session_dump(server.ip4, 0) - self.assertEqual(len(sessions), 0) - - def test_static_lb_2(self): - """ NAT44ED local service load balancing (asymmetrical rule) """ - external_addr = self.nat_addr - external_port = 80 - local_port = 8080 - server1 = self.pg0.remote_hosts[0] - server2 = self.pg0.remote_hosts[1] - - locals = [{'addr': server1.ip4, - 'port': local_port, - 'probability': 70, - 'vrf_id': 0}, - {'addr': server2.ip4, - 'port': local_port, - 'probability': 30, - 'vrf_id': 0}] - - self.vapi.nat44_forwarding_enable_disable(enable=1) - flags = self.config_flags.NAT_IS_OUT2IN_ONLY - self.vapi.nat44_add_del_lb_static_mapping(is_add=1, flags=flags, - external_addr=external_addr, - external_port=external_port, - protocol=IP_PROTOS.tcp, - local_num=len(locals), - locals=locals) - flags = self.config_flags.NAT_IS_INSIDE - self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg0.sw_if_index, - flags=flags, is_add=1) - self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg1.sw_if_index, - is_add=1) - - # from client to service - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - TCP(sport=12345, dport=external_port)) - self.pg1.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg0.get_capture(1) - p = capture[0] - server = None - try: - ip = p[IP] - tcp = p[TCP] - self.assertIn(ip.dst, [server1.ip4, server2.ip4]) - if ip.dst == server1.ip4: - server = server1 - else: - server = server2 - self.assertEqual(tcp.dport, local_port) - self.assert_packet_checksums_valid(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - # from service back to client - p = (Ether(src=server.mac, dst=self.pg0.local_mac) / - IP(src=server.ip4, dst=self.pg1.remote_ip4) / - TCP(sport=local_port, dport=12345)) - 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, self.nat_addr) - self.assertEqual(tcp.sport, external_port) - self.assert_packet_checksums_valid(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - # from client to server (no translation) - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=server1.ip4) / - TCP(sport=12346, dport=local_port)) - self.pg1.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg0.get_capture(1) - p = capture[0] - server = None - try: - ip = p[IP] - tcp = p[TCP] - self.assertEqual(ip.dst, server1.ip4) - self.assertEqual(tcp.dport, local_port) - self.assert_packet_checksums_valid(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - # from service back to client (no translation) - p = (Ether(src=server1.mac, dst=self.pg0.local_mac) / - IP(src=server1.ip4, dst=self.pg1.remote_ip4) / - TCP(sport=local_port, dport=12346)) - 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, server1.ip4) - self.assertEqual(tcp.sport, local_port) - self.assert_packet_checksums_valid(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - def test_lb_affinity(self): - """ NAT44ED local service load balancing affinity """ - external_addr = self.nat_addr - external_port = 80 - local_port = 8080 - server1 = self.pg0.remote_hosts[0] - server2 = self.pg0.remote_hosts[1] - - locals = [{'addr': server1.ip4, - 'port': local_port, - 'probability': 50, - 'vrf_id': 0}, - {'addr': server2.ip4, - 'port': local_port, - 'probability': 50, - 'vrf_id': 0}] - - self.nat_add_address(self.nat_addr) - self.vapi.nat44_add_del_lb_static_mapping(is_add=1, - external_addr=external_addr, - external_port=external_port, - protocol=IP_PROTOS.tcp, - affinity=10800, - local_num=len(locals), - locals=locals) - flags = self.config_flags.NAT_IS_INSIDE - self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg0.sw_if_index, - flags=flags, is_add=1) - self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg1.sw_if_index, - is_add=1) - - p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - TCP(sport=1025, dport=external_port)) - self.pg1.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg0.get_capture(1) - backend = capture[0][IP].dst - - sessions = self.vapi.nat44_user_session_dump(backend, 0) - self.assertEqual(len(sessions), 1) - self.assertTrue(sessions[0].flags & - self.config_flags.NAT_IS_EXT_HOST_VALID) - self.vapi.nat44_del_session( - address=sessions[0].inside_ip_address, - port=sessions[0].inside_port, - protocol=sessions[0].protocol, - flags=(self.config_flags.NAT_IS_INSIDE | - self.config_flags.NAT_IS_EXT_HOST_VALID), - ext_host_address=sessions[0].ext_host_address, - ext_host_port=sessions[0].ext_host_port) - - pkts = [] - for port in range(1030, 1100): - p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - TCP(sport=port, dport=external_port)) - pkts.append(p) - self.pg1.add_stream(pkts) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg0.get_capture(len(pkts)) - for p in capture: - self.assertEqual(p[IP].dst, backend) - - def test_multiple_vrf(self): - """ NAT44ED Multiple VRF setup """ - - external_addr = '1.2.3.4' - external_port = 80 - local_port = 8080 - port = 0 - - self.vapi.nat44_forwarding_enable_disable(enable=1) - self.nat_add_address(self.nat_addr) - flags = self.config_flags.NAT_IS_INSIDE - self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg0.sw_if_index, - is_add=1) - self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg0.sw_if_index, - is_add=1, flags=flags) - self.vapi.nat44_interface_add_del_output_feature( - sw_if_index=self.pg1.sw_if_index, - is_add=1) - self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg5.sw_if_index, - is_add=1) - self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg5.sw_if_index, - is_add=1, flags=flags) - self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg6.sw_if_index, - is_add=1) - flags = self.config_flags.NAT_IS_OUT2IN_ONLY - self.nat_add_static_mapping(self.pg5.remote_ip4, external_addr, - local_port, external_port, vrf_id=1, - proto=IP_PROTOS.tcp, flags=flags) - self.nat_add_static_mapping( - self.pg0.remote_ip4, - external_sw_if_index=self.pg0.sw_if_index, - local_port=local_port, - vrf_id=0, - external_port=external_port, - proto=IP_PROTOS.tcp, - flags=flags - ) - - # from client to service (both VRF1) - p = (Ether(src=self.pg6.remote_mac, dst=self.pg6.local_mac) / - IP(src=self.pg6.remote_ip4, dst=external_addr) / - TCP(sport=12345, dport=external_port)) - self.pg6.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg5.get_capture(1) - p = capture[0] - try: - ip = p[IP] - tcp = p[TCP] - self.assertEqual(ip.dst, self.pg5.remote_ip4) - self.assertEqual(tcp.dport, local_port) - self.assert_packet_checksums_valid(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - # from service back to client (both VRF1) - p = (Ether(src=self.pg5.remote_mac, dst=self.pg5.local_mac) / - IP(src=self.pg5.remote_ip4, dst=self.pg6.remote_ip4) / - TCP(sport=local_port, dport=12345)) - self.pg5.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg6.get_capture(1) - p = capture[0] - try: - ip = p[IP] - tcp = p[TCP] - self.assertEqual(ip.src, external_addr) - self.assertEqual(tcp.sport, external_port) - self.assert_packet_checksums_valid(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - # dynamic NAT from VRF1 to VRF0 (output-feature) - p = (Ether(src=self.pg5.remote_mac, dst=self.pg5.local_mac) / - IP(src=self.pg5.remote_ip4, dst=self.pg1.remote_ip4) / - TCP(sport=2345, dport=22)) - self.pg5.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, self.nat_addr) - self.assert_packet_checksums_valid(p) - port = tcp.sport - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - TCP(sport=22, dport=port)) - self.pg1.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg5.get_capture(1) - p = capture[0] - try: - ip = p[IP] - tcp = p[TCP] - self.assertEqual(ip.dst, self.pg5.remote_ip4) - self.assertEqual(tcp.dport, 2345) - self.assert_packet_checksums_valid(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - # from client VRF1 to service VRF0 - p = (Ether(src=self.pg6.remote_mac, dst=self.pg6.local_mac) / - IP(src=self.pg6.remote_ip4, dst=self.pg0.local_ip4) / - TCP(sport=12346, dport=external_port)) - self.pg6.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.dst, self.pg0.remote_ip4) - self.assertEqual(tcp.dport, local_port) - self.assert_packet_checksums_valid(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - # from service VRF0 back to client VRF1 - p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / - IP(src=self.pg0.remote_ip4, dst=self.pg6.remote_ip4) / - TCP(sport=local_port, dport=12346)) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg6.get_capture(1) - p = capture[0] - try: - ip = p[IP] - tcp = p[TCP] - self.assertEqual(ip.src, self.pg0.local_ip4) - self.assertEqual(tcp.sport, external_port) - self.assert_packet_checksums_valid(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - # from client VRF0 to service VRF1 - p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / - IP(src=self.pg0.remote_ip4, dst=external_addr) / - TCP(sport=12347, dport=external_port)) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg5.get_capture(1) - p = capture[0] - try: - ip = p[IP] - tcp = p[TCP] - self.assertEqual(ip.dst, self.pg5.remote_ip4) - self.assertEqual(tcp.dport, local_port) - self.assert_packet_checksums_valid(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - # from service VRF1 back to client VRF0 - p = (Ether(src=self.pg5.remote_mac, dst=self.pg5.local_mac) / - IP(src=self.pg5.remote_ip4, dst=self.pg0.remote_ip4) / - TCP(sport=local_port, dport=12347)) - self.pg5.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, external_addr) - self.assertEqual(tcp.sport, external_port) - self.assert_packet_checksums_valid(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - # from client to server (both VRF1, no translation) - p = (Ether(src=self.pg6.remote_mac, dst=self.pg6.local_mac) / - IP(src=self.pg6.remote_ip4, dst=self.pg5.remote_ip4) / - TCP(sport=12348, dport=local_port)) - self.pg6.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg5.get_capture(1) - p = capture[0] - try: - ip = p[IP] - tcp = p[TCP] - self.assertEqual(ip.dst, self.pg5.remote_ip4) - self.assertEqual(tcp.dport, local_port) - self.assert_packet_checksums_valid(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - # from server back to client (both VRF1, no translation) - p = (Ether(src=self.pg5.remote_mac, dst=self.pg5.local_mac) / - IP(src=self.pg5.remote_ip4, dst=self.pg6.remote_ip4) / - TCP(sport=local_port, dport=12348)) - self.pg5.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg6.get_capture(1) - p = capture[0] - try: - ip = p[IP] - tcp = p[TCP] - self.assertEqual(ip.src, self.pg5.remote_ip4) - self.assertEqual(tcp.sport, local_port) - self.assert_packet_checksums_valid(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - # from client VRF1 to server VRF0 (no translation) - p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / - IP(src=self.pg0.remote_ip4, dst=self.pg6.remote_ip4) / - TCP(sport=local_port, dport=12349)) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg6.get_capture(1) - p = capture[0] - try: - ip = p[IP] - tcp = p[TCP] - self.assertEqual(ip.src, self.pg0.remote_ip4) - self.assertEqual(tcp.sport, local_port) - self.assert_packet_checksums_valid(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - # from server VRF0 back to client VRF1 (no translation) - p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / - IP(src=self.pg0.remote_ip4, dst=self.pg6.remote_ip4) / - TCP(sport=local_port, dport=12349)) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg6.get_capture(1) - p = capture[0] - try: - ip = p[IP] - tcp = p[TCP] - self.assertEqual(ip.src, self.pg0.remote_ip4) - self.assertEqual(tcp.sport, local_port) - self.assert_packet_checksums_valid(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - # from client VRF0 to server VRF1 (no translation) - p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / - IP(src=self.pg0.remote_ip4, dst=self.pg5.remote_ip4) / - TCP(sport=12344, dport=local_port)) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg5.get_capture(1) - p = capture[0] - try: - ip = p[IP] - tcp = p[TCP] - self.assertEqual(ip.dst, self.pg5.remote_ip4) - self.assertEqual(tcp.dport, local_port) - self.assert_packet_checksums_valid(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - # from server VRF1 back to client VRF0 (no translation) - p = (Ether(src=self.pg5.remote_mac, dst=self.pg5.local_mac) / - IP(src=self.pg5.remote_ip4, dst=self.pg0.remote_ip4) / - TCP(sport=local_port, dport=12344)) - self.pg5.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.pg5.remote_ip4) - self.assertEqual(tcp.sport, local_port) - self.assert_packet_checksums_valid(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - def test_outside_address_distribution(self): - """ Outside address distribution based on source address """ - - x = 100 - nat_addresses = [] - - for i in range(1, x): - a = "10.0.0.%d" % i - nat_addresses.append(a) - - self.nat_add_inside_interface(self.pg0) - self.nat_add_outside_interface(self.pg1) - - self.vapi.nat44_add_del_address_range( - first_ip_address=nat_addresses[0], - last_ip_address=nat_addresses[-1], - vrf_id=0xFFFFFFFF, is_add=1, flags=0) - - self.pg0.generate_remote_hosts(x) - - pkts = [] - for i in range(x): - info = self.create_packet_info(self.pg0, self.pg1) - payload = self.info_to_payload(info) - p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / - IP(src=self.pg0.remote_hosts[i].ip4, - dst=self.pg1.remote_ip4) / - UDP(sport=7000+i, dport=8000+i) / - Raw(payload)) - info.data = p - pkts.append(p) - - self.pg0.add_stream(pkts) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - recvd = self.pg1.get_capture(len(pkts)) - for p_recvd in recvd: - payload_info = self.payload_to_info(p_recvd[Raw]) - packet_index = payload_info.index - info = self._packet_infos[packet_index] - self.assertTrue(info is not None) - self.assertEqual(packet_index, info.index) - p_sent = info.data - packed = socket.inet_aton(p_sent[IP].src) - numeric = struct.unpack("!L", packed)[0] - numeric = socket.htonl(numeric) - a = nat_addresses[(numeric-1) % len(nat_addresses)] - self.assertEqual( - a, p_recvd[IP].src, - "Invalid packet (src IP %s translated to %s, but expected %s)" - % (p_sent[IP].src, p_recvd[IP].src, a)) - - -class TestNAT44EDMW(TestNAT44ED): - """ NAT44ED MW Test Case """ - vpp_worker_count = 4 - max_sessions = 5000 - - @unittest.skip('MW fix required') - def test_users_dump(self): - """ NAT44ED API test - nat44_user_dump """ - - @unittest.skip('MW fix required') - def test_frag_out_of_order_do_not_translate(self): - """ NAT44ED don't translate fragments arriving out of order """ - - @unittest.skip('MW fix required') - def test_forwarding(self): - """ NAT44ED forwarding test """ - - @unittest.skip('MW fix required') - def test_twice_nat(self): - """ NAT44ED Twice NAT """ - - @unittest.skip('MW fix required') - def test_twice_nat_lb(self): - """ NAT44ED Twice NAT local service load balancing """ - - @unittest.skip('MW fix required') - def test_output_feature(self): - """ NAT44ED interface output feature (in2out postrouting) """ - - @unittest.skip('MW fix required') - def test_static_with_port_out2(self): - """ NAT44ED 1:1 NAPT asymmetrical rule """ - - @unittest.skip('MW fix required') - def test_output_feature_and_service2(self): - """ NAT44ED interface output feature and service host direct access """ - - @unittest.skip('MW fix required') - def test_static_lb(self): - """ NAT44ED local service load balancing """ - - @unittest.skip('MW fix required') - def test_static_lb_2(self): - """ NAT44ED local service load balancing (asymmetrical rule) """ - - @unittest.skip('MW fix required') - def test_lb_affinity(self): - """ NAT44ED local service load balancing affinity """ - - @unittest.skip('MW fix required') - def test_multiple_vrf(self): - """ NAT44ED Multiple VRF setup """ - - @unittest.skip('MW fix required') - def test_self_twice_nat_positive(self): - """ NAT44ED Self Twice NAT (positive test) """ - - @unittest.skip('MW fix required') - def test_self_twice_nat_lb_positive(self): - """ NAT44ED Self Twice NAT local service load balancing (positive test) - """ - - def test_dynamic(self): - """ NAT44ED dynamic translation test """ - pkt_count = 1500 - tcp_port_offset = 20 - udp_port_offset = 20 - icmp_id_offset = 20 - - self.nat_add_address(self.nat_addr) - self.nat_add_inside_interface(self.pg0) - self.nat_add_outside_interface(self.pg1) - - # in2out - tc1 = self.statistics['/nat44-ed/in2out/slowpath/tcp'] - uc1 = self.statistics['/nat44-ed/in2out/slowpath/udp'] - ic1 = self.statistics['/nat44-ed/in2out/slowpath/icmp'] - dc1 = self.statistics['/nat44-ed/in2out/slowpath/drops'] - - i2o_pkts = [[] for x in range(0, self.vpp_worker_count)] - - for i in range(pkt_count): - 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=tcp_port_offset + i, dport=20)) - i2o_pkts[p[TCP].sport % self.vpp_worker_count].append(p) - - p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / - IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / - UDP(sport=udp_port_offset + i, dport=20)) - i2o_pkts[p[UDP].sport % self.vpp_worker_count].append(p) - - p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / - IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / - ICMP(id=icmp_id_offset + i, type='echo-request')) - i2o_pkts[p[ICMP].id % self.vpp_worker_count].append(p) - - for i in range(0, self.vpp_worker_count): - if len(i2o_pkts[i]) > 0: - self.pg0.add_stream(i2o_pkts[i], worker=i) - - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg1.get_capture(pkt_count * 3) - - if_idx = self.pg0.sw_if_index - tc2 = self.statistics['/nat44-ed/in2out/slowpath/tcp'] - uc2 = self.statistics['/nat44-ed/in2out/slowpath/udp'] - ic2 = self.statistics['/nat44-ed/in2out/slowpath/icmp'] - dc2 = self.statistics['/nat44-ed/in2out/slowpath/drops'] - - self.assertEqual( - tc2[:, if_idx].sum() - tc1[:, if_idx].sum(), pkt_count) - self.assertEqual( - uc2[:, if_idx].sum() - uc1[:, if_idx].sum(), pkt_count) - self.assertEqual( - ic2[:, if_idx].sum() - ic1[:, if_idx].sum(), pkt_count) - self.assertEqual(dc2[:, if_idx].sum() - dc1[:, if_idx].sum(), 0) - - self.logger.info(self.vapi.cli("show trace")) - - # out2in - tc1 = self.statistics['/nat44-ed/out2in/fastpath/tcp'] - uc1 = self.statistics['/nat44-ed/out2in/fastpath/udp'] - ic1 = self.statistics['/nat44-ed/out2in/fastpath/icmp'] - dc1 = self.statistics['/nat44-ed/out2in/fastpath/drops'] - - recvd_tcp_ports = set() - recvd_udp_ports = set() - recvd_icmp_ids = set() - - for p in capture: - if TCP in p: - recvd_tcp_ports.add(p[TCP].sport) - if UDP in p: - recvd_udp_ports.add(p[UDP].sport) - if ICMP in p: - recvd_icmp_ids.add(p[ICMP].id) - - recvd_tcp_ports = list(recvd_tcp_ports) - recvd_udp_ports = list(recvd_udp_ports) - recvd_icmp_ids = list(recvd_icmp_ids) - - o2i_pkts = [[] for x in range(0, self.vpp_worker_count)] - for i in range(pkt_count): - p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - TCP(dport=choice(recvd_tcp_ports), sport=20)) - o2i_pkts[p[TCP].dport % self.vpp_worker_count].append(p) - - p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - UDP(dport=choice(recvd_udp_ports), sport=20)) - o2i_pkts[p[UDP].dport % self.vpp_worker_count].append(p) - - p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - ICMP(id=choice(recvd_icmp_ids), type='echo-reply')) - o2i_pkts[p[ICMP].id % self.vpp_worker_count].append(p) - - for i in range(0, self.vpp_worker_count): - if len(o2i_pkts[i]) > 0: - self.pg1.add_stream(o2i_pkts[i], worker=i) - - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg0.get_capture(pkt_count * 3) - for packet in capture: - try: - self.assert_packet_checksums_valid(packet) - self.assertEqual(packet[IP].dst, self.pg0.remote_ip4) - if packet.haslayer(TCP): - self.assert_in_range( - packet[TCP].dport, tcp_port_offset, - tcp_port_offset + pkt_count, "dst TCP port") - elif packet.haslayer(UDP): - self.assert_in_range( - packet[UDP].dport, udp_port_offset, - udp_port_offset + pkt_count, "dst UDP port") - else: - self.assert_in_range( - packet[ICMP].id, icmp_id_offset, - icmp_id_offset + pkt_count, "ICMP id") - except: - self.logger.error(ppp("Unexpected or invalid packet " - "(inside network):", packet)) - raise - - if_idx = self.pg1.sw_if_index - tc2 = self.statistics['/nat44-ed/out2in/fastpath/tcp'] - uc2 = self.statistics['/nat44-ed/out2in/fastpath/udp'] - ic2 = self.statistics['/nat44-ed/out2in/fastpath/icmp'] - dc2 = self.statistics['/nat44-ed/out2in/fastpath/drops'] - - self.assertEqual( - tc2[:, if_idx].sum() - tc1[:, if_idx].sum(), pkt_count) - self.assertEqual( - uc2[:, if_idx].sum() - uc1[:, if_idx].sum(), pkt_count) - self.assertEqual( - ic2[:, if_idx].sum() - ic1[:, if_idx].sum(), pkt_count) - self.assertEqual(dc2[:, if_idx].sum() - dc1[:, if_idx].sum(), 0) - - sc = self.statistics['/nat44-ed/total-sessions'] - self.assertEqual(sc[:, 0].sum(), len(recvd_tcp_ports) + - len(recvd_udp_ports) + len(recvd_icmp_ids)) - - def test_frag_in_order(self): - """ NAT44ED translate fragments arriving in order """ - - self.nat_add_address(self.nat_addr) - self.nat_add_inside_interface(self.pg0) - self.nat_add_outside_interface(self.pg1) - - self.frag_in_order(proto=IP_PROTOS.tcp, ignore_port=True) - self.frag_in_order(proto=IP_PROTOS.udp, ignore_port=True) - self.frag_in_order(proto=IP_PROTOS.icmp, ignore_port=True) - - def test_frag_in_order_do_not_translate(self): - """ NAT44ED don't translate fragments arriving in order """ - - self.nat_add_address(self.nat_addr) - self.nat_add_inside_interface(self.pg0) - self.nat_add_outside_interface(self.pg1) - self.vapi.nat44_forwarding_enable_disable(enable=True) - - self.frag_in_order(proto=IP_PROTOS.tcp, dont_translate=True) - - def test_frag_out_of_order(self): - """ NAT44ED translate fragments arriving out of order """ - - self.nat_add_address(self.nat_addr) - self.nat_add_inside_interface(self.pg0) - self.nat_add_outside_interface(self.pg1) - - self.frag_out_of_order(proto=IP_PROTOS.tcp, ignore_port=True) - self.frag_out_of_order(proto=IP_PROTOS.udp, ignore_port=True) - self.frag_out_of_order(proto=IP_PROTOS.icmp, ignore_port=True) - - def test_frag_in_order_in_plus_out(self): - """ NAT44ED in+out interface fragments in order """ - - in_port = self.random_port() - out_port = self.random_port() - - self.nat_add_address(self.nat_addr) - self.nat_add_inside_interface(self.pg0) - self.nat_add_outside_interface(self.pg0) - self.nat_add_inside_interface(self.pg1) - self.nat_add_outside_interface(self.pg1) - - # add static mappings for server - self.nat_add_static_mapping(self.server_addr, - self.nat_addr, - in_port, - out_port, - proto=IP_PROTOS.tcp) - self.nat_add_static_mapping(self.server_addr, - self.nat_addr, - in_port, - out_port, - proto=IP_PROTOS.udp) - self.nat_add_static_mapping(self.server_addr, - self.nat_addr, - proto=IP_PROTOS.icmp) - - # run tests for each protocol - self.frag_in_order_in_plus_out(self.server_addr, - self.nat_addr, - in_port, - out_port, - IP_PROTOS.tcp) - self.frag_in_order_in_plus_out(self.server_addr, - self.nat_addr, - in_port, - out_port, - IP_PROTOS.udp) - self.frag_in_order_in_plus_out(self.server_addr, - self.nat_addr, - in_port, - out_port, - IP_PROTOS.icmp) - - def test_frag_out_of_order_in_plus_out(self): - """ NAT44ED in+out interface fragments out of order """ - - in_port = self.random_port() - out_port = self.random_port() - - self.nat_add_address(self.nat_addr) - self.nat_add_inside_interface(self.pg0) - self.nat_add_outside_interface(self.pg0) - self.nat_add_inside_interface(self.pg1) - self.nat_add_outside_interface(self.pg1) - - # add static mappings for server - self.nat_add_static_mapping(self.server_addr, - self.nat_addr, - in_port, - out_port, - proto=IP_PROTOS.tcp) - self.nat_add_static_mapping(self.server_addr, - self.nat_addr, - in_port, - out_port, - proto=IP_PROTOS.udp) - self.nat_add_static_mapping(self.server_addr, - self.nat_addr, - proto=IP_PROTOS.icmp) - - # run tests for each protocol - self.frag_out_of_order_in_plus_out(self.server_addr, - self.nat_addr, - in_port, - out_port, - IP_PROTOS.tcp) - self.frag_out_of_order_in_plus_out(self.server_addr, - self.nat_addr, - in_port, - out_port, - IP_PROTOS.udp) - self.frag_out_of_order_in_plus_out(self.server_addr, - self.nat_addr, - in_port, - out_port, - IP_PROTOS.icmp) - - def test_reass_hairpinning(self): - """ NAT44ED fragments hairpinning """ - - server_addr = self.pg0.remote_hosts[1].ip4 - - host_in_port = self.random_port() - server_in_port = self.random_port() - server_out_port = self.random_port() - - self.nat_add_address(self.nat_addr) - self.nat_add_inside_interface(self.pg0) - self.nat_add_outside_interface(self.pg1) - - # add static mapping for server - self.nat_add_static_mapping(server_addr, self.nat_addr, - server_in_port, server_out_port, - proto=IP_PROTOS.tcp) - self.nat_add_static_mapping(server_addr, self.nat_addr, - server_in_port, server_out_port, - proto=IP_PROTOS.udp) - self.nat_add_static_mapping(server_addr, self.nat_addr) - - self.reass_hairpinning(server_addr, server_in_port, server_out_port, - host_in_port, proto=IP_PROTOS.tcp, - ignore_port=True) - self.reass_hairpinning(server_addr, server_in_port, server_out_port, - host_in_port, proto=IP_PROTOS.udp, - ignore_port=True) - self.reass_hairpinning(server_addr, server_in_port, server_out_port, - host_in_port, proto=IP_PROTOS.icmp, - ignore_port=True) - - def test_session_limit_per_vrf(self): - """ NAT44ED per vrf session limit """ - - inside = self.pg0 - inside_vrf10 = self.pg2 - outside = self.pg1 - - limit = 5 - - # 2 interfaces pg0, pg1 (vrf10, limit 1 tcp session) - # non existing vrf_id makes process core dump - self.vapi.nat44_set_session_limit(session_limit=limit, vrf_id=10) - - self.nat_add_inside_interface(inside) - self.nat_add_inside_interface(inside_vrf10) - self.nat_add_outside_interface(outside) - - # vrf independent - self.nat_add_interface_address(outside) - - # BUG: causing core dump - when bad vrf_id is specified - # self.nat_add_address(outside.local_ip4, vrf_id=20) - - stream = self.create_tcp_stream(inside_vrf10, outside, limit * 2) - inside_vrf10.add_stream(stream) - - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - - capture = outside.get_capture(limit) - - stream = self.create_tcp_stream(inside, outside, limit * 2) - inside.add_stream(stream) - - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - - capture = outside.get_capture(len(stream)) - - def test_show_max_translations(self): - """ NAT44ED API test - max translations per thread """ - nat_config = self.vapi.nat_show_config_2() - self.assertEqual(self.max_sessions, - nat_config.max_translations_per_thread) - - def test_lru_cleanup(self): - """ NAT44ED LRU cleanup algorithm """ - - self.nat_add_address(self.nat_addr) - self.nat_add_inside_interface(self.pg0) - self.nat_add_outside_interface(self.pg1) - - self.vapi.nat_set_timeouts( - udp=1, tcp_established=7440, tcp_transitory=30, icmp=1) - - tcp_port_out = self.init_tcp_session(self.pg0, self.pg1, 2000, 80) - pkts = [] - for i in range(0, self.max_sessions - 1): - p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / - IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4, ttl=64) / - UDP(sport=7000+i, dport=80)) - pkts.append(p) - - self.pg0.add_stream(pkts) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg1.get_capture(len(pkts)) - self.sleep(1.5, "wait for timeouts") - - pkts = [] - for i in range(0, self.max_sessions - 1): - p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / - IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4, ttl=64) / - ICMP(id=8000+i, type='echo-request')) - pkts.append(p) - - self.pg0.add_stream(pkts) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg1.get_capture(len(pkts)) - - def test_session_rst_timeout(self): - """ NAT44ED session RST timeouts """ - - self.nat_add_address(self.nat_addr) - self.nat_add_inside_interface(self.pg0) - self.nat_add_outside_interface(self.pg1) - - self.vapi.nat_set_timeouts(udp=300, tcp_established=7440, - tcp_transitory=5, icmp=60) - - self.init_tcp_session(self.pg0, self.pg1, self.tcp_port_in, - self.tcp_external_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=self.tcp_port_in, dport=self.tcp_external_port, - flags="R")) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg1.get_capture(1) - - self.sleep(6) - - 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 + 1, dport=self.tcp_external_port + 1, - flags="S")) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg1.get_capture(1) - - def test_dynamic_out_of_ports(self): - """ NAT44ED dynamic translation test: out of ports """ - - self.nat_add_inside_interface(self.pg0) - self.nat_add_outside_interface(self.pg1) - - # in2out and no NAT addresses added - err_old = self.statistics.get_err_counter( - '/err/nat44-ed-in2out-slowpath/out of ports') - - pkts = self.create_stream_in(self.pg0, self.pg1) - self.pg0.add_stream(pkts) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg1.get_capture(0, timeout=1) - - err_new = self.statistics.get_err_counter( - '/err/nat44-ed-in2out-slowpath/out of ports') - - self.assertEqual(err_new - err_old, len(pkts)) - - # in2out after NAT addresses added - self.nat_add_address(self.nat_addr) - - err_old = self.statistics.get_err_counter( - '/err/nat44-ed-in2out-slowpath/out of ports') - - 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, ignore_port=True) - - err_new = self.statistics.get_err_counter( - '/err/nat44-ed-in2out-slowpath/out of ports') - - self.assertEqual(err_new, err_old) - - def test_unknown_proto(self): - """ NAT44ED translate packet with unknown protocol """ - - self.nat_add_address(self.nat_addr) - self.nat_add_inside_interface(self.pg0) - self.nat_add_outside_interface(self.pg1) - - # 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.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.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.assertEqual(packet.haslayer(GRE), 1) - self.assert_packet_checksums_valid(packet) - except: - self.logger.error(ppp("Unexpected or invalid packet:", packet)) - raise - - def test_hairpinning_unknown_proto(self): - """ NAT44ED translate packet with unknown protocol - hairpinning """ - host = self.pg0.remote_hosts[0] - server = self.pg0.remote_hosts[1] - host_in_port = 1234 - server_out_port = 8765 - server_nat_ip = "10.0.0.11" - - self.nat_add_address(self.nat_addr) - self.nat_add_inside_interface(self.pg0) - self.nat_add_outside_interface(self.pg1) - - # add static mapping for server - self.nat_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() - 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.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, self.nat_addr) - self.assertEqual(packet[IP].dst, server.ip4) - self.assertEqual(packet.haslayer(GRE), 1) - self.assert_packet_checksums_valid(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.pg2.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.assertEqual(packet.haslayer(GRE), 1) - self.assert_packet_checksums_valid(packet) - except: - self.logger.error(ppp("Unexpected or invalid packet:", packet)) - raise - - def test_output_feature_and_service(self): - """ NAT44ED interface output feature and services """ - external_addr = '1.2.3.4' - external_port = 80 - local_port = 8080 - - self.vapi.nat44_forwarding_enable_disable(enable=1) - self.nat_add_address(self.nat_addr) - flags = self.config_flags.NAT_IS_ADDR_ONLY - self.vapi.nat44_add_del_identity_mapping( - ip_address=self.pg1.remote_ip4, sw_if_index=0xFFFFFFFF, - flags=flags, is_add=1) - flags = self.config_flags.NAT_IS_OUT2IN_ONLY - self.nat_add_static_mapping(self.pg0.remote_ip4, external_addr, - local_port, external_port, - proto=IP_PROTOS.tcp, flags=flags) - - self.nat_add_inside_interface(self.pg0) - self.nat_add_outside_interface(self.pg0) - self.vapi.nat44_interface_add_del_output_feature( - sw_if_index=self.pg1.sw_if_index, is_add=1) - - # from client to service - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=external_addr) / - TCP(sport=12345, dport=external_port)) - 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.dst, self.pg0.remote_ip4) - self.assertEqual(tcp.dport, local_port) - self.assert_packet_checksums_valid(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - # from service back to client - 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=local_port, dport=12345)) - 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, external_addr) - self.assertEqual(tcp.sport, external_port) - self.assert_packet_checksums_valid(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - # from local network host to external network - 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, ignore_port=True) - 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, ignore_port=True) - - # from external network back to local network host - 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_and_service3(self): - """ NAT44ED interface output feature and DST NAT """ - external_addr = '1.2.3.4' - external_port = 80 - local_port = 8080 - - self.vapi.nat44_forwarding_enable_disable(enable=1) - self.nat_add_address(self.nat_addr) - flags = self.config_flags.NAT_IS_OUT2IN_ONLY - self.nat_add_static_mapping(self.pg1.remote_ip4, external_addr, - local_port, external_port, - proto=IP_PROTOS.tcp, flags=flags) - - self.nat_add_inside_interface(self.pg0) - self.nat_add_outside_interface(self.pg0) - self.vapi.nat44_interface_add_del_output_feature( - sw_if_index=self.pg1.sw_if_index, is_add=1) - - p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / - IP(src=self.pg0.remote_ip4, dst=external_addr) / - TCP(sport=12345, 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, self.pg0.remote_ip4) - self.assertEqual(tcp.sport, 12345) - self.assertEqual(ip.dst, self.pg1.remote_ip4) - self.assertEqual(tcp.dport, local_port) - self.assert_packet_checksums_valid(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=self.pg0.remote_ip4) / - TCP(sport=local_port, dport=12345)) - 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, external_addr) - self.assertEqual(tcp.sport, external_port) - self.assertEqual(ip.dst, self.pg0.remote_ip4) - self.assertEqual(tcp.dport, 12345) - self.assert_packet_checksums_valid(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - def test_self_twice_nat_lb_negative(self): - """ NAT44ED Self Twice NAT local service load balancing (negative test) - """ - self.twice_nat_common(lb=True, self_twice_nat=True, same_pg=True, - client_id=2) - - def test_self_twice_nat_negative(self): - """ NAT44ED Self Twice NAT (negative test) """ - self.twice_nat_common(self_twice_nat=True) - - def test_static_lb_multi_clients(self): - """ NAT44ED local service load balancing - multiple clients""" - - external_addr = self.nat_addr - external_port = 80 - local_port = 8080 - server1 = self.pg0.remote_hosts[0] - server2 = self.pg0.remote_hosts[1] - server3 = self.pg0.remote_hosts[2] - - locals = [{'addr': server1.ip4, - 'port': local_port, - 'probability': 90, - 'vrf_id': 0}, - {'addr': server2.ip4, - 'port': local_port, - 'probability': 10, - 'vrf_id': 0}] - - flags = self.config_flags.NAT_IS_INSIDE - self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg0.sw_if_index, - flags=flags, is_add=1) - self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg1.sw_if_index, - is_add=1) - - self.nat_add_address(self.nat_addr) - self.vapi.nat44_add_del_lb_static_mapping(is_add=1, - external_addr=external_addr, - external_port=external_port, - protocol=IP_PROTOS.tcp, - local_num=len(locals), - locals=locals) - - server1_n = 0 - server2_n = 0 - clients = ip4_range(self.pg1.remote_ip4, 10, 50) - pkts = [] - for client in clients: - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=client, dst=self.nat_addr) / - TCP(sport=12345, dport=external_port)) - pkts.append(p) - self.pg1.add_stream(pkts) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg0.get_capture(len(pkts)) - for p in capture: - if p[IP].dst == server1.ip4: - server1_n += 1 - else: - server2_n += 1 - self.assertGreaterEqual(server1_n, server2_n) - - local = { - 'addr': server3.ip4, - 'port': local_port, - 'probability': 20, - 'vrf_id': 0 - } - - # add new back-end - self.vapi.nat44_lb_static_mapping_add_del_local( - is_add=1, - external_addr=external_addr, - external_port=external_port, - local=local, - protocol=IP_PROTOS.tcp) - server1_n = 0 - server2_n = 0 - server3_n = 0 - clients = ip4_range(self.pg1.remote_ip4, 60, 110) - pkts = [] - for client in clients: - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=client, dst=self.nat_addr) / - TCP(sport=12346, dport=external_port)) - pkts.append(p) - self.assertGreater(len(pkts), 0) - self.pg1.add_stream(pkts) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg0.get_capture(len(pkts)) - for p in capture: - if p[IP].dst == server1.ip4: - server1_n += 1 - elif p[IP].dst == server2.ip4: - server2_n += 1 - else: - server3_n += 1 - self.assertGreater(server1_n, 0) - self.assertGreater(server2_n, 0) - self.assertGreater(server3_n, 0) - - local = { - 'addr': server2.ip4, - 'port': local_port, - 'probability': 10, - 'vrf_id': 0 - } - - # remove one back-end - self.vapi.nat44_lb_static_mapping_add_del_local( - is_add=0, - external_addr=external_addr, - external_port=external_port, - local=local, - protocol=IP_PROTOS.tcp) - server1_n = 0 - server2_n = 0 - server3_n = 0 - self.pg1.add_stream(pkts) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg0.get_capture(len(pkts)) - for p in capture: - if p[IP].dst == server1.ip4: - server1_n += 1 - elif p[IP].dst == server2.ip4: - server2_n += 1 - else: - server3_n += 1 - self.assertGreater(server1_n, 0) - self.assertEqual(server2_n, 0) - self.assertGreater(server3_n, 0) - - def test_syslog_sess(self): - """ NAT44ED Test syslog session creation and deletion """ - 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) - - self.nat_add_address(self.nat_addr) - self.nat_add_inside_interface(self.pg0) - self.nat_add_outside_interface(self.pg1) - - 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=self.tcp_external_port)) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg1.get_capture(1) - self.tcp_port_out = capture[0][TCP].sport - capture = self.pg3.get_capture(1) - self.verify_syslog_sess(capture[0][Raw].load) - - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.nat_add_address(self.nat_addr, is_add=0) - capture = self.pg3.get_capture(1) - self.verify_syslog_sess(capture[0][Raw].load, False) - - def test_twice_nat_interface_addr(self): - """ NAT44ED Acquire twice NAT addresses from interface """ - flags = self.config_flags.NAT_IS_TWICE_NAT - self.vapi.nat44_add_del_interface_addr( - sw_if_index=self.pg11.sw_if_index, - flags=flags, is_add=1) - - # no address in NAT pool - adresses = self.vapi.nat44_address_dump() - self.assertEqual(0, len(adresses)) - - # configure interface address and check NAT address pool - self.pg11.config_ip4() - adresses = self.vapi.nat44_address_dump() - self.assertEqual(1, len(adresses)) - self.assertEqual(str(adresses[0].ip_address), - self.pg11.local_ip4) - self.assertEqual(adresses[0].flags, flags) - - # remove interface address and check NAT address pool - self.pg11.unconfig_ip4() - adresses = self.vapi.nat44_address_dump() - self.assertEqual(0, len(adresses)) - - def test_output_feature_stateful_acl(self): - """ NAT44ED output feature works with stateful ACL """ - - self.nat_add_address(self.nat_addr) - self.vapi.nat44_interface_add_del_output_feature( - sw_if_index=self.pg0.sw_if_index, - flags=self.config_flags.NAT_IS_INSIDE, is_add=1) - self.vapi.nat44_interface_add_del_output_feature( - sw_if_index=self.pg1.sw_if_index, - flags=self.config_flags.NAT_IS_OUTSIDE, is_add=1) - - # First ensure that the NAT is working sans ACL - - # send packets out2in, no sessions yet so packets should drop - pkts_out2in = self.create_stream_out(self.pg1) - self.send_and_assert_no_replies(self.pg1, pkts_out2in) - - # send packets into inside intf, ensure received via outside intf - pkts_in2out = self.create_stream_in(self.pg0, self.pg1) - capture = self.send_and_expect(self.pg0, pkts_in2out, self.pg1, - len(pkts_in2out)) - self.verify_capture_out(capture, ignore_port=True) - - # send out2in again, with sessions created it should work now - pkts_out2in = self.create_stream_out(self.pg1) - capture = self.send_and_expect(self.pg1, pkts_out2in, self.pg0, - len(pkts_out2in)) - self.verify_capture_in(capture, self.pg0) - - # Create an ACL blocking everything - out2in_deny_rule = AclRule(is_permit=0) - out2in_acl = VppAcl(self, rules=[out2in_deny_rule]) - out2in_acl.add_vpp_config() - - # create an ACL to permit/reflect everything - in2out_reflect_rule = AclRule(is_permit=2) - in2out_acl = VppAcl(self, rules=[in2out_reflect_rule]) - in2out_acl.add_vpp_config() - - # apply as input acl on interface and confirm it blocks everything - acl_if = VppAclInterface(self, sw_if_index=self.pg1.sw_if_index, - n_input=1, acls=[out2in_acl]) - acl_if.add_vpp_config() - self.send_and_assert_no_replies(self.pg1, pkts_out2in) - - # apply output acl - acl_if.acls = [out2in_acl, in2out_acl] - acl_if.add_vpp_config() - # send in2out to generate ACL state (NAT state was created earlier) - capture = self.send_and_expect(self.pg0, pkts_in2out, self.pg1, - len(pkts_in2out)) - self.verify_capture_out(capture, ignore_port=True) - - # send out2in again. ACL state exists so it should work now. - # TCP packets with the syn flag set also need the ack flag - for p in pkts_out2in: - if p.haslayer(TCP) and p[TCP].flags & 0x02: - p[TCP].flags |= 0x10 - capture = self.send_and_expect(self.pg1, pkts_out2in, self.pg0, - len(pkts_out2in)) - self.verify_capture_in(capture, self.pg0) - self.logger.info(self.vapi.cli("show trace")) - - def test_tcp_close(self): - """ NAT44ED Close TCP session from inside network - output feature """ - old_timeouts = self.vapi.nat_get_timeouts() - new_transitory = 2 - self.vapi.nat_set_timeouts( - udp=old_timeouts.udp, - tcp_established=old_timeouts.tcp_established, - icmp=old_timeouts.icmp, - tcp_transitory=new_transitory) - - self.vapi.nat44_forwarding_enable_disable(enable=1) - self.nat_add_address(self.pg1.local_ip4) - twice_nat_addr = '10.0.1.3' - service_ip = '192.168.16.150' - self.nat_add_address(twice_nat_addr, twice_nat=1) - - flags = self.config_flags.NAT_IS_INSIDE - self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg0.sw_if_index, - is_add=1) - self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg0.sw_if_index, - flags=flags, is_add=1) - self.vapi.nat44_interface_add_del_output_feature( - is_add=1, - sw_if_index=self.pg1.sw_if_index) - - flags = (self.config_flags.NAT_IS_OUT2IN_ONLY | - self.config_flags.NAT_IS_TWICE_NAT) - self.nat_add_static_mapping(self.pg0.remote_ip4, - service_ip, 80, 80, - proto=IP_PROTOS.tcp, - flags=flags) - sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0) - start_sessnum = len(sessions) - - # SYN packet out->in - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=service_ip) / - TCP(sport=33898, dport=80, flags="S")) - self.pg1.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg0.get_capture(1) - p = capture[0] - tcp_port = p[TCP].sport - - # SYN + ACK packet in->out - p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / - IP(src=self.pg0.remote_ip4, dst=twice_nat_addr) / - TCP(sport=80, dport=tcp_port, flags="SA")) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg1.get_capture(1) - - # ACK packet out->in - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=service_ip) / - TCP(sport=33898, dport=80, flags="A")) - self.pg1.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg0.get_capture(1) - - # FIN packet in -> out - p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / - IP(src=self.pg0.remote_ip4, dst=twice_nat_addr) / - TCP(sport=80, dport=tcp_port, flags="FA", seq=100, ack=300)) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg1.get_capture(1) - - # FIN+ACK packet out -> in - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=service_ip) / - TCP(sport=33898, dport=80, flags="FA", seq=300, ack=101)) - self.pg1.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg0.get_capture(1) - - # ACK packet in -> out - p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / - IP(src=self.pg0.remote_ip4, dst=twice_nat_addr) / - TCP(sport=80, dport=tcp_port, flags="A", seq=101, ack=301)) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg1.get_capture(1) - - # session now in transitory timeout - # try SYN packet out->in - should be dropped - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=service_ip) / - TCP(sport=33898, dport=80, flags="S")) - self.pg1.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - - self.sleep(new_transitory, "wait for transitory timeout") - self.pg0.assert_nothing_captured(0) - - # session should still exist - sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0) - self.assertEqual(len(sessions) - start_sessnum, 1) - - # send FIN+ACK packet out -> in - will cause session to be wiped - # but won't create a new session - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=service_ip) / - TCP(sport=33898, dport=80, flags="FA", seq=300, ack=101)) - self.pg1.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - - sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0) - self.assertEqual(len(sessions) - start_sessnum, 0) - self.pg0.assert_nothing_captured(0) - - def test_tcp_session_close_in(self): - """ NAT44ED Close TCP session from inside network """ - - in_port = self.tcp_port_in - out_port = 10505 - ext_port = self.tcp_external_port - - self.nat_add_address(self.nat_addr) - self.nat_add_inside_interface(self.pg0) - self.nat_add_outside_interface(self.pg1) - self.nat_add_static_mapping(self.pg0.remote_ip4, self.nat_addr, - in_port, out_port, proto=IP_PROTOS.tcp, - flags=self.config_flags.NAT_IS_TWICE_NAT) - - sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0) - session_n = len(sessions) - - self.vapi.nat_set_timeouts(udp=300, tcp_established=7440, - tcp_transitory=2, icmp=5) - - self.init_tcp_session(self.pg0, self.pg1, in_port, ext_port) - - # 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=in_port, dport=ext_port, - flags="FA", seq=100, ack=300)) - 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=ext_port, dport=out_port, - flags="A", seq=300, ack=101)) - 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=ext_port, dport=out_port, - flags="FA", seq=300, ack=101)) - 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=in_port, dport=ext_port, - flags="A", seq=101, ack=301)) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg1.get_capture(1) - - sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0) - self.assertEqual(len(sessions) - session_n, 1) - - out2in_drops = self.get_err_counter( - '/err/nat44-ed-out2in/drops due to TCP in transitory timeout') - in2out_drops = self.get_err_counter( - '/err/nat44-ed-in2out/drops due to TCP in transitory timeout') - - # extra FIN packet out -> in - this should be dropped - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - TCP(sport=ext_port, dport=out_port, - flags="FA", seq=300, ack=101)) - - self.pg1.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg0.assert_nothing_captured() - - # extra ACK packet in -> out - this should be dropped - 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=in_port, dport=ext_port, - flags="A", seq=101, ack=301)) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg1.assert_nothing_captured() - - stats = self.get_err_counter( - '/err/nat44-ed-out2in/drops due to TCP in transitory timeout') - self.assertEqual(stats - out2in_drops, 1) - stats = self.get_err_counter( - '/err/nat44-ed-in2out/drops due to TCP in transitory timeout') - self.assertEqual(stats - in2out_drops, 1) - - self.sleep(3) - # extra ACK packet in -> out - this will cause session to be wiped - 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=in_port, dport=ext_port, - flags="A", seq=101, ack=301)) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg1.assert_nothing_captured() - sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0) - self.assertEqual(len(sessions) - session_n, 0) - - def test_tcp_session_close_out(self): - """ NAT44ED Close TCP session from outside network """ - - in_port = self.tcp_port_in - out_port = 10505 - ext_port = self.tcp_external_port - - self.nat_add_address(self.nat_addr) - self.nat_add_inside_interface(self.pg0) - self.nat_add_outside_interface(self.pg1) - self.nat_add_static_mapping(self.pg0.remote_ip4, self.nat_addr, - in_port, out_port, proto=IP_PROTOS.tcp, - flags=self.config_flags.NAT_IS_TWICE_NAT) - - sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0) - session_n = len(sessions) - - self.vapi.nat_set_timeouts(udp=300, tcp_established=7440, - tcp_transitory=2, icmp=5) - - _ = self.init_tcp_session(self.pg0, self.pg1, in_port, ext_port) - - # 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=ext_port, dport=out_port, - flags="FA", seq=100, ack=300)) - self.pg1.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg0.get_capture(1) - - # FIN+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=in_port, dport=ext_port, - flags="FA", seq=300, ack=101)) - - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg1.get_capture(1) - - # 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=ext_port, dport=out_port, - flags="A", seq=101, ack=301)) - self.pg1.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg0.get_capture(1) - - sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0) - self.assertEqual(len(sessions) - session_n, 1) - - out2in_drops = self.get_err_counter( - '/err/nat44-ed-out2in/drops due to TCP in transitory timeout') - in2out_drops = self.get_err_counter( - '/err/nat44-ed-in2out/drops due to TCP in transitory timeout') - - # extra FIN packet out -> in - this should be dropped - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - TCP(sport=ext_port, dport=out_port, - flags="FA", seq=300, ack=101)) - - self.pg1.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg0.assert_nothing_captured() - - # extra ACK packet in -> out - this should be dropped - 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=in_port, dport=ext_port, - flags="A", seq=101, ack=301)) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg1.assert_nothing_captured() - - stats = self.get_err_counter( - '/err/nat44-ed-out2in/drops due to TCP in transitory timeout') - self.assertEqual(stats - out2in_drops, 1) - stats = self.get_err_counter( - '/err/nat44-ed-in2out/drops due to TCP in transitory timeout') - self.assertEqual(stats - in2out_drops, 1) - - self.sleep(3) - # extra ACK packet in -> out - this will cause session to be wiped - 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=in_port, dport=ext_port, - flags="A", seq=101, ack=301)) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg1.assert_nothing_captured() - sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0) - self.assertEqual(len(sessions) - session_n, 0) - - def test_tcp_session_close_simultaneous(self): - """ NAT44ED Close TCP session from inside network """ - - in_port = self.tcp_port_in - ext_port = 10505 - - self.nat_add_address(self.nat_addr) - self.nat_add_inside_interface(self.pg0) - self.nat_add_outside_interface(self.pg1) - self.nat_add_static_mapping(self.pg0.remote_ip4, self.nat_addr, - in_port, ext_port, proto=IP_PROTOS.tcp, - flags=self.config_flags.NAT_IS_TWICE_NAT) - - sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0) - session_n = len(sessions) - - self.vapi.nat_set_timeouts(udp=300, tcp_established=7440, - tcp_transitory=2, icmp=5) - - out_port = self.init_tcp_session(self.pg0, self.pg1, in_port, ext_port) - - # 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=in_port, dport=ext_port, - flags="FA", seq=100, ack=300)) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg1.get_capture(1) - - # 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=ext_port, dport=out_port, - flags="FA", seq=300, ack=100)) - self.pg1.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg0.get_capture(1) - - # 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=in_port, dport=ext_port, - flags="A", seq=101, ack=301)) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg1.get_capture(1) - - # 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=ext_port, dport=out_port, - flags="A", seq=301, ack=101)) - self.pg1.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg0.get_capture(1) - - sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0) - self.assertEqual(len(sessions) - session_n, 1) - - out2in_drops = self.get_err_counter( - '/err/nat44-ed-out2in/drops due to TCP in transitory timeout') - in2out_drops = self.get_err_counter( - '/err/nat44-ed-in2out/drops due to TCP in transitory timeout') - - # extra FIN packet out -> in - this should be dropped - p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / - IP(src=self.pg1.remote_ip4, dst=self.nat_addr) / - TCP(sport=ext_port, dport=out_port, - flags="FA", seq=300, ack=101)) - - self.pg1.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg0.assert_nothing_captured() - - # extra ACK packet in -> out - this should be dropped - 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=in_port, dport=ext_port, - flags="A", seq=101, ack=301)) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg1.assert_nothing_captured() - - stats = self.get_err_counter( - '/err/nat44-ed-out2in/drops due to TCP in transitory timeout') - self.assertEqual(stats - out2in_drops, 1) - stats = self.get_err_counter( - '/err/nat44-ed-in2out/drops due to TCP in transitory timeout') - self.assertEqual(stats - in2out_drops, 1) - - self.sleep(3) - # extra ACK packet in -> out - this will cause session to be wiped - 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=in_port, dport=ext_port, - flags="A", seq=101, ack=301)) - self.pg0.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - self.pg1.assert_nothing_captured() - sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0) - self.assertEqual(len(sessions) - session_n, 0) - - def test_dynamic_vrf(self): - """ NAT44ED dynamic translation test: different VRF""" - - vrf_id_in = 33 - vrf_id_out = 34 - - self.nat_add_address(self.nat_addr, vrf_id=vrf_id_in) - - try: - self.configure_ip4_interface(self.pg7, table_id=vrf_id_in) - self.configure_ip4_interface(self.pg8, table_id=vrf_id_out) - - self.nat_add_inside_interface(self.pg7) - self.nat_add_outside_interface(self.pg8) - - # just basic stuff nothing special - 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, ignore_port=True) - - 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) - - finally: - self.pg7.unconfig() - self.pg8.unconfig() - - self.vapi.ip_table_add_del(is_add=0, - table={'table_id': vrf_id_in}) - self.vapi.ip_table_add_del(is_add=0, - table={'table_id': vrf_id_out}) - - def test_dynamic_output_feature_vrf(self): - """ NAT44ED dynamic translation test: output-feature, VRF""" - - # other then default (0) - new_vrf_id = 22 - - self.nat_add_address(self.nat_addr) - flags = self.config_flags.NAT_IS_INSIDE - self.vapi.nat44_interface_add_del_output_feature( - sw_if_index=self.pg7.sw_if_index, - flags=flags, is_add=1) - self.vapi.nat44_interface_add_del_output_feature( - sw_if_index=self.pg8.sw_if_index, - is_add=1) - - try: - self.configure_ip4_interface(self.pg7, table_id=new_vrf_id) - self.configure_ip4_interface(self.pg8, table_id=new_vrf_id) - - # in2out - tcpn = self.statistics['/nat44-ed/in2out/slowpath/tcp'] - udpn = self.statistics['/nat44-ed/in2out/slowpath/udp'] - icmpn = self.statistics['/nat44-ed/in2out/slowpath/icmp'] - drops = self.statistics['/nat44-ed/in2out/slowpath/drops'] - - 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, ignore_port=True) - - if_idx = self.pg7.sw_if_index - cnt = self.statistics['/nat44-ed/in2out/slowpath/tcp'] - self.assertEqual(cnt[:, if_idx].sum() - tcpn[:, if_idx].sum(), 2) - cnt = self.statistics['/nat44-ed/in2out/slowpath/udp'] - self.assertEqual(cnt[:, if_idx].sum() - udpn[:, if_idx].sum(), 1) - cnt = self.statistics['/nat44-ed/in2out/slowpath/icmp'] - self.assertEqual(cnt[:, if_idx].sum() - icmpn[:, if_idx].sum(), 1) - cnt = self.statistics['/nat44-ed/in2out/slowpath/drops'] - self.assertEqual(cnt[:, if_idx].sum() - drops[:, if_idx].sum(), 0) - - # out2in - tcpn = self.statistics['/nat44-ed/out2in/fastpath/tcp'] - udpn = self.statistics['/nat44-ed/out2in/fastpath/udp'] - icmpn = self.statistics['/nat44-ed/out2in/fastpath/icmp'] - drops = self.statistics['/nat44-ed/out2in/fastpath/drops'] - - 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) - - if_idx = self.pg8.sw_if_index - cnt = self.statistics['/nat44-ed/out2in/fastpath/tcp'] - self.assertEqual(cnt[:, if_idx].sum() - tcpn[:, if_idx].sum(), 2) - cnt = self.statistics['/nat44-ed/out2in/fastpath/udp'] - self.assertEqual(cnt[:, if_idx].sum() - udpn[:, if_idx].sum(), 1) - cnt = self.statistics['/nat44-ed/out2in/fastpath/icmp'] - self.assertEqual(cnt[:, if_idx].sum() - icmpn[:, if_idx].sum(), 1) - cnt = self.statistics['/nat44-ed/out2in/fastpath/drops'] - self.assertEqual(cnt[:, if_idx].sum() - drops[:, if_idx].sum(), 0) - - sessions = self.statistics['/nat44-ed/total-sessions'] - self.assertEqual(sessions[:, 0].sum(), 3) - - finally: - self.pg7.unconfig() - self.pg8.unconfig() - - self.vapi.ip_table_add_del(is_add=0, - table={'table_id': new_vrf_id}) - - def test_next_src_nat(self): - """ NAT44ED On way back forward packet to nat44-in2out node. """ - - twice_nat_addr = '10.0.1.3' - external_port = 80 - local_port = 8080 - post_twice_nat_port = 0 - - self.vapi.nat44_forwarding_enable_disable(enable=1) - self.nat_add_address(twice_nat_addr, twice_nat=1) - flags = (self.config_flags.NAT_IS_OUT2IN_ONLY | - self.config_flags.NAT_IS_SELF_TWICE_NAT) - self.nat_add_static_mapping(self.pg6.remote_ip4, self.pg1.remote_ip4, - local_port, external_port, - proto=IP_PROTOS.tcp, vrf_id=1, - flags=flags) - self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg6.sw_if_index, - is_add=1) - - p = (Ether(src=self.pg6.remote_mac, dst=self.pg6.local_mac) / - IP(src=self.pg6.remote_ip4, dst=self.pg1.remote_ip4) / - TCP(sport=12345, dport=external_port)) - self.pg6.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg6.get_capture(1) - p = capture[0] - try: - ip = p[IP] - tcp = p[TCP] - self.assertEqual(ip.src, twice_nat_addr) - self.assertNotEqual(tcp.sport, 12345) - post_twice_nat_port = tcp.sport - self.assertEqual(ip.dst, self.pg6.remote_ip4) - self.assertEqual(tcp.dport, local_port) - self.assert_packet_checksums_valid(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - p = (Ether(src=self.pg6.remote_mac, dst=self.pg6.local_mac) / - IP(src=self.pg6.remote_ip4, dst=twice_nat_addr) / - TCP(sport=local_port, dport=post_twice_nat_port)) - self.pg6.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg6.get_capture(1) - p = capture[0] - try: - ip = p[IP] - tcp = p[TCP] - self.assertEqual(ip.src, self.pg1.remote_ip4) - self.assertEqual(tcp.sport, external_port) - self.assertEqual(ip.dst, self.pg6.remote_ip4) - self.assertEqual(tcp.dport, 12345) - self.assert_packet_checksums_valid(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - def test_one_armed_nat44_static(self): - """ NAT44ED One armed NAT and 1:1 NAPT asymmetrical rule """ - - remote_host = self.pg4.remote_hosts[0] - local_host = self.pg4.remote_hosts[1] - external_port = 80 - local_port = 8080 - eh_port_in = 0 - - self.vapi.nat44_forwarding_enable_disable(enable=1) - self.nat_add_address(self.nat_addr, twice_nat=1) - flags = (self.config_flags.NAT_IS_OUT2IN_ONLY | - self.config_flags.NAT_IS_TWICE_NAT) - self.nat_add_static_mapping(local_host.ip4, self.nat_addr, - local_port, external_port, - proto=IP_PROTOS.tcp, flags=flags) - flags = self.config_flags.NAT_IS_INSIDE - self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg4.sw_if_index, - is_add=1) - self.vapi.nat44_interface_add_del_feature( - sw_if_index=self.pg4.sw_if_index, - flags=flags, is_add=1) - - # from client to service - p = (Ether(src=self.pg4.remote_mac, dst=self.pg4.local_mac) / - IP(src=remote_host.ip4, dst=self.nat_addr) / - TCP(sport=12345, dport=external_port)) - self.pg4.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg4.get_capture(1) - p = capture[0] - try: - ip = p[IP] - tcp = p[TCP] - self.assertEqual(ip.dst, local_host.ip4) - self.assertEqual(ip.src, self.nat_addr) - self.assertEqual(tcp.dport, local_port) - self.assertNotEqual(tcp.sport, 12345) - eh_port_in = tcp.sport - self.assert_packet_checksums_valid(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - # from service back to client - p = (Ether(src=self.pg4.remote_mac, dst=self.pg4.local_mac) / - IP(src=local_host.ip4, dst=self.nat_addr) / - TCP(sport=local_port, dport=eh_port_in)) - self.pg4.add_stream(p) - self.pg_enable_capture(self.pg_interfaces) - self.pg_start() - capture = self.pg4.get_capture(1) - p = capture[0] - try: - ip = p[IP] - tcp = p[TCP] - self.assertEqual(ip.src, self.nat_addr) - self.assertEqual(ip.dst, remote_host.ip4) - self.assertEqual(tcp.sport, external_port) - self.assertEqual(tcp.dport, 12345) - self.assert_packet_checksums_valid(p) - except: - self.logger.error(ppp("Unexpected or invalid packet:", p)) - raise - - -if __name__ == '__main__': - unittest.main(testRunner=VppTestRunner) |