8 from framework import VppTestCase, VppTestRunner
10 from scapy.packet import Raw
11 from scapy.layers.l2 import Ether
12 from scapy.layers.inet import IP, UDP
16 class TestClassifier(VppTestCase):
17 """ Classifier Test Case """
21 Perform test setup before test case.
24 - create 4 pg interfaces
25 - untagged pg0/pg1/pg2 interface
26 pg0 -------> pg1 (IP ACL)
32 - put it into UP state
34 - resolve neighbor address using ARP
36 :ivar list interfaces: pg interfaces.
37 :ivar list pg_if_packet_sizes: packet sizes in test.
38 :ivar dict acl_tbl_idx: ACL table index.
39 :ivar int pbr_vrfid: VRF id for PBR test.
41 super(TestClassifier, self).setUp()
43 # create 4 pg interfaces
44 self.create_pg_interfaces(range(4))
46 # packet sizes to test
47 self.pg_if_packet_sizes = [64, 9018]
49 self.interfaces = list(self.pg_interfaces)
55 # setup all interfaces
56 for intf in self.interfaces:
62 """Run standard test teardown and acl related log."""
63 super(TestClassifier, self).tearDown()
65 self.logger.info(self.vapi.cli("show classify table verbose"))
66 self.logger.info(self.vapi.cli("show ip fib"))
68 def config_pbr_fib_entry(self, intf, is_add=1):
69 """Configure fib entry to route traffic toward PBR VRF table
71 :param VppInterface intf: destination interface to be routed for PBR.
75 self.vapi.ip_add_del_route(intf.local_ip4n,
78 table_id=self.pbr_vrfid,
81 def create_stream(self, src_if, dst_if, packet_sizes):
82 """Create input packet stream for defined interfaces.
84 :param VppInterface src_if: Source Interface for packet stream.
85 :param VppInterface dst_if: Destination Interface for packet stream.
86 :param list packet_sizes: packet size to test.
89 for size in packet_sizes:
90 info = self.create_packet_info(src_if, dst_if)
91 payload = self.info_to_payload(info)
92 p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
93 IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4) /
94 UDP(sport=1234, dport=5678) /
97 self.extend_packet(p, size)
101 def verify_capture(self, dst_if, capture):
102 """Verify captured input packet stream for defined interface.
104 :param VppInterface dst_if: Interface to verify captured packet stream.
105 :param list capture: Captured packet stream.
107 self.logger.info("Verifying capture on interface %s" % dst_if.name)
109 for i in self.interfaces:
110 last_info[i.sw_if_index] = None
111 dst_sw_if_index = dst_if.sw_if_index
112 for packet in capture:
116 payload_info = self.payload_to_info(str(packet[Raw]))
117 packet_index = payload_info.index
118 self.assertEqual(payload_info.dst, dst_sw_if_index)
120 "Got packet on port %s: src=%u (id=%u)" %
121 (dst_if.name, payload_info.src, packet_index))
122 next_info = self.get_next_packet_info_for_interface2(
123 payload_info.src, dst_sw_if_index,
124 last_info[payload_info.src])
125 last_info[payload_info.src] = next_info
126 self.assertTrue(next_info is not None)
127 self.assertEqual(packet_index, next_info.index)
128 saved_packet = next_info.data
129 # Check standard fields
130 self.assertEqual(ip.src, saved_packet[IP].src)
131 self.assertEqual(ip.dst, saved_packet[IP].dst)
132 self.assertEqual(udp.sport, saved_packet[UDP].sport)
133 self.assertEqual(udp.dport, saved_packet[UDP].dport)
135 self.logger.error(ppp("Unexpected or invalid packet:", packet))
137 for i in self.interfaces:
138 remaining_packet = self.get_next_packet_info_for_interface2(
139 i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index])
140 self.assertTrue(remaining_packet is None,
141 "Interface %s: Packet expected from interface %s "
142 "didn't arrive" % (dst_if.name, i.name))
144 def verify_vrf(self, vrf_id):
146 Check if the FIB table / VRF ID is configured.
148 :param int vrf_id: The FIB table / VRF ID to be verified.
149 :return: 1 if the FIB table / VRF ID is configured, otherwise return 0.
151 ip_fib_dump = self.vapi.ip_fib_dump()
153 for ip_fib_details in ip_fib_dump:
154 if ip_fib_details[2] == vrf_id:
157 self.logger.info("IPv4 VRF ID %d is not configured" % vrf_id)
160 self.logger.info("IPv4 VRF ID %d is configured" % vrf_id)
164 def build_ip_mask(proto='', src_ip='', dst_ip='',
165 src_port='', dst_port=''):
166 """Build IP ACL mask data with hexstring format
168 :param str proto: protocol number <0-ff>
169 :param str src_ip: source ip address <0-ffffffff>
170 :param str dst_ip: destination ip address <0-ffffffff>
171 :param str src_port: source port number <0-ffff>
172 :param str dst_port: destination port number <0-ffff>
175 return ('{:0>20}{:0>12}{:0>8}{:0>12}{:0>4}'.format(
176 proto, src_ip, dst_ip, src_port, dst_port)).rstrip('0')
179 def build_ip_match(proto='', src_ip='', dst_ip='',
180 src_port='', dst_port=''):
181 """Build IP ACL match data with hexstring format
183 :param str proto: protocol number with valid option "<0-ff>"
184 :param str src_ip: source ip address with format of "x.x.x.x"
185 :param str dst_ip: destination ip address with format of "x.x.x.x"
186 :param str src_port: source port number <0-ffff>
187 :param str dst_port: destination port number <0-ffff>
190 src_ip = socket.inet_aton(src_ip).encode('hex')
192 dst_ip = socket.inet_aton(dst_ip).encode('hex')
194 return ('{:0>20}{:0>12}{:0>8}{:0>12}{:0>4}'.format(
195 proto, src_ip, dst_ip, src_port, dst_port)).rstrip('0')
198 def build_mac_mask(dst_mac='', src_mac='', ether_type=''):
199 """Build MAC ACL mask data with hexstring format
201 :param str dst_mac: source MAC address <0-ffffffffffff>
202 :param str src_mac: destination MAC address <0-ffffffffffff>
203 :param str ether_type: ethernet type <0-ffff>
206 return ('{:0>12}{:0>12}{:0>4}'.format(dst_mac, src_mac,
207 ether_type)).rstrip('0')
210 def build_mac_match(dst_mac='', src_mac='', ether_type=''):
211 """Build MAC ACL match data with hexstring format
213 :param str dst_mac: source MAC address <x:x:x:x:x:x>
214 :param str src_mac: destination MAC address <x:x:x:x:x:x>
215 :param str ether_type: ethernet type <0-ffff>
218 dst_mac = dst_mac.replace(':', '')
220 src_mac = src_mac.replace(':', '')
222 return ('{:0>12}{:0>12}{:0>4}'.format(dst_mac, src_mac,
223 ether_type)).rstrip('0')
225 def create_classify_table(self, key, mask, data_offset=0, is_add=1):
226 """Create Classify Table
228 :param str key: key for classify table (ex, ACL name).
229 :param str mask: mask value for interested traffic.
230 :param int match_n_vectors:
231 :param int is_add: option to configure classify table.
232 - create(1) or delete(0)
234 r = self.vapi.classify_add_del_table(
236 binascii.unhexlify(mask),
237 match_n_vectors=(len(mask) - 1) // 32 + 1,
240 current_data_offset=data_offset)
241 self.assertIsNotNone(r, msg='No response msg for add_del_table')
242 self.acl_tbl_idx[key] = r.new_table_index
244 def create_classify_session(self, intf, table_index, match,
245 pbr_option=0, vrfid=0, is_add=1):
246 """Create Classify Session
248 :param VppInterface intf: Interface to apply classify session.
249 :param int table_index: table index to identify classify table.
250 :param str match: matched value for interested traffic.
251 :param int pbr_action: enable/disable PBR feature.
252 :param int vrfid: VRF id.
253 :param int is_add: option to configure classify session.
254 - create(1) or delete(0)
256 r = self.vapi.classify_add_del_session(
259 binascii.unhexlify(match),
263 self.assertIsNotNone(r, msg='No response msg for add_del_session')
265 def input_acl_set_interface(self, intf, table_index, is_add=1):
266 """Configure Input ACL interface
268 :param VppInterface intf: Interface to apply Input ACL feature.
269 :param int table_index: table index to identify classify table.
270 :param int is_add: option to configure classify session.
271 - enable(1) or disable(0)
273 r = self.vapi.input_acl_set_interface(
276 ip4_table_index=table_index)
277 self.assertIsNotNone(r, msg='No response msg for acl_set_interface')
279 def output_acl_set_interface(self, intf, table_index, is_add=1):
280 """Configure Output ACL interface
282 :param VppInterface intf: Interface to apply Output ACL feature.
283 :param int table_index: table index to identify classify table.
284 :param int is_add: option to configure classify session.
285 - enable(1) or disable(0)
287 r = self.vapi.output_acl_set_interface(
290 ip4_table_index=table_index)
291 self.assertIsNotNone(r, msg='No response msg for acl_set_interface')
293 def test_acl_ip(self):
296 Test scenario for basic IP ACL with source IP
297 - Create IPv4 stream for pg0 -> pg1 interface.
298 - Create ACL with source IP address.
299 - Send and verify received packets on pg1 interface.
302 # Basic ACL testing with source IP
303 pkts = self.create_stream(self.pg0, self.pg1, self.pg_if_packet_sizes)
304 self.pg0.add_stream(pkts)
306 self.create_classify_table('ip', self.build_ip_mask(src_ip='ffffffff'))
307 self.create_classify_session(
308 self.pg0, self.acl_tbl_idx.get('ip'),
309 self.build_ip_match(src_ip=self.pg0.remote_ip4))
310 self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('ip'))
312 self.pg_enable_capture(self.pg_interfaces)
315 pkts = self.pg1.get_capture(len(pkts))
316 self.verify_capture(self.pg1, pkts)
317 self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('ip'), 0)
318 self.pg0.assert_nothing_captured(remark="packets forwarded")
319 self.pg2.assert_nothing_captured(remark="packets forwarded")
320 self.pg3.assert_nothing_captured(remark="packets forwarded")
322 def test_acl_ip_out(self):
323 """ Output IP ACL test
325 Test scenario for basic IP ACL with source IP
326 - Create IPv4 stream for pg1 -> pg0 interface.
327 - Create ACL with source IP address.
328 - Send and verify received packets on pg0 interface.
331 # Basic ACL testing with source IP
332 pkts = self.create_stream(self.pg1, self.pg0, self.pg_if_packet_sizes)
333 self.pg1.add_stream(pkts)
335 self.create_classify_table('ip', self.build_ip_mask(src_ip='ffffffff'),
337 self.create_classify_session(
338 self.pg1, self.acl_tbl_idx.get('ip'),
339 self.build_ip_match(src_ip=self.pg1.remote_ip4))
340 self.output_acl_set_interface(self.pg0, self.acl_tbl_idx.get('ip'))
342 self.pg_enable_capture(self.pg_interfaces)
345 pkts = self.pg0.get_capture(len(pkts))
346 self.verify_capture(self.pg0, pkts)
347 self.output_acl_set_interface(self.pg0, self.acl_tbl_idx.get('ip'), 0)
348 self.pg1.assert_nothing_captured(remark="packets forwarded")
349 self.pg2.assert_nothing_captured(remark="packets forwarded")
350 self.pg3.assert_nothing_captured(remark="packets forwarded")
352 def test_acl_mac(self):
355 Test scenario for basic MAC ACL with source MAC
356 - Create IPv4 stream for pg0 -> pg2 interface.
357 - Create ACL with source MAC address.
358 - Send and verify received packets on pg2 interface.
361 # Basic ACL testing with source MAC
362 pkts = self.create_stream(self.pg0, self.pg2, self.pg_if_packet_sizes)
363 self.pg0.add_stream(pkts)
365 self.create_classify_table('mac',
366 self.build_mac_mask(src_mac='ffffffffffff'),
368 self.create_classify_session(
369 self.pg0, self.acl_tbl_idx.get('mac'),
370 self.build_mac_match(src_mac=self.pg0.remote_mac))
371 self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('mac'))
373 self.pg_enable_capture(self.pg_interfaces)
376 pkts = self.pg2.get_capture(len(pkts))
377 self.verify_capture(self.pg2, pkts)
378 self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('mac'), 0)
379 self.pg0.assert_nothing_captured(remark="packets forwarded")
380 self.pg1.assert_nothing_captured(remark="packets forwarded")
381 self.pg3.assert_nothing_captured(remark="packets forwarded")
383 def test_acl_pbr(self):
386 Test scenario for PBR with source IP
387 - Create IPv4 stream for pg0 -> pg3 interface.
388 - Configure PBR fib entry for packet forwarding.
389 - Send and verify received packets on pg3 interface.
392 # PBR testing with source IP
393 pkts = self.create_stream(self.pg0, self.pg3, self.pg_if_packet_sizes)
394 self.pg0.add_stream(pkts)
396 self.create_classify_table(
397 'pbr', self.build_ip_mask(
400 # this will create the VRF/table in which we will insert the route
401 self.create_classify_session(
402 self.pg0, self.acl_tbl_idx.get('pbr'),
403 self.build_ip_match(src_ip=self.pg0.remote_ip4),
404 pbr_option, self.pbr_vrfid)
405 self.assertTrue(self.verify_vrf(self.pbr_vrfid))
406 self.config_pbr_fib_entry(self.pg3)
407 self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('pbr'))
409 self.pg_enable_capture(self.pg_interfaces)
412 pkts = self.pg3.get_capture(len(pkts))
413 self.verify_capture(self.pg3, pkts)
414 self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('pbr'), 0)
415 self.pg0.assert_nothing_captured(remark="packets forwarded")
416 self.pg1.assert_nothing_captured(remark="packets forwarded")
417 self.pg2.assert_nothing_captured(remark="packets forwarded")
419 # remove the classify session and the route
420 self.config_pbr_fib_entry(self.pg3, is_add=0)
421 self.create_classify_session(
422 self.pg0, self.acl_tbl_idx.get('pbr'),
423 self.build_ip_match(src_ip=self.pg0.remote_ip4),
424 pbr_option, self.pbr_vrfid, is_add=0)
426 # and the table should be gone.
427 self.assertFalse(self.verify_vrf(self.pbr_vrfid))
429 if __name__ == '__main__':
430 unittest.main(testRunner=VppTestRunner)