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
14 class TestClassifier(VppTestCase):
15 """ Classifier Test Case """
19 Perform test setup before test case.
22 - create 4 pg interfaces
23 - untagged pg0/pg1/pg2 interface
24 pg0 -------> pg1 (IP ACL)
30 - put it into UP state
32 - resolve neighbor address using ARP
34 :ivar list interfaces: pg interfaces.
35 :ivar list pg_if_packet_sizes: packet sizes in test.
36 :ivar dict acl_tbl_idx: ACL table index.
37 :ivar int pbr_vrfid: VRF id for PBR test.
39 super(TestClassifier, self).setUp()
41 # create 4 pg interfaces
42 self.create_pg_interfaces(range(4))
44 # packet sizes to test
45 self.pg_if_packet_sizes = [64, 9018]
47 self.interfaces = list(self.pg_interfaces)
53 # setup all interfaces
54 for intf in self.interfaces:
60 """Run standard test teardown and acl related log."""
61 super(TestClassifier, self).tearDown()
63 self.logger.info(self.vapi.cli("show classify table verbose"))
64 self.logger.info(self.vapi.cli("show ip fib"))
66 def config_pbr_fib_entry(self, intf):
67 """Configure fib entry to route traffic toward PBR VRF table
69 :param VppInterface intf: destination interface to be routed for PBR.
73 self.vapi.ip_add_del_route(intf.local_ip4n,
76 table_id=self.pbr_vrfid)
78 def create_stream(self, src_if, dst_if, packet_sizes):
79 """Create input packet stream for defined interfaces.
81 :param VppInterface src_if: Source Interface for packet stream.
82 :param VppInterface dst_if: Destination Interface for packet stream.
83 :param list packet_sizes: packet size to test.
86 for size in packet_sizes:
87 info = self.create_packet_info(src_if.sw_if_index,
89 payload = self.info_to_payload(info)
90 p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
91 IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4) /
92 UDP(sport=1234, dport=5678) /
95 self.extend_packet(p, size)
99 def verify_capture(self, dst_if, capture):
100 """Verify captured input packet stream for defined interface.
102 :param VppInterface dst_if: Interface to verify captured packet stream.
103 :param list capture: Captured packet stream.
105 self.logger.info("Verifying capture on interface %s" % dst_if.name)
107 for i in self.interfaces:
108 last_info[i.sw_if_index] = None
109 dst_sw_if_index = dst_if.sw_if_index
110 for packet in capture:
114 payload_info = self.payload_to_info(str(packet[Raw]))
115 packet_index = payload_info.index
116 self.assertEqual(payload_info.dst, dst_sw_if_index)
117 self.logger.debug("Got packet on port %s: src=%u (id=%u)" %
118 (dst_if.name, payload_info.src, packet_index))
119 next_info = self.get_next_packet_info_for_interface2(
120 payload_info.src, dst_sw_if_index,
121 last_info[payload_info.src])
122 last_info[payload_info.src] = next_info
123 self.assertTrue(next_info is not None)
124 self.assertEqual(packet_index, next_info.index)
125 saved_packet = next_info.data
126 # Check standard fields
127 self.assertEqual(ip.src, saved_packet[IP].src)
128 self.assertEqual(ip.dst, saved_packet[IP].dst)
129 self.assertEqual(udp.sport, saved_packet[UDP].sport)
130 self.assertEqual(udp.dport, saved_packet[UDP].dport)
132 self.logger.error(ppp("Unexpected or invalid packet:", packet))
134 for i in self.interfaces:
135 remaining_packet = self.get_next_packet_info_for_interface2(
136 i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index])
137 self.assertTrue(remaining_packet is None,
138 "Interface %s: Packet expected from interface %s "
139 "didn't arrive" % (dst_if.name, i.name))
142 def build_ip_mask(proto='', src_ip='', dst_ip='',
143 src_port='', dst_port=''):
144 """Build IP ACL mask data with hexstring format
146 :param str proto: protocol number <0-ff>
147 :param str src_ip: source ip address <0-ffffffff>
148 :param str dst_ip: destination ip address <0-ffffffff>
149 :param str src_port: source port number <0-ffff>
150 :param str dst_port: destination port number <0-ffff>
153 return ('{:0>20}{:0>12}{:0>8}{:0>12}{:0>4}'.format(proto, src_ip,
154 dst_ip, src_port, dst_port)).rstrip('0')
157 def build_ip_match(proto='', src_ip='', dst_ip='',
158 src_port='', dst_port=''):
159 """Build IP ACL match data with hexstring format
161 :param str proto: protocol number with valid option "<0-ff>"
162 :param str src_ip: source ip address with format of "x.x.x.x"
163 :param str dst_ip: destination ip address with format of "x.x.x.x"
164 :param str src_port: source port number <0-ffff>
165 :param str dst_port: destination port number <0-ffff>
167 if src_ip: src_ip = socket.inet_aton(src_ip).encode('hex')
168 if dst_ip: dst_ip = socket.inet_aton(dst_ip).encode('hex')
170 return ('{:0>20}{:0>12}{:0>8}{:0>12}{:0>4}'.format(proto, src_ip,
171 dst_ip, src_port, dst_port)).rstrip('0')
174 def build_mac_mask(dst_mac='', src_mac='', ether_type=''):
175 """Build MAC ACL mask data with hexstring format
177 :param str dst_mac: source MAC address <0-ffffffffffff>
178 :param str src_mac: destination MAC address <0-ffffffffffff>
179 :param str ether_type: ethernet type <0-ffff>
182 return ('{:0>12}{:0>12}{:0>4}'.format(dst_mac, src_mac,
183 ether_type)).rstrip('0')
186 def build_mac_match(dst_mac='', src_mac='', ether_type=''):
187 """Build MAC ACL match data with hexstring format
189 :param str dst_mac: source MAC address <x:x:x:x:x:x>
190 :param str src_mac: destination MAC address <x:x:x:x:x:x>
191 :param str ether_type: ethernet type <0-ffff>
193 if dst_mac: dst_mac = dst_mac.replace(':', '')
194 if src_mac: src_mac = src_mac.replace(':', '')
196 return ('{:0>12}{:0>12}{:0>4}'.format(dst_mac, src_mac,
197 ether_type)).rstrip('0')
199 def create_classify_table(self, key, mask, data_offset=0, is_add=1):
200 """Create Classify Table
202 :param str key: key for classify table (ex, ACL name).
203 :param str mask: mask value for interested traffic.
204 :param int match_n_vectors:
205 :param int is_add: option to configure classify table.
206 - create(1) or delete(0)
208 r = self.vapi.classify_add_del_table(
210 binascii.unhexlify(mask),
211 match_n_vectors=(len(mask)-1)//32 + 1,
214 current_data_offset=data_offset)
215 self.assertIsNotNone(r, msg='No response msg for add_del_table')
216 self.acl_tbl_idx[key] = r.new_table_index
218 def create_classify_session(self, intf, table_index, match,
219 pbr_option=0, vrfid=0, is_add=1):
220 """Create Classify Session
222 :param VppInterface intf: Interface to apply classify session.
223 :param int table_index: table index to identify classify table.
224 :param str match: matched value for interested traffic.
225 :param int pbr_action: enable/disable PBR feature.
226 :param int vrfid: VRF id.
227 :param int is_add: option to configure classify session.
228 - create(1) or delete(0)
230 r = self.vapi.classify_add_del_session(
233 binascii.unhexlify(match),
237 self.assertIsNotNone(r, msg='No response msg for add_del_session')
239 def input_acl_set_interface(self, intf, table_index, is_add=1):
240 """Configure Input ACL interface
242 :param VppInterface intf: Interface to apply Input ACL feature.
243 :param int table_index: table index to identify classify table.
244 :param int is_add: option to configure classify session.
245 - enable(1) or disable(0)
247 r = self.vapi.input_acl_set_interface(
250 ip4_table_index=table_index)
251 self.assertIsNotNone(r, msg='No response msg for acl_set_interface')
253 def test_acl_ip(self):
256 Test scenario for basic IP ACL with source IP
257 - Create IPv4 stream for pg0 -> pg1 interface.
258 - Create ACL with source IP address.
259 - Send and verify received packets on pg1 interface.
262 # Basic ACL testing with source IP
263 pkts = self.create_stream(self.pg0, self.pg1, self.pg_if_packet_sizes)
264 self.pg0.add_stream(pkts)
266 self.create_classify_table('ip', self.build_ip_mask(src_ip='ffffffff'))
267 self.create_classify_session(self.pg0, self.acl_tbl_idx.get('ip'),
268 self.build_ip_match(src_ip=self.pg0.remote_ip4))
269 self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('ip'))
271 self.pg_enable_capture(self.pg_interfaces)
274 pkts = self.pg1.get_capture()
275 self.verify_capture(self.pg1, pkts)
276 self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('ip'), 0)
277 self.pg0.assert_nothing_captured(remark="packets forwarded")
278 self.pg2.assert_nothing_captured(remark="packets forwarded")
279 self.pg3.assert_nothing_captured(remark="packets forwarded")
281 def test_acl_mac(self):
284 Test scenario for basic MAC ACL with source MAC
285 - Create IPv4 stream for pg0 -> pg2 interface.
286 - Create ACL with source MAC address.
287 - Send and verify received packets on pg2 interface.
290 # Basic ACL testing with source MAC
291 pkts = self.create_stream(self.pg0, self.pg2, self.pg_if_packet_sizes)
292 self.pg0.add_stream(pkts)
294 self.create_classify_table('mac',
295 self.build_mac_mask(src_mac='ffffffffffff'), data_offset=-14)
296 self.create_classify_session(self.pg0, self.acl_tbl_idx.get('mac'),
297 self.build_mac_match(src_mac=self.pg0.remote_mac))
298 self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('mac'))
300 self.pg_enable_capture(self.pg_interfaces)
303 pkts = self.pg2.get_capture()
304 self.verify_capture(self.pg2, pkts)
305 self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('mac'), 0)
306 self.pg0.assert_nothing_captured(remark="packets forwarded")
307 self.pg1.assert_nothing_captured(remark="packets forwarded")
308 self.pg3.assert_nothing_captured(remark="packets forwarded")
310 def test_acl_pbr(self):
313 Test scenario for PBR with source IP
314 - Create IPv4 stream for pg0 -> pg3 interface.
315 - Configure PBR fib entry for packet forwarding.
316 - Send and verify received packets on pg3 interface.
319 # PBR testing with source IP
320 pkts = self.create_stream(self.pg0, self.pg3, self.pg_if_packet_sizes)
321 self.pg0.add_stream(pkts)
323 self.create_classify_table('pbr', self.build_ip_mask(src_ip='ffffffff'))
325 self.create_classify_session(self.pg0, self.acl_tbl_idx.get('pbr'),
326 self.build_ip_match(src_ip=self.pg0.remote_ip4),
327 pbr_option, self.pbr_vrfid)
328 self.config_pbr_fib_entry(self.pg3)
329 self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('pbr'))
331 self.pg_enable_capture(self.pg_interfaces)
334 pkts = self.pg3.get_capture()
335 self.verify_capture(self.pg3, pkts)
336 self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('pbr'), 0)
337 self.pg0.assert_nothing_captured(remark="packets forwarded")
338 self.pg1.assert_nothing_captured(remark="packets forwarded")
339 self.pg2.assert_nothing_captured(remark="packets forwarded")
342 if __name__ == '__main__':
343 unittest.main(testRunner=VppTestRunner)