diff options
Diffstat (limited to 'test/test_nat44_ed_output.py')
| -rw-r--r-- | test/test_nat44_ed_output.py | 229 |
1 files changed, 229 insertions, 0 deletions
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) |