X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftest_snat.py;h=78919a026edfefb8b539ca9aeb108040f0581966;hb=6b7fcda46689bf540538f839c870a9981752fc2c;hp=967d4b37c71a7edc2d25db92ed4f001a6008d2b1;hpb=09d96f4a611fa989bfbbfb7e683d668dbe73ac1a;p=vpp.git diff --git a/test/test_snat.py b/test/test_snat.py index 967d4b37c71..78919a026ed 100644 --- a/test/test_snat.py +++ b/test/test_snat.py @@ -6,6 +6,7 @@ import struct from framework import VppTestCase, VppTestRunner from scapy.layers.inet import IP, TCP, UDP, ICMP +from scapy.layers.inet import IPerror, TCPerror, UDPerror, ICMPerror from scapy.layers.l2 import Ether, ARP from scapy.data import IP_PROTOS from util import ppp @@ -64,59 +65,61 @@ class TestSNAT(VppTestCase): super(TestSNAT, cls).tearDownClass() raise - def create_stream_in(self, in_if, out_if): + def create_stream_in(self, in_if, out_if, ttl=64): """ Create packet stream for inside network :param in_if: Inside interface :param out_if: Outside interface + :param ttl: TTL of generated packets """ pkts = [] # TCP p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) / - IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) / + IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) / TCP(sport=self.tcp_port_in)) pkts.append(p) # UDP p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) / - IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) / + IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) / UDP(sport=self.udp_port_in)) pkts.append(p) # ICMP p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) / - IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) / + IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) / ICMP(id=self.icmp_id_in, type='echo-request')) pkts.append(p) return pkts - def create_stream_out(self, out_if, dst_ip=None): + def create_stream_out(self, out_if, dst_ip=None, ttl=64): """ Create packet stream for outside network :param out_if: Outside interface :param dst_ip: Destination IP address (Default use global SNAT address) + :param ttl: TTL of generated packets """ if dst_ip is None: dst_ip = self.snat_addr pkts = [] # TCP p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) / - IP(src=out_if.remote_ip4, dst=dst_ip) / + IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) / TCP(dport=self.tcp_port_out)) pkts.append(p) # UDP p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) / - IP(src=out_if.remote_ip4, dst=dst_ip) / + IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) / UDP(dport=self.udp_port_out)) pkts.append(p) # ICMP p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) / - IP(src=out_if.remote_ip4, dst=dst_ip) / + IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) / ICMP(id=self.icmp_id_out, type='echo-reply')) pkts.append(p) @@ -209,6 +212,75 @@ class TestSNAT(VppTestCase): "(inside network):", packet)) raise + def verify_capture_out_with_icmp_errors(self, capture, src_ip=None, + packet_num=3, icmp_type=11): + """ + Verify captured packets with ICMP errors on outside network + + :param capture: Captured packets + :param src_ip: Translated IP address or IP address of VPP + (Default use global SNAT address) + :param packet_num: Expected number of packets (Default 3) + :param icmp_type: Type of error ICMP packet + we are expecting (Default 11) + """ + if src_ip is None: + src_ip = self.snat_addr + self.assertEqual(packet_num, len(capture)) + for packet in capture: + try: + self.assertEqual(packet[IP].src, src_ip) + self.assertTrue(packet.haslayer(ICMP)) + icmp = packet[ICMP] + self.assertEqual(icmp.type, icmp_type) + self.assertTrue(icmp.haslayer(IPerror)) + inner_ip = icmp[IPerror] + if inner_ip.haslayer(TCPerror): + self.assertEqual(inner_ip[TCPerror].dport, + self.tcp_port_out) + elif inner_ip.haslayer(UDPerror): + self.assertEqual(inner_ip[UDPerror].dport, + self.udp_port_out) + else: + self.assertEqual(inner_ip[ICMPerror].id, self.icmp_id_out) + except: + self.logger.error(ppp("Unexpected or invalid packet " + "(outside network):", packet)) + raise + + def verify_capture_in_with_icmp_errors(self, capture, in_if, packet_num=3, + icmp_type=11): + """ + Verify captured packets with ICMP errors on inside network + + :param capture: Captured packets + :param in_if: Inside interface + :param packet_num: Expected number of packets (Default 3) + :param icmp_type: Type of error ICMP packet + we are expecting (Default 11) + """ + self.assertEqual(packet_num, len(capture)) + for packet in capture: + try: + self.assertEqual(packet[IP].dst, in_if.remote_ip4) + self.assertTrue(packet.haslayer(ICMP)) + icmp = packet[ICMP] + self.assertEqual(icmp.type, icmp_type) + self.assertTrue(icmp.haslayer(IPerror)) + inner_ip = icmp[IPerror] + if inner_ip.haslayer(TCPerror): + self.assertEqual(inner_ip[TCPerror].sport, + self.tcp_port_in) + elif inner_ip.haslayer(UDPerror): + self.assertEqual(inner_ip[UDPerror].sport, + self.udp_port_in) + else: + self.assertEqual(inner_ip[ICMPerror].id, self.icmp_id_in) + except: + self.logger.error(ppp("Unexpected or invalid packet " + "(inside network):", packet)) + raise + def verify_ipfix_nat44_ses(self, data): """ Verify IPFIX NAT44 session create/delete event @@ -333,7 +405,7 @@ class TestSNAT(VppTestCase): proto, is_add) - def snat_add_address(self, ip, is_add=1): + def snat_add_address(self, ip, is_add=1, vrf_id=0xFFFFFFFF): """ Add/delete S-NAT address @@ -341,7 +413,8 @@ class TestSNAT(VppTestCase): :param is_add: 1 if add, 0 if delete (Default add) """ snat_addr = socket.inet_pton(socket.AF_INET, ip) - self.vapi.snat_add_address_range(snat_addr, snat_addr, is_add) + self.vapi.snat_add_address_range(snat_addr, snat_addr, is_add, + vrf_id=vrf_id) def test_dynamic(self): """ SNAT dynamic translation test """ @@ -367,6 +440,141 @@ class TestSNAT(VppTestCase): capture = self.pg0.get_capture(len(pkts)) self.verify_capture_in(capture, self.pg0) + def test_dynamic_icmp_errors_in2out_ttl_1(self): + """ SNAT handling of client packets with TTL=1 """ + + self.snat_add_address(self.snat_addr) + self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index) + self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index, + is_inside=0) + + # Client side - generate traffic + pkts = self.create_stream_in(self.pg0, self.pg1, ttl=1) + self.pg0.add_stream(pkts) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + + # Client side - verify ICMP type 11 packets + capture = self.pg0.get_capture(len(pkts)) + self.verify_capture_in_with_icmp_errors(capture, self.pg0) + + def test_dynamic_icmp_errors_out2in_ttl_1(self): + """ SNAT handling of server packets with TTL=1 """ + + self.snat_add_address(self.snat_addr) + self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index) + self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index, + is_inside=0) + + # Client side - create sessions + pkts = self.create_stream_in(self.pg0, self.pg1) + self.pg0.add_stream(pkts) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + + # Server side - generate traffic + capture = self.pg1.get_capture(len(pkts)) + self.verify_capture_out(capture) + pkts = self.create_stream_out(self.pg1, ttl=1) + self.pg1.add_stream(pkts) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + + # Server side - verify ICMP type 11 packets + capture = self.pg1.get_capture(len(pkts)) + self.verify_capture_out_with_icmp_errors(capture, + src_ip=self.pg1.local_ip4) + + def test_dynamic_icmp_errors_in2out_ttl_2(self): + """ SNAT handling of error responses to client packets with TTL=2 """ + + self.snat_add_address(self.snat_addr) + self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index) + self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index, + is_inside=0) + + # Client side - generate traffic + pkts = self.create_stream_in(self.pg0, self.pg1, ttl=2) + self.pg0.add_stream(pkts) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + + # Server side - simulate ICMP type 11 response + capture = self.pg1.get_capture(len(pkts)) + pkts = [Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / + IP(src=self.pg1.remote_ip4, dst=self.snat_addr) / + ICMP(type=11) / packet[IP] for packet in capture] + self.pg1.add_stream(pkts) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + + # Client side - verify ICMP type 11 packets + capture = self.pg0.get_capture(len(pkts)) + self.verify_capture_in_with_icmp_errors(capture, self.pg0) + + def test_dynamic_icmp_errors_out2in_ttl_2(self): + """ SNAT handling of error responses to server packets with TTL=2 """ + + self.snat_add_address(self.snat_addr) + self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index) + self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index, + is_inside=0) + + # Client side - create sessions + pkts = self.create_stream_in(self.pg0, self.pg1) + self.pg0.add_stream(pkts) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + + # Server side - generate traffic + capture = self.pg1.get_capture(len(pkts)) + self.verify_capture_out(capture) + pkts = self.create_stream_out(self.pg1, ttl=2) + self.pg1.add_stream(pkts) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + + # Client side - simulate ICMP type 11 response + capture = self.pg0.get_capture(len(pkts)) + pkts = [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) / packet[IP] for packet in capture] + self.pg0.add_stream(pkts) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + + # Server side - verify ICMP type 11 packets + capture = self.pg1.get_capture(len(pkts)) + self.verify_capture_out_with_icmp_errors(capture) + + def test_ping_out_interface_from_outside(self): + """ Ping SNAT out interface from outside """ + + self.snat_add_address(self.snat_addr) + self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index) + self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index, + is_inside=0) + + p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / + IP(src=self.pg1.remote_ip4, dst=self.pg1.local_ip4) / + ICMP(id=self.icmp_id_out, type='echo-request')) + pkts = [p] + self.pg1.add_stream(pkts) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + capture = self.pg1.get_capture(len(pkts)) + self.assertEqual(1, len(capture)) + packet = capture[0] + try: + self.assertEqual(packet[IP].src, self.pg1.local_ip4) + self.assertEqual(packet[IP].dst, self.pg1.remote_ip4) + self.assertEqual(packet[ICMP].id, self.icmp_id_in) + self.assertEqual(packet[ICMP].type, 0) # echo reply + except: + self.logger.error(ppp("Unexpected or invalid packet " + "(outside network):", packet)) + raise + def test_static_in(self): """ SNAT 1:1 NAT initialized from inside network """ @@ -663,6 +871,27 @@ class TestSNAT(VppTestCase): capture = self.pg5.get_capture(len(pkts)) self.verify_capture_in(capture, self.pg5) + # pg5 session dump + addresses = self.vapi.snat_address_dump() + self.assertEqual(len(addresses), 1) + sessions = self.vapi.snat_user_session_dump(self.pg5.remote_ip4n, 10) + self.assertEqual(len(sessions), 3) + for session in sessions: + self.assertFalse(session.is_static) + self.assertEqual(session.inside_ip_address[0:4], + self.pg5.remote_ip4n) + self.assertEqual(session.outside_ip_address, + addresses[0].ip_address) + self.assertEqual(sessions[0].protocol, IP_PROTOS.tcp) + self.assertEqual(sessions[1].protocol, IP_PROTOS.udp) + self.assertEqual(sessions[2].protocol, IP_PROTOS.icmp) + self.assertEqual(sessions[0].inside_port, self.tcp_port_in) + self.assertEqual(sessions[1].inside_port, self.udp_port_in) + self.assertEqual(sessions[2].inside_port, self.icmp_id_in) + self.assertEqual(sessions[0].outside_port, self.tcp_port_out) + self.assertEqual(sessions[1].outside_port, self.udp_port_out) + self.assertEqual(sessions[2].outside_port, self.icmp_id_out) + # in2out 3rd interface pkts = self.create_stream_in(self.pg6, self.pg3) self.pg6.add_stream(pkts) @@ -679,6 +908,44 @@ class TestSNAT(VppTestCase): capture = self.pg6.get_capture(len(pkts)) self.verify_capture_in(capture, self.pg6) + # general user and session dump verifications + users = self.vapi.snat_user_dump() + self.assertTrue(len(users) >= 3) + addresses = self.vapi.snat_address_dump() + self.assertEqual(len(addresses), 1) + for user in users: + sessions = self.vapi.snat_user_session_dump(user.ip_address, + user.vrf_id) + for session in sessions: + self.assertEqual(user.ip_address, session.inside_ip_address) + self.assertTrue(session.total_bytes > session.total_pkts > 0) + self.assertTrue(session.protocol in + [IP_PROTOS.tcp, IP_PROTOS.udp, + IP_PROTOS.icmp]) + + # pg4 session dump + sessions = self.vapi.snat_user_session_dump(self.pg4.remote_ip4n, 10) + self.assertTrue(len(sessions) >= 4) + for session in sessions: + self.assertFalse(session.is_static) + self.assertEqual(session.inside_ip_address[0:4], + self.pg4.remote_ip4n) + self.assertEqual(session.outside_ip_address, + addresses[0].ip_address) + + # pg6 session dump + sessions = self.vapi.snat_user_session_dump(self.pg6.remote_ip4n, 20) + self.assertTrue(len(sessions) >= 3) + for session in sessions: + self.assertTrue(session.is_static) + self.assertEqual(session.inside_ip_address[0:4], + self.pg6.remote_ip4n) + self.assertEqual(map(ord, session.outside_ip_address[0:4]), + map(int, static_nat_ip.split('.'))) + self.assertTrue(session.inside_port in + [self.tcp_port_in, self.udp_port_in, + self.icmp_id_in]) + def test_hairpinning(self): """ SNAT hairpinning """ @@ -790,9 +1057,11 @@ class TestSNAT(VppTestCase): self.snat_add_static_mapping('1.2.3.4', external_sw_if_index=self.pg7.sw_if_index) - # no static mappings + # static mappings with external interface static_mappings = self.vapi.snat_static_mapping_dump() - self.assertEqual(0, len(static_mappings)) + self.assertEqual(1, len(static_mappings)) + self.assertEqual(self.pg7.sw_if_index, + static_mappings[0].external_sw_if_index) # configure interface address and check static mappings self.pg7.config_ip4() @@ -800,6 +1069,7 @@ class TestSNAT(VppTestCase): self.assertEqual(1, len(static_mappings)) self.assertEqual(static_mappings[0].external_ip_address[0:4], self.pg7.local_ip4n) + self.assertEqual(0xFFFFFFFF, static_mappings[0].external_sw_if_index) # remove interface address and check static mappings self.pg7.unconfig_ip4() @@ -932,6 +1202,73 @@ class TestSNAT(VppTestCase): self.pg_start() capture = self.pg1.get_capture(0) + def test_vrf_mode(self): + """ S-NAT tenant VRF aware address pool mode """ + + vrf_id1 = 1 + vrf_id2 = 2 + nat_ip1 = "10.0.0.10" + nat_ip2 = "10.0.0.11" + + self.pg0.unconfig_ip4() + self.pg1.unconfig_ip4() + self.pg0.set_table_ip4(vrf_id1) + self.pg1.set_table_ip4(vrf_id2) + self.pg0.config_ip4() + self.pg1.config_ip4() + + self.snat_add_address(nat_ip1, vrf_id=vrf_id1) + self.snat_add_address(nat_ip2, vrf_id=vrf_id2) + self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index) + self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index) + self.vapi.snat_interface_add_del_feature(self.pg2.sw_if_index, + is_inside=0) + + # first VRF + pkts = self.create_stream_in(self.pg0, self.pg2) + self.pg0.add_stream(pkts) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + capture = self.pg2.get_capture(len(pkts)) + self.verify_capture_out(capture, nat_ip1) + + # second VRF + pkts = self.create_stream_in(self.pg1, self.pg2) + self.pg1.add_stream(pkts) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + capture = self.pg2.get_capture(len(pkts)) + self.verify_capture_out(capture, nat_ip2) + + def test_vrf_feature_independent(self): + """ S-NAT tenant VRF independent address pool mode """ + + nat_ip1 = "10.0.0.10" + nat_ip2 = "10.0.0.11" + + self.snat_add_address(nat_ip1) + self.snat_add_address(nat_ip2) + self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index) + self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index) + self.vapi.snat_interface_add_del_feature(self.pg2.sw_if_index, + is_inside=0) + + # first VRF + pkts = self.create_stream_in(self.pg0, self.pg2) + self.pg0.add_stream(pkts) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + capture = self.pg2.get_capture(len(pkts)) + self.verify_capture_out(capture, nat_ip1) + + # second VRF + pkts = self.create_stream_in(self.pg1, self.pg2) + self.pg1.add_stream(pkts) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + capture = self.pg2.get_capture(len(pkts)) + self.verify_capture_out(capture, nat_ip1) + def tearDown(self): super(TestSNAT, self).tearDown() if not self.vpp_dead: @@ -939,5 +1276,57 @@ class TestSNAT(VppTestCase): self.clear_snat() +class TestDeterministicNAT(VppTestCase): + """ Deterministic NAT Test Cases """ + + @classmethod + def setUpConstants(cls): + super(TestDeterministicNAT, cls).setUpConstants() + cls.vpp_cmdline.extend(["snat", "{", "deterministic", "}"]) + + @classmethod + def setUpClass(cls): + super(TestDeterministicNAT, cls).setUpClass() + + try: + cls.create_pg_interfaces(range(2)) + cls.interfaces = list(cls.pg_interfaces) + + for i in cls.interfaces: + i.admin_up() + i.config_ip4() + i.resolve_arp() + + except Exception: + super(TestDeterministicNAT, cls).tearDownClass() + raise + + def test_deterministic_mode(self): + """ S-NAT run deterministic mode """ + in_addr = '172.16.255.0' + out_addr = '172.17.255.50' + in_addr_t = '172.16.255.20' + in_addr_n = socket.inet_aton(in_addr) + out_addr_n = socket.inet_aton(out_addr) + in_addr_t_n = socket.inet_aton(in_addr_t) + in_plen = 24 + out_plen = 32 + + snat_config = self.vapi.snat_show_config() + self.assertEqual(1, snat_config.deterministic) + + self.vapi.snat_add_det_map(in_addr_n, in_plen, out_addr_n, out_plen) + + rep1 = self.vapi.snat_det_forward(in_addr_t_n) + self.assertEqual(rep1.out_addr[:4], out_addr_n) + rep2 = self.vapi.snat_det_reverse(out_addr_n, rep1.out_port_hi) + self.assertEqual(rep2.in_addr[:4], in_addr_t_n) + + def tearDown(self): + super(TestDeterministicNAT, self).tearDown() + if not self.vpp_dead: + self.logger.info(self.vapi.cli("show snat detail")) + + if __name__ == '__main__': unittest.main(testRunner=VppTestRunner)