2 """NAT44 ED output-feature tests"""
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
13 def get_nat44_ed_in2out_worker_index(ip, vpp_worker_count):
14 if 0 == vpp_worker_count:
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
23 class TestNAT44EDOutput(VppTestCase):
24 """ NAT44 ED output feature Test Case """
30 cls.create_pg_interfaces(range(2))
31 cls.interfaces = list(cls.pg_interfaces)
34 def tearDownClass(cls):
35 super().tearDownClass()
39 for i in self.interfaces:
43 self.vapi.nat44_ed_plugin_enable_disable(sessions=self.max_sessions,
48 self.logger.debug(self.vapi.cli("show nat44 sessions"))
51 for i in self.pg_interfaces:
54 self.vapi.nat44_ed_plugin_enable_disable(enable=0)
56 def test_static_dynamic(self):
57 """ Create static mapping which matches existing dynamic mapping """
59 old_timeouts = self.vapi.nat_get_timeouts()
61 self.vapi.nat_set_timeouts(
63 tcp_established=old_timeouts.tcp_established,
64 icmp=old_timeouts.icmp,
65 tcp_transitory=new_transitory)
67 local_host = self.pg0.remote_ip4
68 remote_host = self.pg1.remote_ip4
70 outside_addr = nat_intf.local_ip4
72 self.vapi.nat44_add_del_address_range(first_ip_address=outside_addr,
73 last_ip_address=outside_addr,
77 self.vapi.nat44_interface_add_del_feature(
78 sw_if_index=self.pg0.sw_if_index,
80 self.vapi.nat44_interface_add_del_feature(
81 sw_if_index=self.pg0.sw_if_index,
82 flags=VppEnum.vl_api_nat_config_flags_t.NAT_IS_INSIDE, is_add=1)
83 self.vapi.nat44_interface_add_del_output_feature(
85 sw_if_index=self.pg1.sw_if_index)
87 thread_index = get_nat44_ed_in2out_worker_index(
88 local_host, self.vpp_worker_count)
89 port_per_thread = int((0xffff-1024) / max(1, self.vpp_worker_count))
90 local_sport = 1024 + random.randint(1, port_per_thread)
91 if self.vpp_worker_count > 0:
92 local_sport += port_per_thread * (thread_index - 1)
99 # first setup a dynamic TCP session
102 p = (Ether(src=pg0.remote_mac, dst=pg0.local_mac) /
103 IP(src=local_host, dst=remote_host) /
104 TCP(sport=local_sport, dport=remote_dport, flags="S"))
105 p = self.send_and_expect(pg0, [p], pg1)[0]
107 self.assertEqual(p[IP].src, outside_addr)
108 self.assertEqual(p[TCP].sport, local_sport)
109 outside_port = p[TCP].sport
111 # SYN+ACK packet out->in
112 p = (Ether(src=pg1.remote_mac, dst=pg1.local_mac) /
113 IP(src=remote_host, dst=outside_addr) /
114 TCP(sport=remote_dport, dport=outside_port, flags="SA"))
115 self.send_and_expect(pg1, [p], pg0)
118 p = (Ether(src=pg0.remote_mac, dst=pg0.local_mac) /
119 IP(src=local_host, dst=remote_host) /
120 TCP(sport=local_sport, dport=remote_dport, flags="A"))
121 self.send_and_expect(pg0, [p], pg1)
123 # now we have a session up, create a conflicting static mapping
124 self.vapi.nat44_add_del_static_mapping(
126 local_ip_address=local_host,
127 external_ip_address=outside_addr,
128 external_sw_if_index=0xffffffff,
129 local_port=local_sport,
130 external_port=outside_port,
131 protocol=IP_PROTOS.tcp,
132 flags=VppEnum.vl_api_nat_config_flags_t.NAT_IS_OUT2IN_ONLY)
134 sessions = self.vapi.nat44_user_session_dump(local_host, 0)
135 self.assertEqual(1, len(sessions))
137 # now send some more data over existing session - it should pass
140 p = (Ether(src=pg0.remote_mac, dst=pg0.local_mac) /
141 IP(src=local_host, dst=remote_host) /
142 TCP(sport=local_sport, dport=remote_dport) /
144 self.send_and_expect(pg0, [p], pg1)
147 p = (Ether(src=pg1.remote_mac, dst=pg1.local_mac) /
148 IP(src=remote_host, dst=outside_addr) /
149 TCP(sport=remote_dport, dport=outside_port) /
150 Raw("flippity flop"))
151 self.send_and_expect(pg1, [p], pg0)
153 # now close the session
155 # FIN packet in -> out
156 p = (Ether(src=pg0.remote_mac, dst=pg0.local_mac) /
157 IP(src=local_host, dst=remote_host) /
158 TCP(sport=local_sport, dport=remote_dport, flags="FA", seq=100,
160 self.send_and_expect(pg0, [p], pg1)
162 # FIN+ACK packet out -> in
163 p = (Ether(src=pg1.remote_mac, dst=pg1.local_mac) /
164 IP(src=remote_host, dst=outside_addr) /
165 TCP(sport=remote_dport, dport=outside_port, flags="FA", seq=300,
167 self.send_and_expect(pg1, [p], pg0)
169 # ACK packet in -> out
170 p = (Ether(src=pg0.remote_mac, dst=pg0.local_mac) /
171 IP(src=local_host, dst=remote_host) /
172 TCP(sport=local_sport, dport=remote_dport, flags="A", seq=101,
174 self.send_and_expect(pg0, [p], pg1)
176 # session now in transitory timeout
177 # try SYN packet in->out - should be dropped
178 p = (Ether(src=pg0.remote_mac, dst=pg0.local_mac) /
179 IP(src=local_host, dst=remote_host) /
180 TCP(sport=local_sport, dport=remote_dport, flags="S"))
182 self.pg_enable_capture()
185 self.sleep(new_transitory, "wait for transitory timeout")
186 pg0.assert_nothing_captured(0)
188 # session should still exist
189 sessions = self.vapi.nat44_user_session_dump(pg0.remote_ip4, 0)
190 self.assertEqual(1, len(sessions))
192 # send FIN+ACK packet in->out - will cause session to be wiped
193 # but won't create a new session
194 p = (Ether(src=pg0.remote_mac, dst=pg0.local_mac) /
195 IP(src=local_host, dst=remote_host) /
196 TCP(sport=local_sport, dport=remote_dport, flags="FA", seq=300,
199 self.pg_enable_capture()
201 pg0.assert_nothing_captured(0)
203 sessions = self.vapi.nat44_user_session_dump(pg0.remote_ip4, 0)
204 self.assertEqual(0, len(sessions))
206 # create a new session and make sure the outside port is remapped
209 p = (Ether(src=pg0.remote_mac, dst=pg0.local_mac) /
210 IP(src=local_host, dst=remote_host) /
211 TCP(sport=local_sport, dport=remote_dport, flags="S"))
212 p = self.send_and_expect(pg0, [p], pg1)[0]
214 self.assertEqual(p[IP].src, outside_addr)
215 self.assertNotEqual(p[TCP].sport, local_sport)
217 # make sure static mapping works and creates a new session
219 p = (Ether(src=pg1.remote_mac, dst=pg1.local_mac) /
220 IP(src=remote_host, dst=outside_addr) /
221 TCP(sport=remote_dport, dport=outside_port, flags="S"))
222 self.send_and_expect(pg1, [p], pg0)
224 sessions = self.vapi.nat44_user_session_dump(pg0.remote_ip4, 0)
225 self.assertEqual(2, len(sessions))
228 if __name__ == '__main__':
229 unittest.main(testRunner=VppTestRunner)