diff options
Diffstat (limited to 'test')
| -rwxr-xr-x | test/framework.py | 2 | ||||
| -rw-r--r-- | test/test_nat44_ed.py | 15 | ||||
| -rw-r--r-- | test/test_nat44_ed_output.py | 229 | ||||
| -rw-r--r-- | test/test_nat44_ei.py | 2 |
4 files changed, 238 insertions, 10 deletions
diff --git a/test/framework.py b/test/framework.py index aa533f7b641..486553befa1 100755 --- a/test/framework.py +++ b/test/framework.py @@ -1304,6 +1304,8 @@ class VppTestCase(CPUInterface, unittest.TestCase): n_rx = len(pkts) self.pg_send(intf, pkts, worker=worker, trace=trace) rx = output.get_capture(n_rx) + if trace: + self.logger.debug(self.vapi.cli("show trace")) return rx def send_and_expect_only(self, intf, pkts, output, timeout=None): diff --git a/test/test_nat44_ed.py b/test/test_nat44_ed.py index ec8d7c8d159..77f2555024b 100644 --- a/test/test_nat44_ed.py +++ b/test/test_nat44_ed.py @@ -19,7 +19,8 @@ from vpp_ip_route import VppIpRoute, VppRoutePath from vpp_papi import VppEnum -class NAT44EDTestCase(VppTestCase): +class TestNAT44ED(VppTestCase): + """ NAT44ED Test Case """ nat_addr = '10.0.0.3' @@ -37,11 +38,11 @@ class NAT44EDTestCase(VppTestCase): max_sessions = 100 def setUp(self): - super(NAT44EDTestCase, self).setUp() + super().setUp() self.plugin_enable() def tearDown(self): - super(NAT44EDTestCase, self).tearDown() + super().tearDown() if not self.vpp_dead: self.plugin_disable() @@ -146,7 +147,7 @@ class NAT44EDTestCase(VppTestCase): @classmethod def setUpClass(cls): - super(NAT44EDTestCase, cls).setUpClass() + super().setUpClass() cls.create_pg_interfaces(range(12)) cls.interfaces = list(cls.pg_interfaces[:4]) @@ -910,10 +911,6 @@ class NAT44EDTestCase(VppTestCase): self.assertEqual(sd_params.get('XDPORT'), "%d" % self.tcp_external_port) - -class TestNAT44ED(NAT44EDTestCase): - """ NAT44ED Test Case """ - def test_icmp_error(self): """ NAT44ED test ICMP error message with inner header""" @@ -2258,7 +2255,7 @@ class TestNAT44EDMW(TestNAT44ED): self.pg_enable_capture(self.pg_interfaces) self.pg_start() - capture = self.pg1.get_capture(pkt_count * 3) + capture = self.pg1.get_capture(pkt_count * 3, timeout=5) if_idx = self.pg0.sw_if_index tc2 = self.statistics['/nat44-ed/in2out/slowpath/tcp'] diff --git a/test/test_nat44_ed_output.py b/test/test_nat44_ed_output.py new file mode 100644 index 00000000000..ea5c14e7064 --- /dev/null +++ b/test/test_nat44_ed_output.py @@ -0,0 +1,229 @@ +#!/usr/bin/env python3 +"""NAT44 ED output-feature tests""" + +import random +import unittest +from scapy.layers.inet import ICMP, Ether, IP, TCP +from scapy.packet import Raw +from scapy.data import IP_PROTOS +from framework import VppTestCase, VppTestRunner +from vpp_papi import VppEnum + + +def get_nat44_ed_in2out_worker_index(ip, vpp_worker_count): + if 0 == vpp_worker_count: + return 0 + numeric = socket.inet_aton(ip) + numeric = struct.unpack("!L", numeric)[0] + numeric = socket.htonl(numeric) + h = numeric + (numeric >> 8) + (numeric >> 16) + (numeric >> 24) + return 1 + h % vpp_worker_count + + +class TestNAT44EDOutput(VppTestCase): + """ NAT44 ED output feature Test Case """ + max_sessions = 1024 + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.create_pg_interfaces(range(2)) + cls.interfaces = list(cls.pg_interfaces) + + @classmethod + def tearDownClass(cls): + super().tearDownClass() + + def setUp(self): + super().setUp() + for i in self.interfaces: + i.admin_up() + i.config_ip4() + i.resolve_arp() + self.vapi.nat44_ed_plugin_enable_disable(sessions=self.max_sessions, + enable=1) + + def tearDown(self): + if not self.vpp_dead: + self.logger.debug(self.vapi.cli("show nat44 sessions")) + super().tearDown() + if not self.vpp_dead: + for i in self.pg_interfaces: + i.unconfig_ip4() + i.admin_down() + self.vapi.nat44_ed_plugin_enable_disable(enable=0) + + def test_static_dynamic(self): + """ Create static mapping which matches existing dynamic mapping """ + + 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) + + local_host = self.pg0.remote_ip4 + remote_host = self.pg1.remote_ip4 + nat_intf = self.pg1 + outside_addr = nat_intf.local_ip4 + + self.vapi.nat44_add_del_address_range(first_ip_address=outside_addr, + last_ip_address=outside_addr, + vrf_id=0xffffffff, + is_add=1, + flags=0) + 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=VppEnum.vl_api_nat_config_flags_t.NAT_IS_INSIDE, is_add=1) + self.vapi.nat44_interface_add_del_output_feature( + is_add=1, + sw_if_index=self.pg1.sw_if_index) + + thread_index = get_nat44_ed_in2out_worker_index( + local_host, self.vpp_worker_count) + port_per_thread = int((0xffff-1024) / max(1, self.vpp_worker_count)) + local_sport = 1024 + random.randint(1, port_per_thread) + if self.vpp_worker_count > 0: + local_sport += port_per_thread * (thread_index - 1) + + remote_dport = 10000 + + pg0 = self.pg0 + pg1 = self.pg1 + + # first setup a dynamic TCP session + + # SYN packet in->out + p = (Ether(src=pg0.remote_mac, dst=pg0.local_mac) / + IP(src=local_host, dst=remote_host) / + TCP(sport=local_sport, dport=remote_dport, flags="S")) + p = self.send_and_expect(pg0, [p], pg1)[0] + + self.assertEqual(p[IP].src, outside_addr) + self.assertEqual(p[TCP].sport, local_sport) + outside_port = p[TCP].sport + + # SYN+ACK packet out->in + p = (Ether(src=pg1.remote_mac, dst=pg1.local_mac) / + IP(src=remote_host, dst=outside_addr) / + TCP(sport=remote_dport, dport=outside_port, flags="SA")) + self.send_and_expect(pg1, [p], pg0) + + # ACK packet in->out + p = (Ether(src=pg0.remote_mac, dst=pg0.local_mac) / + IP(src=local_host, dst=remote_host) / + TCP(sport=local_sport, dport=remote_dport, flags="A")) + self.send_and_expect(pg0, [p], pg1) + + # now we have a session up, create a conflicting static mapping + self.vapi.nat44_add_del_static_mapping( + is_add=1, + local_ip_address=local_host, + external_ip_address=outside_addr, + external_sw_if_index=0xffffffff, + local_port=local_sport, + external_port=outside_port, + protocol=IP_PROTOS.tcp, + flags=VppEnum.vl_api_nat_config_flags_t.NAT_IS_OUT2IN_ONLY) + + sessions = self.vapi.nat44_user_session_dump(local_host, 0) + self.assertEqual(1, len(sessions)) + + # now send some more data over existing session - it should pass + + # in->out + p = (Ether(src=pg0.remote_mac, dst=pg0.local_mac) / + IP(src=local_host, dst=remote_host) / + TCP(sport=local_sport, dport=remote_dport) / + Raw("zippity zap")) + self.send_and_expect(pg0, [p], pg1) + + # out->in + p = (Ether(src=pg1.remote_mac, dst=pg1.local_mac) / + IP(src=remote_host, dst=outside_addr) / + TCP(sport=remote_dport, dport=outside_port) / + Raw("flippity flop")) + self.send_and_expect(pg1, [p], pg0) + + # now close the session + + # FIN packet in -> out + p = (Ether(src=pg0.remote_mac, dst=pg0.local_mac) / + IP(src=local_host, dst=remote_host) / + TCP(sport=local_sport, dport=remote_dport, flags="FA", seq=100, + ack=300)) + self.send_and_expect(pg0, [p], pg1) + + # FIN+ACK packet out -> in + p = (Ether(src=pg1.remote_mac, dst=pg1.local_mac) / + IP(src=remote_host, dst=outside_addr) / + TCP(sport=remote_dport, dport=outside_port, flags="FA", seq=300, + ack=101)) + self.send_and_expect(pg1, [p], pg0) + + # ACK packet in -> out + p = (Ether(src=pg0.remote_mac, dst=pg0.local_mac) / + IP(src=local_host, dst=remote_host) / + TCP(sport=local_sport, dport=remote_dport, flags="A", seq=101, + ack=301)) + self.send_and_expect(pg0, [p], pg1) + + # session now in transitory timeout + # try SYN packet in->out - should be dropped + p = (Ether(src=pg0.remote_mac, dst=pg0.local_mac) / + IP(src=local_host, dst=remote_host) / + TCP(sport=local_sport, dport=remote_dport, flags="S")) + pg0.add_stream(p) + self.pg_enable_capture() + self.pg_start() + + self.sleep(new_transitory, "wait for transitory timeout") + pg0.assert_nothing_captured(0) + + # session should still exist + sessions = self.vapi.nat44_user_session_dump(pg0.remote_ip4, 0) + self.assertEqual(1, len(sessions)) + + # send FIN+ACK packet in->out - will cause session to be wiped + # but won't create a new session + p = (Ether(src=pg0.remote_mac, dst=pg0.local_mac) / + IP(src=local_host, dst=remote_host) / + TCP(sport=local_sport, dport=remote_dport, flags="FA", seq=300, + ack=101)) + pg1.add_stream(p) + self.pg_enable_capture() + self.pg_start() + pg0.assert_nothing_captured(0) + + sessions = self.vapi.nat44_user_session_dump(pg0.remote_ip4, 0) + self.assertEqual(0, len(sessions)) + + # create a new session and make sure the outside port is remapped + # SYN packet in->out + + p = (Ether(src=pg0.remote_mac, dst=pg0.local_mac) / + IP(src=local_host, dst=remote_host) / + TCP(sport=local_sport, dport=remote_dport, flags="S")) + p = self.send_and_expect(pg0, [p], pg1)[0] + + self.assertEqual(p[IP].src, outside_addr) + self.assertNotEqual(p[TCP].sport, local_sport) + + # make sure static mapping works and creates a new session + # SYN packet out->in + p = (Ether(src=pg1.remote_mac, dst=pg1.local_mac) / + IP(src=remote_host, dst=outside_addr) / + TCP(sport=remote_dport, dport=outside_port, flags="S")) + self.send_and_expect(pg1, [p], pg0) + + sessions = self.vapi.nat44_user_session_dump(pg0.remote_ip4, 0) + self.assertEqual(2, len(sessions)) + + +if __name__ == '__main__': + unittest.main(testRunner=VppTestRunner) diff --git a/test/test_nat44_ei.py b/test/test_nat44_ei.py index 74a082eee0a..5fdcf3fa3c7 100644 --- a/test/test_nat44_ei.py +++ b/test/test_nat44_ei.py @@ -605,7 +605,7 @@ class MethodHolder(VppTestCase): self.assertEqual(struct.pack("!H", self.udp_port_out), record[227]) else: - self.fail("Invalid protocol") + self.fail(f"Invalid protocol {scapy.compat.orb(record[4])}") self.assertEqual(3, nat44_ses_create_num) self.assertEqual(3, nat44_ses_delete_num) |