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
  26     max_sessions = 1024
  27
  28     @classmethod
  29     def setUpClass(cls):
  30         super().setUpClass()
  31         cls.create_pg_interfaces(range(2))
  32         cls.interfaces = list(cls.pg_interfaces)
  33
  34     @classmethod
  35     def tearDownClass(cls):
  36         super().tearDownClass()
  37
  38     def setUp(self):
  39         super().setUp()
  40         for i in self.interfaces:
  41             i.admin_up()
  42             i.config_ip4()
  43             i.resolve_arp()
  44         self.vapi.nat44_ed_plugin_enable_disable(sessions=self.max_sessions, 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
  69         local_host = self.pg0.remote_ip4
  70         remote_host = self.pg1.remote_ip4
  71         nat_intf = self.pg1
  72         outside_addr = nat_intf.local_ip4
  73
  74         self.vapi.nat44_add_del_address_range(
  75             first_ip_address=outside_addr,
  76             last_ip_address=outside_addr,
  77             vrf_id=0xFFFFFFFF,
  78             is_add=1,
  79             flags=0,
  80         )
  81         self.vapi.nat44_interface_add_del_feature(
  82             sw_if_index=self.pg0.sw_if_index, is_add=1
  83         )
  84         self.vapi.nat44_interface_add_del_feature(
  85             sw_if_index=self.pg0.sw_if_index,
  86             flags=VppEnum.vl_api_nat_config_flags_t.NAT_IS_INSIDE,
  87             is_add=1,
  88         )
  89         self.vapi.nat44_ed_add_del_output_interface(
  90             sw_if_index=self.pg1.sw_if_index, is_add=1
  91         )
  92
  93         thread_index = get_nat44_ed_in2out_worker_index(
  94             local_host, self.vpp_worker_count
  95         )
  96         port_per_thread = int((0xFFFF - 1024) / max(1, self.vpp_worker_count))
  97         local_sport = 1024 + random.randint(1, port_per_thread)
  98         if self.vpp_worker_count > 0:
  99             local_sport += port_per_thread * (thread_index - 1)
 100
 101         remote_dport = 10000
 102
 103         pg0 = self.pg0
 104         pg1 = self.pg1
 105
 106         # first setup a dynamic TCP session
 107
 108         # SYN packet in->out
 109         p = (
 110             Ether(src=pg0.remote_mac, dst=pg0.local_mac)
 111             / IP(src=local_host, dst=remote_host)
 112             / TCP(sport=local_sport, dport=remote_dport, flags="S")
 113         )
 114         p = self.send_and_expect(pg0, [p], pg1)[0]
 115
 116         self.assertEqual(p[IP].src, outside_addr)
 117         self.assertEqual(p[TCP].sport, local_sport)
 118         outside_port = p[TCP].sport
 119
 120         # SYN+ACK packet out->in
 121         p = (
 122             Ether(src=pg1.remote_mac, dst=pg1.local_mac)
 123             / IP(src=remote_host, dst=outside_addr)
 124             / TCP(sport=remote_dport, dport=outside_port, flags="SA")
 125         )
 126         self.send_and_expect(pg1, [p], pg0)
 127
 128         # ACK packet in->out
 129         p = (
 130             Ether(src=pg0.remote_mac, dst=pg0.local_mac)
 131             / IP(src=local_host, dst=remote_host)
 132             / TCP(sport=local_sport, dport=remote_dport, flags="A")
 133         )
 134         self.send_and_expect(pg0, [p], pg1)
 135
 136         # now we have a session up, create a conflicting static mapping
 137         self.vapi.nat44_add_del_static_mapping(
 138             is_add=1,
 139             local_ip_address=local_host,
 140             external_ip_address=outside_addr,
 141             external_sw_if_index=0xFFFFFFFF,
 142             local_port=local_sport,
 143             external_port=outside_port,
 144             protocol=IP_PROTOS.tcp,
 145             flags=VppEnum.vl_api_nat_config_flags_t.NAT_IS_OUT2IN_ONLY,
 146         )
 147
 148         sessions = self.vapi.nat44_user_session_dump(local_host, 0)
 149         self.assertEqual(1, len(sessions))
 150
 151         # now send some more data over existing session - it should pass
 152
 153         # in->out
 154         p = (
 155             Ether(src=pg0.remote_mac, dst=pg0.local_mac)
 156             / IP(src=local_host, dst=remote_host)
 157             / TCP(sport=local_sport, dport=remote_dport)
 158             / Raw("zippity zap")
 159         )
 160         self.send_and_expect(pg0, [p], pg1)
 161
 162         # out->in
 163         p = (
 164             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)
 167             / Raw("flippity flop")
 168         )
 169         self.send_and_expect(pg1, [p], pg0)
 170
 171         # now close the session
 172
 173         # FIN packet in -> out
 174         p = (
 175             Ether(src=pg0.remote_mac, dst=pg0.local_mac)
 176             / IP(src=local_host, dst=remote_host)
 177             / TCP(sport=local_sport, dport=remote_dport, flags="FA", seq=100, ack=300)
 178         )
 179         self.send_and_expect(pg0, [p], pg1)
 180
 181         # FIN+ACK packet out -> in
 182         p = (
 183             Ether(src=pg1.remote_mac, dst=pg1.local_mac)
 184             / IP(src=remote_host, dst=outside_addr)
 185             / TCP(sport=remote_dport, dport=outside_port, flags="FA", seq=300, ack=101)
 186         )
 187         self.send_and_expect(pg1, [p], pg0)
 188
 189         # ACK packet in -> out
 190         p = (
 191             Ether(src=pg0.remote_mac, dst=pg0.local_mac)
 192             / IP(src=local_host, dst=remote_host)
 193             / TCP(sport=local_sport, dport=remote_dport, flags="A", seq=101, ack=301)
 194         )
 195         self.send_and_expect(pg0, [p], pg1)
 196
 197         # session now in transitory timeout
 198         # try SYN packet in->out - should be dropped
 199         p = (
 200             Ether(src=pg0.remote_mac, dst=pg0.local_mac)
 201             / IP(src=local_host, dst=remote_host)
 202             / TCP(sport=local_sport, dport=remote_dport, flags="S")
 203         )
 204         pg0.add_stream(p)
 205         self.pg_enable_capture()
 206         self.pg_start()
 207
 208         self.sleep(new_transitory, "wait for transitory timeout")
 209         pg0.assert_nothing_captured(0)
 210
 211         # session should still exist
 212         sessions = self.vapi.nat44_user_session_dump(pg0.remote_ip4, 0)
 213         self.assertEqual(1, len(sessions))
 214
 215         # send FIN+ACK packet in->out - will cause session to be wiped
 216         # but won't create a new session
 217         p = (
 218             Ether(src=pg0.remote_mac, dst=pg0.local_mac)
 219             / IP(src=local_host, dst=remote_host)
 220             / TCP(sport=local_sport, dport=remote_dport, flags="FA", seq=300, ack=101)
 221         )
 222         pg1.add_stream(p)
 223         self.pg_enable_capture()
 224         self.pg_start()
 225         pg0.assert_nothing_captured(0)
 226
 227         sessions = self.vapi.nat44_user_session_dump(pg0.remote_ip4, 0)
 228         self.assertEqual(0, len(sessions))
 229
 230         # create a new session and make sure the outside port is remapped
 231         # SYN packet in->out
 232
 233         p = (
 234             Ether(src=pg0.remote_mac, dst=pg0.local_mac)
 235             / IP(src=local_host, dst=remote_host)
 236             / TCP(sport=local_sport, dport=remote_dport, flags="S")
 237         )
 238         p = self.send_and_expect(pg0, [p], pg1)[0]
 239
 240         self.assertEqual(p[IP].src, outside_addr)
 241         self.assertNotEqual(p[TCP].sport, local_sport)
 242
 243         # make sure static mapping works and creates a new session
 244         # SYN packet out->in
 245         p = (
 246             Ether(src=pg1.remote_mac, dst=pg1.local_mac)
 247             / IP(src=remote_host, dst=outside_addr)
 248             / TCP(sport=remote_dport, dport=outside_port, flags="S")
 249         )
 250         self.send_and_expect(pg1, [p], pg0)
 251
 252         sessions = self.vapi.nat44_user_session_dump(pg0.remote_ip4, 0)
 253         self.assertEqual(2, len(sessions))
 254
 255
 256 if __name__ == "__main__":
 257     unittest.main(testRunner=VppTestRunner)