lb: remove api boilerplate
[vpp.git] / test / test_classifier.py
1 #!/usr/bin/env python
2
3 import binascii
4 import socket
5 import unittest
6
7 from framework import VppTestCase, VppTestRunner
8
9 from scapy.packet import Raw
10 from scapy.layers.l2 import Ether
11 from scapy.layers.inet import IP, UDP, TCP
12 from util import ppp
13 from vpp_ip_route import VppIpRoute, VppRoutePath
14 from vpp_ip import INVALID_INDEX
15
16
17 class TestClassifier(VppTestCase):
18     """ Classifier Test Case """
19
20     @classmethod
21     def setUpClass(cls):
22         """
23         Perform standard class setup (defined by class method setUpClass in
24         class VppTestCase) before running the test case, set test case related
25         variables and configure VPP.
26         """
27         super(TestClassifier, cls).setUpClass()
28         cls.acl_active_table = ''
29
30     @classmethod
31     def tearDownClass(cls):
32         super(TestClassifier, cls).tearDownClass()
33
34     def setUp(self):
35         """
36         Perform test setup before test case.
37
38         **Config:**
39             - create 4 pg interfaces
40                 - untagged pg0/pg1/pg2 interface
41                     pg0 -------> pg1 (IP ACL)
42                            \
43                             ---> pg2 (MAC ACL))
44                              \
45                               -> pg3 (PBR)
46             - setup interfaces:
47                 - put it into UP state
48                 - set IPv4 addresses
49                 - resolve neighbor address using ARP
50
51         :ivar list interfaces: pg interfaces.
52         :ivar list pg_if_packet_sizes: packet sizes in test.
53         :ivar dict acl_tbl_idx: ACL table index.
54         :ivar int pbr_vrfid: VRF id for PBR test.
55         """
56         self.reset_packet_infos()
57         super(TestClassifier, self).setUp()
58
59         # create 4 pg interfaces
60         self.create_pg_interfaces(range(4))
61
62         # packet sizes to test
63         self.pg_if_packet_sizes = [64, 9018]
64
65         self.interfaces = list(self.pg_interfaces)
66
67         # ACL & PBR vars
68         self.acl_tbl_idx = {}
69         self.pbr_vrfid = 200
70
71         # setup all interfaces
72         for intf in self.interfaces:
73             intf.admin_up()
74             intf.config_ip4()
75             intf.resolve_arp()
76
77     def tearDown(self):
78         """Run standard test teardown and acl related log."""
79         if not self.vpp_dead:
80             if self.acl_active_table == 'ip_out':
81                 self.output_acl_set_interface(
82                     self.pg0, self.acl_tbl_idx.get(self.acl_active_table), 0)
83                 self.acl_active_table = ''
84             elif self.acl_active_table != '':
85                 self.input_acl_set_interface(
86                     self.pg0, self.acl_tbl_idx.get(self.acl_active_table), 0)
87                 self.acl_active_table = ''
88             for intf in self.interfaces:
89                 intf.unconfig_ip4()
90                 intf.admin_down()
91
92         super(TestClassifier, self).tearDown()
93
94     def show_commands_at_teardown(self):
95         self.logger.info(self.vapi.ppcli("show inacl type ip4"))
96         self.logger.info(self.vapi.ppcli("show outacl type ip4"))
97         self.logger.info(self.vapi.cli("show classify table verbose"))
98         self.logger.info(self.vapi.cli("show ip fib"))
99
100     def create_stream(self, src_if, dst_if, packet_sizes,
101                       proto_l=UDP(sport=1234, dport=5678)):
102         """Create input packet stream for defined interfaces.
103
104         :param VppInterface src_if: Source Interface for packet stream.
105         :param VppInterface dst_if: Destination Interface for packet stream.
106         :param list packet_sizes: packet size to test.
107         :param Scapy proto_l: Required IP protocol. Default protocol is UDP.
108         """
109         pkts = []
110
111         for size in packet_sizes:
112             info = self.create_packet_info(src_if, dst_if)
113             payload = self.info_to_payload(info)
114             p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
115                  IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4) /
116                  proto_l /
117                  Raw(payload))
118             info.data = p.copy()
119             self.extend_packet(p, size)
120             pkts.append(p)
121         return pkts
122
123     def verify_capture(self, dst_if, capture, proto_l=UDP):
124         """Verify captured input packet stream for defined interface.
125
126         :param VppInterface dst_if: Interface to verify captured packet stream.
127         :param list capture: Captured packet stream.
128         :param Scapy proto_l: Required IP protocol. Default protocol is UDP.
129         """
130         self.logger.info("Verifying capture on interface %s" % dst_if.name)
131         last_info = dict()
132         for i in self.interfaces:
133             last_info[i.sw_if_index] = None
134         dst_sw_if_index = dst_if.sw_if_index
135         for packet in capture:
136             try:
137                 ip_received = packet[IP]
138                 proto_received = packet[proto_l]
139                 payload_info = self.payload_to_info(packet[Raw])
140                 packet_index = payload_info.index
141                 self.assertEqual(payload_info.dst, dst_sw_if_index)
142                 self.logger.debug(
143                     "Got packet on port %s: src=%u (id=%u)" %
144                     (dst_if.name, payload_info.src, packet_index))
145                 next_info = self.get_next_packet_info_for_interface2(
146                     payload_info.src, dst_sw_if_index,
147                     last_info[payload_info.src])
148                 last_info[payload_info.src] = next_info
149                 self.assertTrue(next_info is not None)
150                 self.assertEqual(packet_index, next_info.index)
151                 saved_packet = next_info.data
152                 ip_saved = saved_packet[IP]
153                 proto_saved = saved_packet[proto_l]
154                 # Check standard fields
155                 self.assertEqual(ip_received.src, ip_saved.src)
156                 self.assertEqual(ip_received.dst, ip_saved.dst)
157                 self.assertEqual(proto_received.sport, proto_saved.sport)
158                 self.assertEqual(proto_received.dport, proto_saved.dport)
159             except:
160                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
161                 raise
162         for i in self.interfaces:
163             remaining_packet = self.get_next_packet_info_for_interface2(
164                 i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index])
165             self.assertTrue(remaining_packet is None,
166                             "Interface %s: Packet expected from interface %s "
167                             "didn't arrive" % (dst_if.name, i.name))
168
169     def verify_vrf(self, vrf_id):
170         """
171         Check if the FIB table / VRF ID is configured.
172
173         :param int vrf_id: The FIB table / VRF ID to be verified.
174         :return: 1 if the FIB table / VRF ID is configured, otherwise return 0.
175         """
176         ip_fib_dump = self.vapi.ip_route_dump(vrf_id, False)
177         vrf_count = len(ip_fib_dump)
178
179         if vrf_count == 0:
180             self.logger.info("IPv4 VRF ID %d is not configured" % vrf_id)
181             return 0
182         else:
183             self.logger.info("IPv4 VRF ID %d is configured" % vrf_id)
184             return 1
185
186     @staticmethod
187     def build_ip_mask(proto='', src_ip='', dst_ip='',
188                       src_port='', dst_port=''):
189         """Build IP ACL mask data with hexstring format.
190
191         :param str proto: protocol number <0-ff>
192         :param str src_ip: source ip address <0-ffffffff>
193         :param str dst_ip: destination ip address <0-ffffffff>
194         :param str src_port: source port number <0-ffff>
195         :param str dst_port: destination port number <0-ffff>
196         """
197
198         return ('{!s:0>20}{!s:0>12}{!s:0>8}{!s:0>4}{!s:0>4}'.format(
199             proto, src_ip, dst_ip, src_port, dst_port)).rstrip('0')
200
201     @staticmethod
202     def build_ip_match(proto=0, src_ip='', dst_ip='',
203                        src_port=0, dst_port=0):
204         """Build IP ACL match data with hexstring format.
205
206         :param int proto: protocol number with valid option "x"
207         :param str src_ip: source ip address with format of "x.x.x.x"
208         :param str dst_ip: destination ip address with format of "x.x.x.x"
209         :param int src_port: source port number "x"
210         :param int dst_port: destination port number "x"
211         """
212         if src_ip:
213             src_ip = binascii.hexlify(socket.inet_aton(src_ip))
214         if dst_ip:
215             dst_ip = binascii.hexlify(socket.inet_aton(dst_ip))
216
217         return ('{!s:0>20}{!s:0>12}{!s:0>8}{!s:0>4}{!s:0>4}'.format(
218             hex(proto)[2:], src_ip, dst_ip, hex(src_port)[2:],
219             hex(dst_port)[2:])).rstrip('0')
220
221     @staticmethod
222     def build_mac_mask(dst_mac='', src_mac='', ether_type=''):
223         """Build MAC ACL mask data with hexstring format.
224
225         :param str dst_mac: source MAC address <0-ffffffffffff>
226         :param str src_mac: destination MAC address <0-ffffffffffff>
227         :param str ether_type: ethernet type <0-ffff>
228         """
229
230         return ('{!s:0>12}{!s:0>12}{!s:0>4}'.format(
231             dst_mac, src_mac, ether_type)).rstrip('0')
232
233     @staticmethod
234     def build_mac_match(dst_mac='', src_mac='', ether_type=''):
235         """Build MAC ACL match data with hexstring format.
236
237         :param str dst_mac: source MAC address <x:x:x:x:x:x>
238         :param str src_mac: destination MAC address <x:x:x:x:x:x>
239         :param str ether_type: ethernet type <0-ffff>
240         """
241         if dst_mac:
242             dst_mac = dst_mac.replace(':', '')
243         if src_mac:
244             src_mac = src_mac.replace(':', '')
245
246         return ('{!s:0>12}{!s:0>12}{!s:0>4}'.format(
247             dst_mac, src_mac, ether_type)).rstrip('0')
248
249     def create_classify_table(self, key, mask, data_offset=0):
250         """Create Classify Table
251
252         :param str key: key for classify table (ex, ACL name).
253         :param str mask: mask value for interested traffic.
254         :param int data_offset:
255         """
256         r = self.vapi.classify_add_del_table(
257             is_add=1,
258             mask=binascii.unhexlify(mask),
259             match_n_vectors=(len(mask) - 1) // 32 + 1,
260             miss_next_index=0,
261             current_data_flag=1,
262             current_data_offset=data_offset)
263         self.assertIsNotNone(r, 'No response msg for add_del_table')
264         self.acl_tbl_idx[key] = r.new_table_index
265
266     def create_classify_session(self, table_index, match, pbr_option=0,
267                                 vrfid=0, is_add=1):
268         """Create Classify Session
269
270         :param int table_index: table index to identify classify table.
271         :param str match: matched value for interested traffic.
272         :param int pbr_option: enable/disable PBR feature.
273         :param int vrfid: VRF id.
274         :param int is_add: option to configure classify session.
275             - create(1) or delete(0)
276         """
277         r = self.vapi.classify_add_del_session(
278             is_add,
279             table_index,
280             binascii.unhexlify(match),
281             opaque_index=0,
282             action=pbr_option,
283             metadata=vrfid)
284         self.assertIsNotNone(r, 'No response msg for add_del_session')
285
286     def input_acl_set_interface(self, intf, table_index, is_add=1):
287         """Configure Input ACL interface
288
289         :param VppInterface intf: Interface to apply Input ACL feature.
290         :param int table_index: table index to identify classify table.
291         :param int is_add: option to configure classify session.
292             - enable(1) or disable(0)
293         """
294         r = self.vapi.input_acl_set_interface(
295             is_add,
296             intf.sw_if_index,
297             ip4_table_index=table_index)
298         self.assertIsNotNone(r, 'No response msg for acl_set_interface')
299
300     def output_acl_set_interface(self, intf, table_index, is_add=1):
301         """Configure Output ACL interface
302
303         :param VppInterface intf: Interface to apply Output ACL feature.
304         :param int table_index: table index to identify classify table.
305         :param int is_add: option to configure classify session.
306             - enable(1) or disable(0)
307         """
308         r = self.vapi.output_acl_set_interface(
309             is_add,
310             intf.sw_if_index,
311             ip4_table_index=table_index)
312         self.assertIsNotNone(r, 'No response msg for acl_set_interface')
313
314
315 # Tests split to different test case classes because of issue reported in
316 # ticket VPP-1336
317 class TestClassifierIP(TestClassifier):
318     """ Classifier IP Test Case """
319
320     @classmethod
321     def setUpClass(cls):
322         super(TestClassifierIP, cls).setUpClass()
323
324     @classmethod
325     def tearDownClass(cls):
326         super(TestClassifierIP, cls).tearDownClass()
327
328     def test_iacl_src_ip(self):
329         """ Source IP iACL test
330
331         Test scenario for basic IP ACL with source IP
332             - Create IPv4 stream for pg0 -> pg1 interface.
333             - Create iACL with source IP address.
334             - Send and verify received packets on pg1 interface.
335         """
336
337         # Basic iACL testing with source IP
338         pkts = self.create_stream(self.pg0, self.pg1, self.pg_if_packet_sizes)
339         self.pg0.add_stream(pkts)
340
341         key = 'ip_src'
342         self.create_classify_table(key, self.build_ip_mask(src_ip='ffffffff'))
343         self.create_classify_session(
344             self.acl_tbl_idx.get(key),
345             self.build_ip_match(src_ip=self.pg0.remote_ip4))
346         self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key))
347         self.acl_active_table = key
348
349         self.pg_enable_capture(self.pg_interfaces)
350         self.pg_start()
351
352         pkts = self.pg1.get_capture(len(pkts))
353         self.verify_capture(self.pg1, pkts)
354         self.pg0.assert_nothing_captured(remark="packets forwarded")
355         self.pg2.assert_nothing_captured(remark="packets forwarded")
356         self.pg3.assert_nothing_captured(remark="packets forwarded")
357
358     def test_iacl_dst_ip(self):
359         """ Destination IP iACL test
360
361         Test scenario for basic IP ACL with destination IP
362             - Create IPv4 stream for pg0 -> pg1 interface.
363             - Create iACL with destination IP address.
364             - Send and verify received packets on pg1 interface.
365         """
366
367         # Basic iACL testing with destination IP
368         pkts = self.create_stream(self.pg0, self.pg1, self.pg_if_packet_sizes)
369         self.pg0.add_stream(pkts)
370
371         key = 'ip_dst'
372         self.create_classify_table(key, self.build_ip_mask(dst_ip='ffffffff'))
373         self.create_classify_session(
374             self.acl_tbl_idx.get(key),
375             self.build_ip_match(dst_ip=self.pg1.remote_ip4))
376         self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key))
377         self.acl_active_table = key
378
379         self.pg_enable_capture(self.pg_interfaces)
380         self.pg_start()
381
382         pkts = self.pg1.get_capture(len(pkts))
383         self.verify_capture(self.pg1, pkts)
384         self.pg0.assert_nothing_captured(remark="packets forwarded")
385         self.pg2.assert_nothing_captured(remark="packets forwarded")
386         self.pg3.assert_nothing_captured(remark="packets forwarded")
387
388     def test_iacl_src_dst_ip(self):
389         """ Source and destination IP iACL test
390
391         Test scenario for basic IP ACL with source and destination IP
392             - Create IPv4 stream for pg0 -> pg1 interface.
393             - Create iACL with source and destination IP addresses.
394             - Send and verify received packets on pg1 interface.
395         """
396
397         # Basic iACL testing with source and destination IP
398         pkts = self.create_stream(self.pg0, self.pg1, self.pg_if_packet_sizes)
399         self.pg0.add_stream(pkts)
400
401         key = 'ip'
402         self.create_classify_table(
403             key, self.build_ip_mask(src_ip='ffffffff', dst_ip='ffffffff'))
404         self.create_classify_session(
405             self.acl_tbl_idx.get(key),
406             self.build_ip_match(src_ip=self.pg0.remote_ip4,
407                                 dst_ip=self.pg1.remote_ip4))
408         self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key))
409         self.acl_active_table = key
410
411         self.pg_enable_capture(self.pg_interfaces)
412         self.pg_start()
413
414         pkts = self.pg1.get_capture(len(pkts))
415         self.verify_capture(self.pg1, pkts)
416         self.pg0.assert_nothing_captured(remark="packets forwarded")
417         self.pg2.assert_nothing_captured(remark="packets forwarded")
418         self.pg3.assert_nothing_captured(remark="packets forwarded")
419
420
421 class TestClassifierUDP(TestClassifier):
422     """ Classifier UDP proto Test Case """
423
424     @classmethod
425     def setUpClass(cls):
426         super(TestClassifierUDP, cls).setUpClass()
427
428     @classmethod
429     def tearDownClass(cls):
430         super(TestClassifierUDP, cls).tearDownClass()
431
432     def test_iacl_proto_udp(self):
433         """ UDP protocol iACL test
434
435         Test scenario for basic protocol ACL with UDP protocol
436             - Create IPv4 stream for pg0 -> pg1 interface.
437             - Create iACL with UDP IP protocol.
438             - Send and verify received packets on pg1 interface.
439         """
440
441         # Basic iACL testing with UDP protocol
442         pkts = self.create_stream(self.pg0, self.pg1, self.pg_if_packet_sizes)
443         self.pg0.add_stream(pkts)
444
445         key = 'proto_udp'
446         self.create_classify_table(key, self.build_ip_mask(proto='ff'))
447         self.create_classify_session(
448             self.acl_tbl_idx.get(key),
449             self.build_ip_match(proto=socket.IPPROTO_UDP))
450         self.input_acl_set_interface(
451             self.pg0, self.acl_tbl_idx.get(key))
452         self.acl_active_table = key
453
454         self.pg_enable_capture(self.pg_interfaces)
455         self.pg_start()
456
457         pkts = self.pg1.get_capture(len(pkts))
458         self.verify_capture(self.pg1, pkts)
459         self.pg0.assert_nothing_captured(remark="packets forwarded")
460         self.pg2.assert_nothing_captured(remark="packets forwarded")
461         self.pg3.assert_nothing_captured(remark="packets forwarded")
462
463     def test_iacl_proto_udp_sport(self):
464         """ UDP source port iACL test
465
466         Test scenario for basic protocol ACL with UDP and sport
467             - Create IPv4 stream for pg0 -> pg1 interface.
468             - Create iACL with UDP IP protocol and defined sport.
469             - Send and verify received packets on pg1 interface.
470         """
471
472         # Basic iACL testing with UDP and sport
473         sport = 38
474         pkts = self.create_stream(self.pg0, self.pg1, self.pg_if_packet_sizes,
475                                   UDP(sport=sport, dport=5678))
476         self.pg0.add_stream(pkts)
477
478         key = 'proto_udp_sport'
479         self.create_classify_table(
480             key, self.build_ip_mask(proto='ff', src_port='ffff'))
481         self.create_classify_session(
482             self.acl_tbl_idx.get(key),
483             self.build_ip_match(proto=socket.IPPROTO_UDP, src_port=sport))
484         self.input_acl_set_interface(
485             self.pg0, self.acl_tbl_idx.get(key))
486         self.acl_active_table = key
487
488         self.pg_enable_capture(self.pg_interfaces)
489         self.pg_start()
490
491         pkts = self.pg1.get_capture(len(pkts))
492         self.verify_capture(self.pg1, pkts)
493         self.pg0.assert_nothing_captured(remark="packets forwarded")
494         self.pg2.assert_nothing_captured(remark="packets forwarded")
495         self.pg3.assert_nothing_captured(remark="packets forwarded")
496
497     def test_iacl_proto_udp_dport(self):
498         """ UDP destination port iACL test
499
500         Test scenario for basic protocol ACL with UDP and dport
501             - Create IPv4 stream for pg0 -> pg1 interface.
502             - Create iACL with UDP IP protocol and defined dport.
503             - Send and verify received packets on pg1 interface.
504         """
505
506         # Basic iACL testing with UDP and dport
507         dport = 427
508         pkts = self.create_stream(self.pg0, self.pg1, self.pg_if_packet_sizes,
509                                   UDP(sport=1234, dport=dport))
510         self.pg0.add_stream(pkts)
511
512         key = 'proto_udp_dport'
513         self.create_classify_table(
514             key, self.build_ip_mask(proto='ff', dst_port='ffff'))
515         self.create_classify_session(
516             self.acl_tbl_idx.get(key),
517             self.build_ip_match(proto=socket.IPPROTO_UDP, dst_port=dport))
518         self.input_acl_set_interface(
519             self.pg0, self.acl_tbl_idx.get(key))
520         self.acl_active_table = key
521
522         self.pg_enable_capture(self.pg_interfaces)
523         self.pg_start()
524
525         pkts = self.pg1.get_capture(len(pkts))
526         self.verify_capture(self.pg1, pkts)
527         self.pg0.assert_nothing_captured(remark="packets forwarded")
528         self.pg2.assert_nothing_captured(remark="packets forwarded")
529         self.pg3.assert_nothing_captured(remark="packets forwarded")
530
531     def test_iacl_proto_udp_sport_dport(self):
532         """ UDP source and destination ports iACL test
533
534         Test scenario for basic protocol ACL with UDP and sport and dport
535             - Create IPv4 stream for pg0 -> pg1 interface.
536             - Create iACL with UDP IP protocol and defined sport and dport.
537             - Send and verify received packets on pg1 interface.
538         """
539
540         # Basic iACL testing with UDP and sport and dport
541         sport = 13720
542         dport = 9080
543         pkts = self.create_stream(self.pg0, self.pg1, self.pg_if_packet_sizes,
544                                   UDP(sport=sport, dport=dport))
545         self.pg0.add_stream(pkts)
546
547         key = 'proto_udp_ports'
548         self.create_classify_table(
549             key,
550             self.build_ip_mask(proto='ff', src_port='ffff', dst_port='ffff'))
551         self.create_classify_session(
552             self.acl_tbl_idx.get(key),
553             self.build_ip_match(proto=socket.IPPROTO_UDP, src_port=sport,
554                                 dst_port=dport))
555         self.input_acl_set_interface(
556             self.pg0, self.acl_tbl_idx.get(key))
557         self.acl_active_table = key
558
559         self.pg_enable_capture(self.pg_interfaces)
560         self.pg_start()
561
562         pkts = self.pg1.get_capture(len(pkts))
563         self.verify_capture(self.pg1, pkts)
564         self.pg0.assert_nothing_captured(remark="packets forwarded")
565         self.pg2.assert_nothing_captured(remark="packets forwarded")
566         self.pg3.assert_nothing_captured(remark="packets forwarded")
567
568
569 class TestClassifierTCP(TestClassifier):
570     """ Classifier TCP proto Test Case """
571
572     @classmethod
573     def setUpClass(cls):
574         super(TestClassifierTCP, cls).setUpClass()
575
576     @classmethod
577     def tearDownClass(cls):
578         super(TestClassifierTCP, cls).tearDownClass()
579
580     def test_iacl_proto_tcp(self):
581         """ TCP protocol iACL test
582
583         Test scenario for basic protocol ACL with TCP protocol
584             - Create IPv4 stream for pg0 -> pg1 interface.
585             - Create iACL with TCP IP protocol.
586             - Send and verify received packets on pg1 interface.
587         """
588
589         # Basic iACL testing with TCP protocol
590         pkts = self.create_stream(self.pg0, self.pg1, self.pg_if_packet_sizes,
591                                   TCP(sport=1234, dport=5678))
592         self.pg0.add_stream(pkts)
593
594         key = 'proto_tcp'
595         self.create_classify_table(key, self.build_ip_mask(proto='ff'))
596         self.create_classify_session(
597             self.acl_tbl_idx.get(key),
598             self.build_ip_match(proto=socket.IPPROTO_TCP))
599         self.input_acl_set_interface(
600             self.pg0, self.acl_tbl_idx.get(key))
601         self.acl_active_table = key
602
603         self.pg_enable_capture(self.pg_interfaces)
604         self.pg_start()
605
606         pkts = self.pg1.get_capture(len(pkts))
607         self.verify_capture(self.pg1, pkts, TCP)
608         self.pg0.assert_nothing_captured(remark="packets forwarded")
609         self.pg2.assert_nothing_captured(remark="packets forwarded")
610         self.pg3.assert_nothing_captured(remark="packets forwarded")
611
612     def test_iacl_proto_tcp_sport(self):
613         """ TCP source port iACL test
614
615         Test scenario for basic protocol ACL with TCP and sport
616             - Create IPv4 stream for pg0 -> pg1 interface.
617             - Create iACL with TCP IP protocol and defined sport.
618             - Send and verify received packets on pg1 interface.
619         """
620
621         # Basic iACL testing with TCP and sport
622         sport = 38
623         pkts = self.create_stream(self.pg0, self.pg1, self.pg_if_packet_sizes,
624                                   TCP(sport=sport, dport=5678))
625         self.pg0.add_stream(pkts)
626
627         key = 'proto_tcp_sport'
628         self.create_classify_table(
629             key, self.build_ip_mask(proto='ff', src_port='ffff'))
630         self.create_classify_session(
631             self.acl_tbl_idx.get(key),
632             self.build_ip_match(proto=socket.IPPROTO_TCP, src_port=sport))
633         self.input_acl_set_interface(
634             self.pg0, self.acl_tbl_idx.get(key))
635         self.acl_active_table = key
636
637         self.pg_enable_capture(self.pg_interfaces)
638         self.pg_start()
639
640         pkts = self.pg1.get_capture(len(pkts))
641         self.verify_capture(self.pg1, pkts, TCP)
642         self.pg0.assert_nothing_captured(remark="packets forwarded")
643         self.pg2.assert_nothing_captured(remark="packets forwarded")
644         self.pg3.assert_nothing_captured(remark="packets forwarded")
645
646     def test_iacl_proto_tcp_dport(self):
647         """ TCP destination port iACL test
648
649         Test scenario for basic protocol ACL with TCP and dport
650             - Create IPv4 stream for pg0 -> pg1 interface.
651             - Create iACL with TCP IP protocol and defined dport.
652             - Send and verify received packets on pg1 interface.
653         """
654
655         # Basic iACL testing with TCP and dport
656         dport = 427
657         pkts = self.create_stream(self.pg0, self.pg1, self.pg_if_packet_sizes,
658                                   TCP(sport=1234, dport=dport))
659         self.pg0.add_stream(pkts)
660
661         key = 'proto_tcp_sport'
662         self.create_classify_table(
663             key, self.build_ip_mask(proto='ff', dst_port='ffff'))
664         self.create_classify_session(
665             self.acl_tbl_idx.get(key),
666             self.build_ip_match(proto=socket.IPPROTO_TCP, dst_port=dport))
667         self.input_acl_set_interface(
668             self.pg0, self.acl_tbl_idx.get(key))
669         self.acl_active_table = key
670
671         self.pg_enable_capture(self.pg_interfaces)
672         self.pg_start()
673
674         pkts = self.pg1.get_capture(len(pkts))
675         self.verify_capture(self.pg1, pkts, TCP)
676         self.pg0.assert_nothing_captured(remark="packets forwarded")
677         self.pg2.assert_nothing_captured(remark="packets forwarded")
678         self.pg3.assert_nothing_captured(remark="packets forwarded")
679
680     def test_iacl_proto_tcp_sport_dport(self):
681         """ TCP source and destination ports iACL test
682
683         Test scenario for basic protocol ACL with TCP and sport and dport
684             - Create IPv4 stream for pg0 -> pg1 interface.
685             - Create iACL with TCP IP protocol and defined sport and dport.
686             - Send and verify received packets on pg1 interface.
687         """
688
689         # Basic iACL testing with TCP and sport and dport
690         sport = 13720
691         dport = 9080
692         pkts = self.create_stream(self.pg0, self.pg1, self.pg_if_packet_sizes,
693                                   TCP(sport=sport, dport=dport))
694         self.pg0.add_stream(pkts)
695
696         key = 'proto_tcp_ports'
697         self.create_classify_table(
698             key,
699             self.build_ip_mask(proto='ff', src_port='ffff', dst_port='ffff'))
700         self.create_classify_session(
701             self.acl_tbl_idx.get(key),
702             self.build_ip_match(proto=socket.IPPROTO_TCP, src_port=sport,
703                                 dst_port=dport))
704         self.input_acl_set_interface(
705             self.pg0, self.acl_tbl_idx.get(key))
706         self.acl_active_table = key
707
708         self.pg_enable_capture(self.pg_interfaces)
709         self.pg_start()
710
711         pkts = self.pg1.get_capture(len(pkts))
712         self.verify_capture(self.pg1, pkts, TCP)
713         self.pg0.assert_nothing_captured(remark="packets forwarded")
714         self.pg2.assert_nothing_captured(remark="packets forwarded")
715         self.pg3.assert_nothing_captured(remark="packets forwarded")
716
717
718 class TestClassifierIPOut(TestClassifier):
719     """ Classifier output IP Test Case """
720
721     @classmethod
722     def setUpClass(cls):
723         super(TestClassifierIPOut, cls).setUpClass()
724
725     @classmethod
726     def tearDownClass(cls):
727         super(TestClassifierIPOut, cls).tearDownClass()
728
729     def test_acl_ip_out(self):
730         """ Output IP ACL test
731
732         Test scenario for basic IP ACL with source IP
733             - Create IPv4 stream for pg1 -> pg0 interface.
734             - Create ACL with source IP address.
735             - Send and verify received packets on pg0 interface.
736         """
737
738         # Basic oACL testing with source IP
739         pkts = self.create_stream(self.pg1, self.pg0, self.pg_if_packet_sizes)
740         self.pg1.add_stream(pkts)
741
742         key = 'ip_out'
743         self.create_classify_table(
744             key, self.build_ip_mask(src_ip='ffffffff'), data_offset=0)
745         self.create_classify_session(
746             self.acl_tbl_idx.get(key),
747             self.build_ip_match(src_ip=self.pg1.remote_ip4))
748         self.output_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key))
749         self.acl_active_table = key
750
751         self.pg_enable_capture(self.pg_interfaces)
752         self.pg_start()
753
754         pkts = self.pg0.get_capture(len(pkts))
755         self.verify_capture(self.pg0, pkts)
756         self.pg1.assert_nothing_captured(remark="packets forwarded")
757         self.pg2.assert_nothing_captured(remark="packets forwarded")
758         self.pg3.assert_nothing_captured(remark="packets forwarded")
759
760
761 class TestClassifierMAC(TestClassifier):
762     """ Classifier MAC Test Case """
763
764     @classmethod
765     def setUpClass(cls):
766         super(TestClassifierMAC, cls).setUpClass()
767
768     @classmethod
769     def tearDownClass(cls):
770         super(TestClassifierMAC, cls).tearDownClass()
771
772     def test_acl_mac(self):
773         """ MAC ACL test
774
775         Test scenario for basic MAC ACL with source MAC
776             - Create IPv4 stream for pg0 -> pg2 interface.
777             - Create ACL with source MAC address.
778             - Send and verify received packets on pg2 interface.
779         """
780
781         # Basic iACL testing with source MAC
782         pkts = self.create_stream(self.pg0, self.pg2, self.pg_if_packet_sizes)
783         self.pg0.add_stream(pkts)
784
785         key = 'mac'
786         self.create_classify_table(
787             key, self.build_mac_mask(src_mac='ffffffffffff'), data_offset=-14)
788         self.create_classify_session(
789             self.acl_tbl_idx.get(key),
790             self.build_mac_match(src_mac=self.pg0.remote_mac))
791         self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key))
792         self.acl_active_table = key
793
794         self.pg_enable_capture(self.pg_interfaces)
795         self.pg_start()
796
797         pkts = self.pg2.get_capture(len(pkts))
798         self.verify_capture(self.pg2, pkts)
799         self.pg0.assert_nothing_captured(remark="packets forwarded")
800         self.pg1.assert_nothing_captured(remark="packets forwarded")
801         self.pg3.assert_nothing_captured(remark="packets forwarded")
802
803
804 class TestClassifierPBR(TestClassifier):
805     """ Classifier PBR Test Case """
806
807     @classmethod
808     def setUpClass(cls):
809         super(TestClassifierPBR, cls).setUpClass()
810
811     @classmethod
812     def tearDownClass(cls):
813         super(TestClassifierPBR, cls).tearDownClass()
814
815     def test_acl_pbr(self):
816         """ IP PBR test
817
818         Test scenario for PBR with source IP
819             - Create IPv4 stream for pg0 -> pg3 interface.
820             - Configure PBR fib entry for packet forwarding.
821             - Send and verify received packets on pg3 interface.
822         """
823
824         # PBR testing with source IP
825         pkts = self.create_stream(self.pg0, self.pg3, self.pg_if_packet_sizes)
826         self.pg0.add_stream(pkts)
827
828         key = 'pbr'
829         self.create_classify_table(key, self.build_ip_mask(src_ip='ffffffff'))
830         pbr_option = 1
831         # this will create the VRF/table in which we will insert the route
832         self.create_classify_session(
833             self.acl_tbl_idx.get(key),
834             self.build_ip_match(src_ip=self.pg0.remote_ip4),
835             pbr_option, self.pbr_vrfid)
836         self.assertTrue(self.verify_vrf(self.pbr_vrfid))
837         r = VppIpRoute(self, self.pg3.local_ip4, 24,
838                        [VppRoutePath(self.pg3.remote_ip4,
839                                      INVALID_INDEX)],
840                        table_id=self.pbr_vrfid)
841         r.add_vpp_config()
842
843         self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key))
844
845         self.pg_enable_capture(self.pg_interfaces)
846         self.pg_start()
847
848         pkts = self.pg3.get_capture(len(pkts))
849         self.verify_capture(self.pg3, pkts)
850         self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key), 0)
851         self.pg0.assert_nothing_captured(remark="packets forwarded")
852         self.pg1.assert_nothing_captured(remark="packets forwarded")
853         self.pg2.assert_nothing_captured(remark="packets forwarded")
854
855         # remove the classify session and the route
856         r.remove_vpp_config()
857         self.create_classify_session(
858             self.acl_tbl_idx.get(key),
859             self.build_ip_match(src_ip=self.pg0.remote_ip4),
860             pbr_option, self.pbr_vrfid, is_add=0)
861
862         # and the table should be gone.
863         self.assertFalse(self.verify_vrf(self.pbr_vrfid))
864
865 if __name__ == '__main__':
866     unittest.main(testRunner=VppTestRunner)