test/test_nat44_ed_output.py

   1 #!/usr/bin/env python3
   2 """NAT44 ED output-feature tests"""
   3
   4 import random
   5 import unittest
   6 from scapy.layers.inet import ICMP, Ether, IP, TCP
   7 from scapy.packet import Raw
   8 from scapy.data import IP_PROTOS
   9 from framework import VppTestCase, VppTestRunner
  10 from vpp_papi import VppEnum
  11
  12
  13 def get_nat44_ed_in2out_worker_index(ip, vpp_worker_count):
  14     if 0 == vpp_worker_count:
  15         return 0
  16     numeric = socket.inet_aton(ip)
  17     numeric = struct.unpack("!L", numeric)[0]
  18     numeric = socket.htonl(numeric)
  19     h = numeric + (numeric >> 8) + (numeric >> 16) + (numeric >> 24)
  20     return 1 + h % vpp_worker_count
  21
  22
  23 class TestNAT44EDOutput(VppTestCase):
  24     """ NAT44 ED output feature Test Case """
  25     max_sessions = 1024
  26
  27     @classmethod
  28     def setUpClass(cls):
  29         super().setUpClass()
  30         cls.create_pg_interfaces(range(2))
  31         cls.interfaces = list(cls.pg_interfaces)
  32
  33     @classmethod
  34     def tearDownClass(cls):
  35         super().tearDownClass()
  36
  37     def setUp(self):
  38         super().setUp()
  39         for i in self.interfaces:
  40             i.admin_up()
  41             i.config_ip4()
  42             i.resolve_arp()
  43         self.vapi.nat44_ed_plugin_enable_disable(sessions=self.max_sessions,
  44                                                  enable=1)
  45
  46     def tearDown(self):
  47         if not self.vpp_dead:
  48             self.logger.debug(self.vapi.cli("show nat44 sessions"))
  49         super().tearDown()
  50         if not self.vpp_dead:
  51             for i in self.pg_interfaces:
  52                 i.unconfig_ip4()
  53                 i.admin_down()
  54             self.vapi.nat44_ed_plugin_enable_disable(enable=0)
  55
  56     def test_static_dynamic(self):
  57         """ Create static mapping which matches existing dynamic mapping """
  58
  59         config = self.vapi.nat44_show_running_config()
  60         old_timeouts = config.timeouts
  61         new_transitory = 2
  62         self.vapi.nat_set_timeouts(
  63             udp=old_timeouts.udp,
  64             tcp_established=old_timeouts.tcp_established,
  65             icmp=old_timeouts.icmp,
  66             tcp_transitory=new_transitory)
  67
  68         local_host = self.pg0.remote_ip4
  69         remote_host = self.pg1.remote_ip4
  70         nat_intf = self.pg1
  71         outside_addr = nat_intf.local_ip4
  72
  73         self.vapi.nat44_add_del_address_range(first_ip_address=outside_addr,
  74                                               last_ip_address=outside_addr,
  75                                               vrf_id=0xffffffff,
  76                                               is_add=1,
  77                                               flags=0)
  78         self.vapi.nat44_interface_add_del_feature(
  79             sw_if_index=self.pg0.sw_if_index,
  80             is_add=1)
  81         self.vapi.nat44_interface_add_del_feature(
  82             sw_if_index=self.pg0.sw_if_index,
  83             flags=VppEnum.vl_api_nat_config_flags_t.NAT_IS_INSIDE, is_add=1)
  84         self.vapi.nat44_ed_add_del_output_interface(
  85             sw_if_index=self.pg1.sw_if_index,
  86             is_add=1)
  87
  88         thread_index = get_nat44_ed_in2out_worker_index(
  89             local_host, self.vpp_worker_count)
  90         port_per_thread = int((0xffff-1024) / max(1, self.vpp_worker_count))
  91         local_sport = 1024 + random.randint(1, port_per_thread)
  92         if self.vpp_worker_count > 0:
  93             local_sport += port_per_thread * (thread_index - 1)
  94
  95         remote_dport = 10000
  96
  97         pg0 = self.pg0
  98         pg1 = self.pg1
  99
 100         # first setup a dynamic TCP session
 101
 102         # SYN packet in->out
 103         p = (Ether(src=pg0.remote_mac, dst=pg0.local_mac) /
 104              IP(src=local_host, dst=remote_host) /
 105              TCP(sport=local_sport, dport=remote_dport, flags="S"))
 106         p = self.send_and_expect(pg0, [p], pg1)[0]
 107
 108         self.assertEqual(p[IP].src, outside_addr)
 109         self.assertEqual(p[TCP].sport, local_sport)
 110         outside_port = p[TCP].sport
 111
 112         # SYN+ACK packet out->in
 113         p = (Ether(src=pg1.remote_mac, dst=pg1.local_mac) /
 114              IP(src=remote_host, dst=outside_addr) /
 115              TCP(sport=remote_dport, dport=outside_port, flags="SA"))
 116         self.send_and_expect(pg1, [p], pg0)
 117
 118         # ACK packet in->out
 119         p = (Ether(src=pg0.remote_mac, dst=pg0.local_mac) /
 120              IP(src=local_host, dst=remote_host) /
 121              TCP(sport=local_sport, dport=remote_dport, flags="A"))
 122         self.send_and_expect(pg0, [p], pg1)
 123
 124         # now we have a session up, create a conflicting static mapping
 125         self.vapi.nat44_add_del_static_mapping(
 126             is_add=1,
 127             local_ip_address=local_host,
 128             external_ip_address=outside_addr,
 129             external_sw_if_index=0xffffffff,
 130             local_port=local_sport,
 131             external_port=outside_port,
 132             protocol=IP_PROTOS.tcp,
 133             flags=VppEnum.vl_api_nat_config_flags_t.NAT_IS_OUT2IN_ONLY)
 134
 135         sessions = self.vapi.nat44_user_session_dump(local_host, 0)
 136         self.assertEqual(1, len(sessions))
 137
 138         # now send some more data over existing session - it should pass
 139
 140         # in->out
 141         p = (Ether(src=pg0.remote_mac, dst=pg0.local_mac) /
 142              IP(src=local_host, dst=remote_host) /
 143              TCP(sport=local_sport, dport=remote_dport) /
 144              Raw("zippity zap"))
 145         self.send_and_expect(pg0, [p], pg1)
 146
 147         # out->in
 148         p = (Ether(src=pg1.remote_mac, dst=pg1.local_mac) /
 149              IP(src=remote_host, dst=outside_addr) /
 150              TCP(sport=remote_dport, dport=outside_port) /
 151              Raw("flippity flop"))
 152         self.send_and_expect(pg1, [p], pg0)
 153
 154         # now close the session
 155
 156         # FIN packet in -> out
 157         p = (Ether(src=pg0.remote_mac, dst=pg0.local_mac) /
 158              IP(src=local_host, dst=remote_host) /
 159              TCP(sport=local_sport, dport=remote_dport, flags="FA", seq=100,
 160                  ack=300))
 161         self.send_and_expect(pg0, [p], pg1)
 162
 163         # FIN+ACK packet out -> in
 164         p = (Ether(src=pg1.remote_mac, dst=pg1.local_mac) /
 165              IP(src=remote_host, dst=outside_addr) /
 166              TCP(sport=remote_dport, dport=outside_port, flags="FA", seq=300,
 167                  ack=101))
 168         self.send_and_expect(pg1, [p], pg0)
 169
 170         # ACK packet in -> out
 171         p = (Ether(src=pg0.remote_mac, dst=pg0.local_mac) /
 172              IP(src=local_host, dst=remote_host) /
 173              TCP(sport=local_sport, dport=remote_dport, flags="A", seq=101,
 174                  ack=301))
 175         self.send_and_expect(pg0, [p], pg1)
 176
 177         # session now in transitory timeout
 178         # try SYN packet in->out - should be dropped
 179         p = (Ether(src=pg0.remote_mac, dst=pg0.local_mac) /
 180              IP(src=local_host, dst=remote_host) /
 181              TCP(sport=local_sport, dport=remote_dport, flags="S"))
 182         pg0.add_stream(p)
 183         self.pg_enable_capture()
 184         self.pg_start()
 185
 186         self.sleep(new_transitory, "wait for transitory timeout")
 187         pg0.assert_nothing_captured(0)
 188
 189         # session should still exist
 190         sessions = self.vapi.nat44_user_session_dump(pg0.remote_ip4, 0)
 191         self.assertEqual(1, len(sessions))
 192
 193         # send FIN+ACK packet in->out - will cause session to be wiped
 194         # but won't create a new session
 195         p = (Ether(src=pg0.remote_mac, dst=pg0.local_mac) /
 196              IP(src=local_host, dst=remote_host) /
 197              TCP(sport=local_sport, dport=remote_dport, flags="FA", seq=300,
 198                  ack=101))
 199         pg1.add_stream(p)
 200         self.pg_enable_capture()
 201         self.pg_start()
 202         pg0.assert_nothing_captured(0)
 203
 204         sessions = self.vapi.nat44_user_session_dump(pg0.remote_ip4, 0)
 205         self.assertEqual(0, len(sessions))
 206
 207         # create a new session and make sure the outside port is remapped
 208         # SYN packet in->out
 209
 210         p = (Ether(src=pg0.remote_mac, dst=pg0.local_mac) /
 211              IP(src=local_host, dst=remote_host) /
 212              TCP(sport=local_sport, dport=remote_dport, flags="S"))
 213         p = self.send_and_expect(pg0, [p], pg1)[0]
 214
 215         self.assertEqual(p[IP].src, outside_addr)
 216         self.assertNotEqual(p[TCP].sport, local_sport)
 217
 218         # make sure static mapping works and creates a new session
 219         # SYN packet out->in
 220         p = (Ether(src=pg1.remote_mac, dst=pg1.local_mac) /
 221              IP(src=remote_host, dst=outside_addr) /
 222              TCP(sport=remote_dport, dport=outside_port, flags="S"))
 223         self.send_and_expect(pg1, [p], pg0)
 224
 225         sessions = self.vapi.nat44_user_session_dump(pg0.remote_ip4, 0)
 226         self.assertEqual(2, len(sessions))
 227
 228
 229 if __name__ == '__main__':
 230     unittest.main(testRunner=VppTestRunner)