2 """ Classifier-based L2 ACL Test Case HLD:
11 from scapy.packet import Raw
12 from scapy.data import ETH_P_IP
13 from scapy.layers.l2 import Ether
14 from scapy.layers.inet import IP, TCP, UDP, ICMP
15 from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest
16 from scapy.layers.inet6 import IPv6ExtHdrFragment
17 from framework import VppTestCase, VppTestRunner
18 from util import Host, ppp
21 class TestClassifyAcl(VppTestCase):
22 """ Classifier-based L2 input and output ACL Test Case """
38 proto = [[6, 17], [1, 58]]
39 proto_map = {1: 'ICMP', 58: 'ICMPv6EchoRequest', 6: 'TCP', 17: 'UDP'}
51 udp_sport_to = udp_sport_from + 5
52 udp_dport_from = 20000
53 udp_dport_to = udp_dport_from + 5000
55 tcp_sport_to = tcp_sport_from + 5
56 tcp_dport_from = 40000
57 tcp_dport_to = tcp_dport_from + 5000
60 udp_sport_to_2 = udp_sport_from_2 + 5
61 udp_dport_from_2 = 30000
62 udp_dport_to_2 = udp_dport_from_2 + 5000
63 tcp_sport_from_2 = 130
64 tcp_sport_to_2 = tcp_sport_from_2 + 5
65 tcp_dport_from_2 = 20000
66 tcp_dport_to_2 = tcp_dport_from_2 + 5000
68 icmp4_type = 8 # echo request
70 icmp6_type = 128 # echo request
86 Perform standard class setup (defined by class method setUpClass in
87 class VppTestCase) before running the test case, set test case related
88 variables and configure VPP.
90 super(TestClassifyAcl, cls).setUpClass()
93 # Create 2 pg interfaces
94 cls.create_pg_interfaces(range(2))
96 # Packet flows mapping pg0 -> pg1, pg2 etc.
98 cls.flows[cls.pg0] = [cls.pg1]
101 cls.pg_if_packet_sizes = [64, 512, 1518, 9018]
103 # Create BD with MAC learning and unknown unicast flooding disabled
104 # and put interfaces to this BD
105 cls.vapi.bridge_domain_add_del(bd_id=cls.bd_id, uu_flood=1,
107 for pg_if in cls.pg_interfaces:
108 cls.vapi.sw_interface_set_l2_bridge(pg_if.sw_if_index,
111 # Set up all interfaces
112 for i in cls.pg_interfaces:
115 # Mapping between packet-generator index and lists of test hosts
116 cls.hosts_by_pg_idx = dict()
117 for pg_if in cls.pg_interfaces:
118 cls.hosts_by_pg_idx[pg_if.sw_if_index] = []
120 # Create list of deleted hosts
121 cls.deleted_hosts_by_pg_idx = dict()
122 for pg_if in cls.pg_interfaces:
123 cls.deleted_hosts_by_pg_idx[pg_if.sw_if_index] = []
125 # warm-up the mac address tables
128 # Holder of the active classify table key
129 cls.acl_active_table = ''
132 super(TestClassifyAcl, cls).tearDownClass()
136 super(TestClassifyAcl, self).setUp()
138 self.acl_tbl_idx = {}
139 self.reset_packet_infos()
143 Show various debug prints after each test.
145 if not self.vpp_dead:
146 self.logger.info(self.vapi.ppcli("show inacl type l2"))
147 self.logger.info(self.vapi.ppcli("show outacl type l2"))
148 self.logger.info(self.vapi.ppcli("show classify tables verbose"))
149 self.logger.info(self.vapi.ppcli("show bridge-domain %s detail"
151 if self.acl_active_table == 'mac_inout':
152 self.output_acl_set_interface(
153 self.pg1, self.acl_tbl_idx.get(self.acl_active_table), 0)
154 self.input_acl_set_interface(
155 self.pg0, self.acl_tbl_idx.get(self.acl_active_table), 0)
156 self.acl_active_table = ''
157 elif self.acl_active_table == 'mac_out':
158 self.output_acl_set_interface(
159 self.pg1, self.acl_tbl_idx.get(self.acl_active_table), 0)
160 self.acl_active_table = ''
161 elif self.acl_active_table == 'mac_in':
162 self.input_acl_set_interface(
163 self.pg0, self.acl_tbl_idx.get(self.acl_active_table), 0)
164 self.acl_active_table = ''
166 super(TestClassifyAcl, self).tearDown()
169 def build_mac_mask(dst_mac='', src_mac='', ether_type=''):
170 """Build MAC ACL mask data with hexstring format
172 :param str dst_mac: source MAC address <0-ffffffffffff>
173 :param str src_mac: destination MAC address <0-ffffffffffff>
174 :param str ether_type: ethernet type <0-ffff>
177 return ('{:0>12}{:0>12}{:0>4}'.format(dst_mac, src_mac,
178 ether_type)).rstrip('0')
181 def build_mac_match(dst_mac='', src_mac='', ether_type=''):
182 """Build MAC ACL match data with hexstring format
184 :param str dst_mac: source MAC address <x:x:x:x:x:x>
185 :param str src_mac: destination MAC address <x:x:x:x:x:x>
186 :param str ether_type: ethernet type <0-ffff>
189 dst_mac = dst_mac.replace(':', '')
191 src_mac = src_mac.replace(':', '')
193 return ('{:0>12}{:0>12}{:0>4}'.format(dst_mac, src_mac,
194 ether_type)).rstrip('0')
196 def create_classify_table(self, key, mask, data_offset=0, is_add=1):
197 """Create Classify Table
199 :param str key: key for classify table (ex, ACL name).
200 :param str mask: mask value for interested traffic.
201 :param int match_n_vectors:
202 :param int is_add: option to configure classify table.
203 - create(1) or delete(0)
205 r = self.vapi.classify_add_del_table(
207 binascii.unhexlify(mask),
208 match_n_vectors=(len(mask) - 1) // 32 + 1,
211 current_data_offset=data_offset)
212 self.assertIsNotNone(r, msg='No response msg for add_del_table')
213 self.acl_tbl_idx[key] = r.new_table_index
215 def create_classify_session(self, intf, table_index, match,
216 hit_next_index=0xffffffff, is_add=1):
217 """Create Classify Session
219 :param VppInterface intf: Interface to apply classify session.
220 :param int table_index: table index to identify classify table.
221 :param str match: matched value for interested traffic.
222 :param int pbr_action: enable/disable PBR feature.
223 :param int vrfid: VRF id.
224 :param int is_add: option to configure classify session.
225 - create(1) or delete(0)
227 r = self.vapi.classify_add_del_session(
230 binascii.unhexlify(match),
231 hit_next_index=hit_next_index)
232 self.assertIsNotNone(r, msg='No response msg for add_del_session')
234 def input_acl_set_interface(self, intf, table_index, is_add=1):
235 """Configure Input ACL interface
237 :param VppInterface intf: Interface to apply Input ACL feature.
238 :param int table_index: table index to identify classify table.
239 :param int is_add: option to configure classify session.
240 - enable(1) or disable(0)
242 r = self.vapi.input_acl_set_interface(
245 l2_table_index=table_index)
246 self.assertIsNotNone(r, msg='No response msg for acl_set_interface')
248 def output_acl_set_interface(self, intf, table_index, is_add=1):
249 """Configure Output ACL interface
251 :param VppInterface intf: Interface to apply Output ACL feature.
252 :param int table_index: table index to identify classify table.
253 :param int is_add: option to configure classify session.
254 - enable(1) or disable(0)
256 r = self.vapi.output_acl_set_interface(
259 l2_table_index=table_index)
260 self.assertIsNotNone(r, msg='No response msg for acl_set_interface')
262 def create_hosts(self, count, start=0):
264 Create required number of host MAC addresses and distribute them among
265 interfaces. Create host IPv4 address for every host MAC address.
267 :param int count: Number of hosts to create MAC/IPv4 addresses for.
268 :param int start: Number to start numbering from.
270 n_int = len(self.pg_interfaces)
271 macs_per_if = count / n_int
273 for pg_if in self.pg_interfaces:
275 start_nr = macs_per_if * i + start
276 end_nr = count + start if i == (n_int - 1) \
277 else macs_per_if * (i + 1) + start
278 hosts = self.hosts_by_pg_idx[pg_if.sw_if_index]
279 for j in range(start_nr, end_nr):
281 "00:00:00:ff:%02x:%02x" % (pg_if.sw_if_index, j),
282 "172.17.1%02x.%u" % (pg_if.sw_if_index, j),
283 "2017:dead:%02x::%u" % (pg_if.sw_if_index, j))
286 def create_upper_layer(self, packet_index, proto, ports=0):
287 p = self.proto_map[proto]
290 return UDP(sport=random.randint(self.udp_sport_from,
292 dport=random.randint(self.udp_dport_from,
295 return UDP(sport=ports, dport=ports)
298 return TCP(sport=random.randint(self.tcp_sport_from,
300 dport=random.randint(self.tcp_dport_from,
303 return TCP(sport=ports, dport=ports)
306 def create_stream(self, src_if, packet_sizes, traffic_type=0, ipv6=0,
307 proto=-1, ports=0, fragments=False,
308 pkt_raw=True, etype=-1):
310 Create input packet stream for defined interface using hosts or
313 :param object src_if: Interface to create packet stream for.
314 :param list packet_sizes: List of required packet sizes.
315 :param traffic_type: 1: ICMP packet, 2: IPv6 with EH, 0: otherwise.
316 :return: Stream of packets.
319 if self.flows.__contains__(src_if):
320 src_hosts = self.hosts_by_pg_idx[src_if.sw_if_index]
321 for dst_if in self.flows[src_if]:
322 dst_hosts = self.hosts_by_pg_idx[dst_if.sw_if_index]
323 n_int = len(dst_hosts) * len(src_hosts)
324 for i in range(0, n_int):
325 dst_host = dst_hosts[i / len(src_hosts)]
326 src_host = src_hosts[i % len(src_hosts)]
327 pkt_info = self.create_packet_info(src_if, dst_if)
333 pkt_info.ip = random.choice([0, 1])
335 pkt_info.proto = random.choice(self.proto[self.IP])
337 pkt_info.proto = proto
338 payload = self.info_to_payload(pkt_info)
339 p = Ether(dst=dst_host.mac, src=src_host.mac)
341 p = Ether(dst=dst_host.mac,
345 p /= IPv6(dst=dst_host.ip6, src=src_host.ip6)
347 p /= IPv6ExtHdrFragment(offset=64, m=1)
350 p /= IP(src=src_host.ip4, dst=dst_host.ip4,
353 p /= IP(src=src_host.ip4, dst=dst_host.ip4)
354 if traffic_type == self.ICMP:
356 p /= ICMPv6EchoRequest(type=self.icmp6_type,
357 code=self.icmp6_code)
359 p /= ICMP(type=self.icmp4_type,
360 code=self.icmp4_code)
362 p /= self.create_upper_layer(i, pkt_info.proto, ports)
365 pkt_info.data = p.copy()
367 size = random.choice(packet_sizes)
368 self.extend_packet(p, size)
372 def verify_capture(self, pg_if, capture,
373 traffic_type=0, ip_type=0, etype=-1):
375 Verify captured input packet stream for defined interface.
377 :param object pg_if: Interface to verify captured packet stream for.
378 :param list capture: Captured packet stream.
379 :param traffic_type: 1: ICMP packet, 2: IPv6 with EH, 0: otherwise.
382 for i in self.pg_interfaces:
383 last_info[i.sw_if_index] = None
384 dst_sw_if_index = pg_if.sw_if_index
385 for packet in capture:
387 if packet[Ether].type != etype:
388 self.logger.error(ppp("Unexpected ethertype in packet:",
393 # Raw data for ICMPv6 are stored in ICMPv6EchoRequest.data
394 if traffic_type == self.ICMP and ip_type == self.IPV6:
395 payload_info = self.payload_to_info(
396 packet[ICMPv6EchoRequest].data)
397 payload = packet[ICMPv6EchoRequest]
399 payload_info = self.payload_to_info(str(packet[Raw]))
400 payload = packet[self.proto_map[payload_info.proto]]
402 self.logger.error(ppp("Unexpected or invalid packet "
403 "(outside network):", packet))
407 self.assertEqual(payload_info.ip, ip_type)
408 if traffic_type == self.ICMP:
410 if payload_info.ip == 0:
411 self.assertEqual(payload.type, self.icmp4_type)
412 self.assertEqual(payload.code, self.icmp4_code)
414 self.assertEqual(payload.type, self.icmp6_type)
415 self.assertEqual(payload.code, self.icmp6_code)
417 self.logger.error(ppp("Unexpected or invalid packet "
418 "(outside network):", packet))
422 ip_version = IPv6 if payload_info.ip == 1 else IP
424 ip = packet[ip_version]
425 packet_index = payload_info.index
427 self.assertEqual(payload_info.dst, dst_sw_if_index)
428 self.logger.debug("Got packet on port %s: src=%u (id=%u)" %
429 (pg_if.name, payload_info.src,
431 next_info = self.get_next_packet_info_for_interface2(
432 payload_info.src, dst_sw_if_index,
433 last_info[payload_info.src])
434 last_info[payload_info.src] = next_info
435 self.assertTrue(next_info is not None)
436 self.assertEqual(packet_index, next_info.index)
437 saved_packet = next_info.data
438 # Check standard fields
439 self.assertEqual(ip.src, saved_packet[ip_version].src)
440 self.assertEqual(ip.dst, saved_packet[ip_version].dst)
441 p = self.proto_map[payload_info.proto]
444 self.assertEqual(tcp.sport, saved_packet[
446 self.assertEqual(tcp.dport, saved_packet[
450 self.assertEqual(udp.sport, saved_packet[
452 self.assertEqual(udp.dport, saved_packet[
455 self.logger.error(ppp("Unexpected or invalid packet:",
458 for i in self.pg_interfaces:
459 remaining_packet = self.get_next_packet_info_for_interface2(
460 i, dst_sw_if_index, last_info[i.sw_if_index])
462 remaining_packet is None,
463 "Port %u: Packet expected from source %u didn't arrive" %
464 (dst_sw_if_index, i.sw_if_index))
466 def run_traffic_no_check(self):
468 # Create incoming packet streams for packet-generator interfaces
469 for i in self.pg_interfaces:
470 if self.flows.__contains__(i):
471 pkts = self.create_stream(i, self.pg_if_packet_sizes)
475 # Enable packet capture and start packet sending
476 self.pg_enable_capture(self.pg_interfaces)
479 def run_verify_test(self, traffic_type=0, ip_type=0, proto=-1, ports=0,
480 frags=False, pkt_raw=True, etype=-1):
482 # Create incoming packet streams for packet-generator interfaces
484 for i in self.pg_interfaces:
485 if self.flows.__contains__(i):
486 pkts = self.create_stream(i, self.pg_if_packet_sizes,
487 traffic_type, ip_type, proto, ports,
488 frags, pkt_raw, etype)
491 pkts_cnt += len(pkts)
493 # Enable packet capture and start packet sendingself.IPV
494 self.pg_enable_capture(self.pg_interfaces)
498 # Verify outgoing packet streams per packet-generator interface
499 for src_if in self.pg_interfaces:
500 if self.flows.__contains__(src_if):
501 for dst_if in self.flows[src_if]:
502 capture = dst_if.get_capture(pkts_cnt)
503 self.logger.info("Verifying capture on interface %s" %
505 self.verify_capture(dst_if, capture,
506 traffic_type, ip_type, etype)
508 def run_verify_negat_test(self, traffic_type=0, ip_type=0, proto=-1,
509 ports=0, frags=False, etype=-1):
511 self.reset_packet_infos()
512 for i in self.pg_interfaces:
513 if self.flows.__contains__(i):
514 pkts = self.create_stream(i, self.pg_if_packet_sizes,
515 traffic_type, ip_type, proto, ports,
520 # Enable packet capture and start packet sending
521 self.pg_enable_capture(self.pg_interfaces)
525 # Verify outgoing packet streams per packet-generator interface
526 for src_if in self.pg_interfaces:
527 if self.flows.__contains__(src_if):
528 for dst_if in self.flows[src_if]:
529 self.logger.info("Verifying capture on interface %s" %
531 capture = dst_if.get_capture(0)
532 self.assertEqual(len(capture), 0)
534 def build_classify_table(self, src_mac='', dst_mac='', ether_type='',
535 etype='', key='mac', hit_next_index=0xffffffff):
537 a_mask = self.build_mac_mask(src_mac=src_mac, dst_mac=dst_mac,
538 ether_type=ether_type)
539 self.create_classify_table(key, a_mask)
540 for host in self.hosts_by_pg_idx[self.pg0.sw_if_index]:
541 s_mac = host.mac if src_mac else ''
543 for dst_if in self.flows[self.pg0]:
544 for dst_host in self.hosts_by_pg_idx[dst_if.sw_if_index]:
545 self.create_classify_session(
546 self.pg0, self.acl_tbl_idx.get(key),
547 self.build_mac_match(src_mac=s_mac,
548 dst_mac=dst_host.mac,
550 hit_next_index=hit_next_index)
552 self.create_classify_session(
553 self.pg0, self.acl_tbl_idx.get(key),
554 self.build_mac_match(src_mac=s_mac, dst_mac='',
556 hit_next_index=hit_next_index)
558 def test_0000_warmup_test(self):
559 """ Learn the MAC addresses
562 self.run_traffic_no_check()
564 def test_0010_inacl_permit_src_mac(self):
565 """ Input L2 ACL test - permit source MAC
567 Test scenario for basic IP ACL with source IP
568 - Create IPv4 stream for pg0 -> pg1 interface.
569 - Create ACL with source MAC address.
570 - Send and verify received packets on pg1 interface.
573 self.build_classify_table(src_mac='ffffffffffff', key=key)
574 self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key))
575 self.acl_active_table = key
576 self.run_verify_test(self.IP, self.IPV4, -1)
578 def test_0011_inacl_permit_dst_mac(self):
579 """ Input L2 ACL test - permit destination MAC
581 Test scenario for basic IP ACL with source IP
582 - Create IPv4 stream for pg0 -> pg1 interface.
583 - Create ACL with destination MAC address.
584 - Send and verify received packets on pg1 interface.
587 self.build_classify_table(dst_mac='ffffffffffff', key=key)
588 self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key))
589 self.acl_active_table = key
590 self.run_verify_test(self.IP, self.IPV4, -1)
592 def test_0012_inacl_permit_src_dst_mac(self):
593 """ Input L2 ACL test - permit source and destination MAC
595 Test scenario for basic IP ACL with source IP
596 - Create IPv4 stream for pg0 -> pg1 interface.
597 - Create ACL with source and destination MAC addresses.
598 - Send and verify received packets on pg1 interface.
601 self.build_classify_table(
602 src_mac='ffffffffffff', dst_mac='ffffffffffff', key=key)
603 self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key))
604 self.acl_active_table = key
605 self.run_verify_test(self.IP, self.IPV4, -1)
607 def test_0013_inacl_permit_ether_type(self):
608 """ Input L2 ACL test - permit ether_type
610 Test scenario for basic IP ACL with source IP
611 - Create IPv4 stream for pg0 -> pg1 interface.
612 - Create ACL with destination MAC address.
613 - Send and verify received packets on pg1 interface.
616 self.build_classify_table(
617 ether_type='ffff', etype=hex(ETH_P_IP)[2:], key=key)
618 self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key))
619 self.acl_active_table = key
620 self.run_verify_test(self.IP, self.IPV4, -1)
622 def test_0015_inacl_deny(self):
623 """ Input L2 ACL test - deny
625 Test scenario for basic IP ACL with source IP
626 - Create IPv4 stream for pg0 -> pg1 interface.
628 - Create ACL with source MAC address.
629 - Send and verify no received packets on pg1 interface.
632 self.build_classify_table(
633 src_mac='ffffffffffff', hit_next_index=0, key=key)
634 self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key))
635 self.acl_active_table = key
636 self.run_verify_negat_test(self.IP, self.IPV4, -1)
638 def test_0020_outacl_permit(self):
639 """ Output L2 ACL test - permit
641 Test scenario for basic IP ACL with source IP
642 - Create IPv4 stream for pg0 -> pg1 interface.
643 - Create ACL with source MAC address.
644 - Send and verify received packets on pg1 interface.
647 self.build_classify_table(src_mac='ffffffffffff', key=key)
648 self.output_acl_set_interface(self.pg1, self.acl_tbl_idx.get(key))
649 self.acl_active_table = key
650 self.run_verify_test(self.IP, self.IPV4, -1)
652 def test_0025_outacl_deny(self):
653 """ Output L2 ACL test - deny
655 Test scenario for basic IP ACL with source IP
656 - Create IPv4 stream for pg0 -> pg1 interface.
657 - Create ACL with source MAC address.
658 - Send and verify no received packets on pg1 interface.
661 self.build_classify_table(
662 src_mac='ffffffffffff', hit_next_index=0, key=key)
663 self.output_acl_set_interface(self.pg1, self.acl_tbl_idx.get(key))
664 self.acl_active_table = key
665 self.run_verify_negat_test(self.IP, self.IPV4, -1)
667 def test_0030_inoutacl_permit(self):
668 """ Input+Output L2 ACL test - permit
670 Test scenario for basic IP ACL with source IP
671 - Create IPv4 stream for pg0 -> pg1 interface.
672 - Create ACLs with source MAC address.
673 - Send and verify received packets on pg1 interface.
676 self.build_classify_table(src_mac='ffffffffffff', key=key)
677 self.output_acl_set_interface(self.pg1, self.acl_tbl_idx.get(key))
678 self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key))
679 self.acl_active_table = key
680 self.run_verify_test(self.IP, self.IPV4, -1)
682 if __name__ == '__main__':
683 unittest.main(testRunner=VppTestRunner)