7 from framework import VppTestCase, VppTestRunner
9 from scapy.packet import Raw
10 from scapy.layers.l2 import Ether
11 from scapy.layers.inet import IP, UDP
15 class TestClassifier(VppTestCase):
16 """ Classifier Test Case """
20 Perform test setup before test case.
23 - create 4 pg interfaces
24 - untagged pg0/pg1/pg2 interface
25 pg0 -------> pg1 (IP ACL)
31 - put it into UP state
33 - resolve neighbor address using ARP
35 :ivar list interfaces: pg interfaces.
36 :ivar list pg_if_packet_sizes: packet sizes in test.
37 :ivar dict acl_tbl_idx: ACL table index.
38 :ivar int pbr_vrfid: VRF id for PBR test.
40 super(TestClassifier, self).setUp()
42 # create 4 pg interfaces
43 self.create_pg_interfaces(range(4))
45 # packet sizes to test
46 self.pg_if_packet_sizes = [64, 9018]
48 self.interfaces = list(self.pg_interfaces)
54 # setup all interfaces
55 for intf in self.interfaces:
61 """Run standard test teardown and acl related log."""
62 super(TestClassifier, self).tearDown()
64 self.logger.info(self.vapi.cli("show classify table verbose"))
65 self.logger.info(self.vapi.cli("show ip fib"))
67 def config_pbr_fib_entry(self, intf, is_add=1):
68 """Configure fib entry to route traffic toward PBR VRF table
70 :param VppInterface intf: destination interface to be routed for PBR.
74 self.vapi.ip_add_del_route(intf.local_ip4n,
77 table_id=self.pbr_vrfid,
80 def create_stream(self, src_if, dst_if, packet_sizes):
81 """Create input packet stream for defined interfaces.
83 :param VppInterface src_if: Source Interface for packet stream.
84 :param VppInterface dst_if: Destination Interface for packet stream.
85 :param list packet_sizes: packet size to test.
88 for size in packet_sizes:
89 info = self.create_packet_info(src_if, dst_if)
90 payload = self.info_to_payload(info)
91 p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
92 IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4) /
93 UDP(sport=1234, dport=5678) /
96 self.extend_packet(p, size)
100 def verify_capture(self, dst_if, capture):
101 """Verify captured input packet stream for defined interface.
103 :param VppInterface dst_if: Interface to verify captured packet stream.
104 :param list capture: Captured packet stream.
106 self.logger.info("Verifying capture on interface %s" % dst_if.name)
108 for i in self.interfaces:
109 last_info[i.sw_if_index] = None
110 dst_sw_if_index = dst_if.sw_if_index
111 for packet in capture:
115 payload_info = self.payload_to_info(str(packet[Raw]))
116 packet_index = payload_info.index
117 self.assertEqual(payload_info.dst, dst_sw_if_index)
119 "Got packet on port %s: src=%u (id=%u)" %
120 (dst_if.name, payload_info.src, packet_index))
121 next_info = self.get_next_packet_info_for_interface2(
122 payload_info.src, dst_sw_if_index,
123 last_info[payload_info.src])
124 last_info[payload_info.src] = next_info
125 self.assertTrue(next_info is not None)
126 self.assertEqual(packet_index, next_info.index)
127 saved_packet = next_info.data
128 # Check standard fields
129 self.assertEqual(ip.src, saved_packet[IP].src)
130 self.assertEqual(ip.dst, saved_packet[IP].dst)
131 self.assertEqual(udp.sport, saved_packet[UDP].sport)
132 self.assertEqual(udp.dport, saved_packet[UDP].dport)
134 self.logger.error(ppp("Unexpected or invalid packet:", packet))
136 for i in self.interfaces:
137 remaining_packet = self.get_next_packet_info_for_interface2(
138 i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index])
139 self.assertTrue(remaining_packet is None,
140 "Interface %s: Packet expected from interface %s "
141 "didn't arrive" % (dst_if.name, i.name))
143 def verify_vrf(self, vrf_id):
145 Check if the FIB table / VRF ID is configured.
147 :param int vrf_id: The FIB table / VRF ID to be verified.
148 :return: 1 if the FIB table / VRF ID is configured, otherwise return 0.
150 ip_fib_dump = self.vapi.ip_fib_dump()
152 for ip_fib_details in ip_fib_dump:
153 if ip_fib_details[2] == vrf_id:
156 self.logger.info("IPv4 VRF ID %d is not configured" % vrf_id)
159 self.logger.info("IPv4 VRF ID %d is configured" % vrf_id)
163 def build_ip_mask(proto='', src_ip='', dst_ip='',
164 src_port='', dst_port=''):
165 """Build IP ACL mask data with hexstring format
167 :param str proto: protocol number <0-ff>
168 :param str src_ip: source ip address <0-ffffffff>
169 :param str dst_ip: destination ip address <0-ffffffff>
170 :param str src_port: source port number <0-ffff>
171 :param str dst_port: destination port number <0-ffff>
174 return ('{:0>20}{:0>12}{:0>8}{:0>12}{:0>4}'.format(
175 proto, src_ip, dst_ip, src_port, dst_port)).rstrip('0')
178 def build_ip_match(proto='', src_ip='', dst_ip='',
179 src_port='', dst_port=''):
180 """Build IP ACL match data with hexstring format
182 :param str proto: protocol number with valid option "<0-ff>"
183 :param str src_ip: source ip address with format of "x.x.x.x"
184 :param str dst_ip: destination ip address with format of "x.x.x.x"
185 :param str src_port: source port number <0-ffff>
186 :param str dst_port: destination port number <0-ffff>
189 src_ip = socket.inet_aton(src_ip).encode('hex')
191 dst_ip = socket.inet_aton(dst_ip).encode('hex')
193 return ('{:0>20}{:0>12}{:0>8}{:0>12}{:0>4}'.format(
194 proto, src_ip, dst_ip, src_port, dst_port)).rstrip('0')
197 def build_mac_mask(dst_mac='', src_mac='', ether_type=''):
198 """Build MAC ACL mask data with hexstring format
200 :param str dst_mac: source MAC address <0-ffffffffffff>
201 :param str src_mac: destination MAC address <0-ffffffffffff>
202 :param str ether_type: ethernet type <0-ffff>
205 return ('{:0>12}{:0>12}{:0>4}'.format(dst_mac, src_mac,
206 ether_type)).rstrip('0')
209 def build_mac_match(dst_mac='', src_mac='', ether_type=''):
210 """Build MAC ACL match data with hexstring format
212 :param str dst_mac: source MAC address <x:x:x:x:x:x>
213 :param str src_mac: destination MAC address <x:x:x:x:x:x>
214 :param str ether_type: ethernet type <0-ffff>
217 dst_mac = dst_mac.replace(':', '')
219 src_mac = src_mac.replace(':', '')
221 return ('{:0>12}{:0>12}{:0>4}'.format(dst_mac, src_mac,
222 ether_type)).rstrip('0')
224 def create_classify_table(self, key, mask, data_offset=0, is_add=1):
225 """Create Classify Table
227 :param str key: key for classify table (ex, ACL name).
228 :param str mask: mask value for interested traffic.
229 :param int match_n_vectors:
230 :param int is_add: option to configure classify table.
231 - create(1) or delete(0)
233 r = self.vapi.classify_add_del_table(
235 binascii.unhexlify(mask),
236 match_n_vectors=(len(mask) - 1) // 32 + 1,
239 current_data_offset=data_offset)
240 self.assertIsNotNone(r, msg='No response msg for add_del_table')
241 self.acl_tbl_idx[key] = r.new_table_index
243 def create_classify_session(self, intf, table_index, match,
244 pbr_option=0, vrfid=0, is_add=1):
245 """Create Classify Session
247 :param VppInterface intf: Interface to apply classify session.
248 :param int table_index: table index to identify classify table.
249 :param str match: matched value for interested traffic.
250 :param int pbr_action: enable/disable PBR feature.
251 :param int vrfid: VRF id.
252 :param int is_add: option to configure classify session.
253 - create(1) or delete(0)
255 r = self.vapi.classify_add_del_session(
258 binascii.unhexlify(match),
262 self.assertIsNotNone(r, msg='No response msg for add_del_session')
264 def input_acl_set_interface(self, intf, table_index, is_add=1):
265 """Configure Input ACL interface
267 :param VppInterface intf: Interface to apply Input ACL feature.
268 :param int table_index: table index to identify classify table.
269 :param int is_add: option to configure classify session.
270 - enable(1) or disable(0)
272 r = self.vapi.input_acl_set_interface(
275 ip4_table_index=table_index)
276 self.assertIsNotNone(r, msg='No response msg for acl_set_interface')
278 def test_acl_ip(self):
281 Test scenario for basic IP ACL with source IP
282 - Create IPv4 stream for pg0 -> pg1 interface.
283 - Create ACL with source IP address.
284 - Send and verify received packets on pg1 interface.
287 # Basic ACL testing with source IP
288 pkts = self.create_stream(self.pg0, self.pg1, self.pg_if_packet_sizes)
289 self.pg0.add_stream(pkts)
291 self.create_classify_table('ip', self.build_ip_mask(src_ip='ffffffff'))
292 self.create_classify_session(
293 self.pg0, self.acl_tbl_idx.get('ip'),
294 self.build_ip_match(src_ip=self.pg0.remote_ip4))
295 self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('ip'))
297 self.pg_enable_capture(self.pg_interfaces)
300 pkts = self.pg1.get_capture(len(pkts))
301 self.verify_capture(self.pg1, pkts)
302 self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('ip'), 0)
303 self.pg0.assert_nothing_captured(remark="packets forwarded")
304 self.pg2.assert_nothing_captured(remark="packets forwarded")
305 self.pg3.assert_nothing_captured(remark="packets forwarded")
307 def test_acl_mac(self):
310 Test scenario for basic MAC ACL with source MAC
311 - Create IPv4 stream for pg0 -> pg2 interface.
312 - Create ACL with source MAC address.
313 - Send and verify received packets on pg2 interface.
316 # Basic ACL testing with source MAC
317 pkts = self.create_stream(self.pg0, self.pg2, self.pg_if_packet_sizes)
318 self.pg0.add_stream(pkts)
320 self.create_classify_table('mac',
321 self.build_mac_mask(src_mac='ffffffffffff'),
323 self.create_classify_session(
324 self.pg0, self.acl_tbl_idx.get('mac'),
325 self.build_mac_match(src_mac=self.pg0.remote_mac))
326 self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('mac'))
328 self.pg_enable_capture(self.pg_interfaces)
331 pkts = self.pg2.get_capture(len(pkts))
332 self.verify_capture(self.pg2, pkts)
333 self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('mac'), 0)
334 self.pg0.assert_nothing_captured(remark="packets forwarded")
335 self.pg1.assert_nothing_captured(remark="packets forwarded")
336 self.pg3.assert_nothing_captured(remark="packets forwarded")
338 def test_acl_pbr(self):
341 Test scenario for PBR with source IP
342 - Create IPv4 stream for pg0 -> pg3 interface.
343 - Configure PBR fib entry for packet forwarding.
344 - Send and verify received packets on pg3 interface.
347 # PBR testing with source IP
348 pkts = self.create_stream(self.pg0, self.pg3, self.pg_if_packet_sizes)
349 self.pg0.add_stream(pkts)
351 self.create_classify_table(
352 'pbr', self.build_ip_mask(
355 # this will create the VRF/table in which we will insert the route
356 self.create_classify_session(
357 self.pg0, self.acl_tbl_idx.get('pbr'),
358 self.build_ip_match(src_ip=self.pg0.remote_ip4),
359 pbr_option, self.pbr_vrfid)
360 self.assertTrue(self.verify_vrf(self.pbr_vrfid))
361 self.config_pbr_fib_entry(self.pg3)
362 self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('pbr'))
364 self.pg_enable_capture(self.pg_interfaces)
367 pkts = self.pg3.get_capture(len(pkts))
368 self.verify_capture(self.pg3, pkts)
369 self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('pbr'), 0)
370 self.pg0.assert_nothing_captured(remark="packets forwarded")
371 self.pg1.assert_nothing_captured(remark="packets forwarded")
372 self.pg2.assert_nothing_captured(remark="packets forwarded")
374 # remove the classify session and the route
375 self.config_pbr_fib_entry(self.pg3, is_add=0)
376 self.create_classify_session(
377 self.pg0, self.acl_tbl_idx.get('pbr'),
378 self.build_ip_match(src_ip=self.pg0.remote_ip4),
379 pbr_option, self.pbr_vrfid, is_add=0)
381 # and the table should be gone.
382 self.assertFalse(self.verify_vrf(self.pbr_vrfid))
384 if __name__ == '__main__':
385 unittest.main(testRunner=VppTestRunner)