2 """Policy 1:1 NAT functional tests"""
5 from scapy.layers.inet import Ether, IP, UDP, ICMP
6 from framework import VppTestCase, VppTestRunner
7 from vpp_papi import VppEnum
10 class TestPNAT(VppTestCase):
11 """ PNAT Test Case """
16 super(TestPNAT, cls).setUpClass()
17 cls.create_pg_interfaces(range(2))
18 cls.interfaces = list(cls.pg_interfaces)
21 def tearDownClass(cls):
22 super(TestPNAT, cls).tearDownClass()
25 super(TestPNAT, self).setUp()
26 for i in self.interfaces:
32 super(TestPNAT, self).tearDown()
34 for i in self.pg_interfaces:
38 def validate(self, rx, expected):
39 self.assertEqual(rx, expected.__class__(expected))
41 def validate_bytes(self, rx, expected):
42 self.assertEqual(rx, expected)
45 """ Verify non matching traffic works. """
46 p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
48 icmpecho = (IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) /
50 reply = (IP(src=self.pg0.local_ip4, dst=self.pg0.remote_ip4) /
51 ICMP(type='echo-reply'))
52 rx = self.send_and_expect(self.pg0, p_ether/icmpecho * 1, self.pg0)
54 reply[IP].id = p[IP].id
55 self.validate(p[1], reply)
60 PNAT_IP4_INPUT = VppEnum.vl_api_pnat_attachment_point_t.PNAT_IP4_INPUT
62 VppEnum.vl_api_pnat_attachment_point_t.PNAT_IP4_OUTPUT
66 'input': PNAT_IP4_INPUT,
67 'sw_if_index': self.pg0.sw_if_index,
68 'match': {'mask': 0xa, 'dst': '10.10.10.10', 'proto': 17,
70 'rewrite': {'mask': 0x2, 'dst': self.pg1.remote_ip4},
71 'send': (IP(src=self.pg0.remote_ip4, dst='10.10.10.10') /
73 'reply': (IP(src=self.pg0.remote_ip4,
74 dst=self.pg1.remote_ip4) /
78 'input': PNAT_IP4_OUTPUT,
79 'sw_if_index': self.pg1.sw_if_index,
80 'match': {'mask': 0x9, 'src': self.pg0.remote_ip4, 'proto': 17,
82 'rewrite': {'mask': 0x1, 'src': '11.11.11.11'},
83 'send': (IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
85 'reply': (IP(src='11.11.11.11', dst=self.pg1.remote_ip4) /
89 'input': PNAT_IP4_INPUT,
90 'sw_if_index': self.pg0.sw_if_index,
91 'match': {'mask': 0xa, 'dst': '10.10.10.10', 'proto': 17,
93 'rewrite': {'mask': 0xa, 'dst': self.pg1.remote_ip4,
95 'send': (IP(src=self.pg0.remote_ip4, dst='10.10.10.10') /
96 UDP(sport=65530, dport=6871)),
97 'reply': (IP(src=self.pg0.remote_ip4,
98 dst=self.pg1.remote_ip4) /
99 UDP(sport=65530, dport=5555))
102 'input': PNAT_IP4_INPUT,
103 'sw_if_index': self.pg0.sw_if_index,
104 'match': {'mask': 0xa, 'dst': self.pg1.remote_ip4, 'proto': 17,
106 'rewrite': {'mask': 0x8, 'dport': 5555},
107 'send': (IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
108 UDP(dport=6871, chksum=0)),
109 'reply': (IP(src=self.pg0.remote_ip4,
110 dst=self.pg1.remote_ip4) /
111 UDP(dport=5555, chksum=0))
114 'input': PNAT_IP4_INPUT,
115 'sw_if_index': self.pg0.sw_if_index,
116 'match': {'mask': 0x2, 'dst': self.pg1.remote_ip4, 'proto': 1},
117 'rewrite': {'mask': 0x1, 'src': '8.8.8.8'},
118 'send': (IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
120 'reply': IP(src='8.8.8.8', dst=self.pg1.remote_ip4)/ICMP(),
124 p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
126 rv = self.vapi.pnat_binding_add(match=t['match'],
127 rewrite=t['rewrite'])
128 self.vapi.pnat_binding_attach(sw_if_index=t['sw_if_index'],
129 attachment=t['input'],
130 binding_index=rv.binding_index)
134 rx = self.send_and_expect(self.pg0, p_ether/t['send']*1, self.pg1)
137 self.validate(p[1], reply)
141 self.vapi.pnat_binding_detach(sw_if_index=t['sw_if_index'],
142 attachment=t['input'],
143 binding_index=rv.binding_index)
144 self.vapi.pnat_binding_del(binding_index=rv.binding_index)
146 def test_pnat_show(self):
147 """ PNAT show tests """
149 PNAT_IP4_INPUT = VppEnum.vl_api_pnat_attachment_point_t.PNAT_IP4_INPUT
151 VppEnum.vl_api_pnat_attachment_point_t.PNAT_IP4_OUTPUT
155 'input': PNAT_IP4_INPUT,
156 'sw_if_index': self.pg0.sw_if_index,
157 'match': {'mask': 0xa, 'dst': '10.10.10.10', 'proto': 17,
159 'rewrite': {'mask': 0x2, 'dst': self.pg1.remote_ip4},
160 'send': (IP(src=self.pg0.remote_ip4, dst='10.10.10.10') /
162 'reply': (IP(src=self.pg0.remote_ip4,
163 dst=self.pg1.remote_ip4) /
167 'input': PNAT_IP4_OUTPUT,
168 'sw_if_index': self.pg1.sw_if_index,
169 'match': {'mask': 0x9, 'src': self.pg0.remote_ip4, 'proto': 17,
171 'rewrite': {'mask': 0x1, 'src': '11.11.11.11'},
172 'send': (IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
174 'reply': (IP(src='11.11.11.11', dst=self.pg1.remote_ip4) /
180 rv = self.vapi.pnat_binding_add(match=t['match'],
181 rewrite=t['rewrite'])
182 binding_index.append(rv.binding_index)
183 self.vapi.pnat_binding_attach(sw_if_index=t['sw_if_index'],
184 attachment=t['input'],
185 binding_index=rv.binding_index)
187 rv, l = self.vapi.pnat_bindings_get()
188 self.assertEqual(len(l), len(tests))
190 rv, l = self.vapi.pnat_interfaces_get()
191 self.assertEqual(len(l), 2)
193 self.logger.info(self.vapi.cli("show pnat translations"))
194 self.logger.info(self.vapi.cli("show pnat interfaces"))
196 for i, t in enumerate(tests):
197 self.vapi.pnat_binding_detach(sw_if_index=t['sw_if_index'],
198 attachment=t['input'],
199 binding_index=binding_index[i])
200 self.vapi.pnat_binding_del(binding_index=binding_index[i])
202 if __name__ == '__main__':
203 unittest.main(testRunner=VppTestRunner)