summaryrefslogtreecommitdiffstats
path: root/test/test_nat44_ed_output.py
diff options
context:
space:
mode:
Diffstat (limited to 'test/test_nat44_ed_output.py')
-rw-r--r--test/test_nat44_ed_output.py229
1 files changed, 0 insertions, 229 deletions
diff --git a/test/test_nat44_ed_output.py b/test/test_nat44_ed_output.py
deleted file mode 100644
index ea5c14e7064..00000000000
--- a/test/test_nat44_ed_output.py
+++ /dev/null
@@ -1,229 +0,0 @@
-#!/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)