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(
109 rx_sw_if_index=pg_if.sw_if_index, bd_id=cls.bd_id)
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 def tearDownClass(cls):
137 super(TestClassifyAcl, cls).tearDownClass()
140 super(TestClassifyAcl, self).setUp()
142 self.acl_tbl_idx = {}
143 self.reset_packet_infos()
147 Show various debug prints after each test.
149 if not self.vpp_dead:
150 if self.acl_active_table == 'mac_inout':
151 self.output_acl_set_interface(
152 self.pg1, self.acl_tbl_idx.get(self.acl_active_table), 0)
153 self.input_acl_set_interface(
154 self.pg0, self.acl_tbl_idx.get(self.acl_active_table), 0)
155 self.acl_active_table = ''
156 elif self.acl_active_table == 'mac_out':
157 self.output_acl_set_interface(
158 self.pg1, self.acl_tbl_idx.get(self.acl_active_table), 0)
159 self.acl_active_table = ''
160 elif self.acl_active_table == 'mac_in':
161 self.input_acl_set_interface(
162 self.pg0, self.acl_tbl_idx.get(self.acl_active_table), 0)
163 self.acl_active_table = ''
165 super(TestClassifyAcl, self).tearDown()
167 def show_commands_at_teardown(self):
168 self.logger.info(self.vapi.ppcli("show inacl type l2"))
169 self.logger.info(self.vapi.ppcli("show outacl type l2"))
170 self.logger.info(self.vapi.ppcli("show classify tables verbose"))
171 self.logger.info(self.vapi.ppcli("show bridge-domain %s detail"
175 def build_mac_mask(dst_mac='', src_mac='', ether_type=''):
176 """Build MAC ACL mask data with hexstring format
178 :param str dst_mac: source MAC address <0-ffffffffffff>
179 :param str src_mac: destination MAC address <0-ffffffffffff>
180 :param str ether_type: ethernet type <0-ffff>
183 return ('{!s:0>12}{!s:0>12}{!s:0>4}'.format(
184 dst_mac, src_mac, ether_type)).rstrip('0')
187 def build_mac_match(dst_mac='', src_mac='', ether_type=''):
188 """Build MAC ACL match data with hexstring format
190 :param str dst_mac: source MAC address <x:x:x:x:x:x>
191 :param str src_mac: destination MAC address <x:x:x:x:x:x>
192 :param str ether_type: ethernet type <0-ffff>
195 dst_mac = dst_mac.replace(':', '')
197 src_mac = src_mac.replace(':', '')
199 return ('{!s:0>12}{!s:0>12}{!s:0>4}'.format(
200 dst_mac, src_mac, ether_type)).rstrip('0')
202 def create_classify_table(self, key, mask, data_offset=0, is_add=1):
203 """Create Classify Table
205 :param str key: key for classify table (ex, ACL name).
206 :param str mask: mask value for interested traffic.
207 :param int match_n_vectors:
208 :param int is_add: option to configure classify table.
209 - create(1) or delete(0)
211 r = self.vapi.classify_add_del_table(
213 binascii.unhexlify(mask),
214 match_n_vectors=(len(mask) - 1) // 32 + 1,
217 current_data_offset=data_offset)
218 self.assertIsNotNone(r, 'No response msg for add_del_table')
219 self.acl_tbl_idx[key] = r.new_table_index
221 def create_classify_session(self, intf, table_index, match,
222 hit_next_index=0xffffffff, is_add=1):
223 """Create Classify Session
225 :param VppInterface intf: Interface to apply classify session.
226 :param int table_index: table index to identify classify table.
227 :param str match: matched value for interested traffic.
228 :param int pbr_action: enable/disable PBR feature.
229 :param int vrfid: VRF id.
230 :param int is_add: option to configure classify session.
231 - create(1) or delete(0)
233 r = self.vapi.classify_add_del_session(
236 binascii.unhexlify(match),
237 hit_next_index=hit_next_index)
238 self.assertIsNotNone(r, 'No response msg for add_del_session')
240 def input_acl_set_interface(self, intf, table_index, is_add=1):
241 """Configure Input ACL interface
243 :param VppInterface intf: Interface to apply Input ACL feature.
244 :param int table_index: table index to identify classify table.
245 :param int is_add: option to configure classify session.
246 - enable(1) or disable(0)
248 r = self.vapi.input_acl_set_interface(
251 l2_table_index=table_index)
252 self.assertIsNotNone(r, 'No response msg for acl_set_interface')
254 def output_acl_set_interface(self, intf, table_index, is_add=1):
255 """Configure Output ACL interface
257 :param VppInterface intf: Interface to apply Output ACL feature.
258 :param int table_index: table index to identify classify table.
259 :param int is_add: option to configure classify session.
260 - enable(1) or disable(0)
262 r = self.vapi.output_acl_set_interface(
265 l2_table_index=table_index)
266 self.assertIsNotNone(r, 'No response msg for acl_set_interface')
268 def create_hosts(self, count, start=0):
270 Create required number of host MAC addresses and distribute them among
271 interfaces. Create host IPv4 address for every host MAC address.
273 :param int count: Number of hosts to create MAC/IPv4 addresses for.
274 :param int start: Number to start numbering from.
276 n_int = len(self.pg_interfaces)
277 macs_per_if = count / n_int
279 for pg_if in self.pg_interfaces:
281 start_nr = macs_per_if * i + start
282 end_nr = count + start if i == (n_int - 1) \
283 else macs_per_if * (i + 1) + start
284 hosts = self.hosts_by_pg_idx[pg_if.sw_if_index]
285 for j in range(start_nr, end_nr):
287 "00:00:00:ff:%02x:%02x" % (pg_if.sw_if_index, j),
288 "172.17.1%02x.%u" % (pg_if.sw_if_index, j),
289 "2017:dead:%02x::%u" % (pg_if.sw_if_index, j))
292 def create_upper_layer(self, packet_index, proto, ports=0):
293 p = self.proto_map[proto]
296 return UDP(sport=random.randint(self.udp_sport_from,
298 dport=random.randint(self.udp_dport_from,
301 return UDP(sport=ports, dport=ports)
304 return TCP(sport=random.randint(self.tcp_sport_from,
306 dport=random.randint(self.tcp_dport_from,
309 return TCP(sport=ports, dport=ports)
312 def create_stream(self, src_if, packet_sizes, traffic_type=0, ipv6=0,
313 proto=-1, ports=0, fragments=False,
314 pkt_raw=True, etype=-1):
316 Create input packet stream for defined interface using hosts or
319 :param object src_if: Interface to create packet stream for.
320 :param list packet_sizes: List of required packet sizes.
321 :param traffic_type: 1: ICMP packet, 2: IPv6 with EH, 0: otherwise.
322 :return: Stream of packets.
325 if self.flows.__contains__(src_if):
326 src_hosts = self.hosts_by_pg_idx[src_if.sw_if_index]
327 for dst_if in self.flows[src_if]:
328 dst_hosts = self.hosts_by_pg_idx[dst_if.sw_if_index]
329 n_int = len(dst_hosts) * len(src_hosts)
330 for i in range(0, n_int):
331 dst_host = dst_hosts[i / len(src_hosts)]
332 src_host = src_hosts[i % len(src_hosts)]
333 pkt_info = self.create_packet_info(src_if, dst_if)
339 pkt_info.ip = random.choice([0, 1])
341 pkt_info.proto = random.choice(self.proto[self.IP])
343 pkt_info.proto = proto
344 payload = self.info_to_payload(pkt_info)
345 p = Ether(dst=dst_host.mac, src=src_host.mac)
347 p = Ether(dst=dst_host.mac,
351 p /= IPv6(dst=dst_host.ip6, src=src_host.ip6)
353 p /= IPv6ExtHdrFragment(offset=64, m=1)
356 p /= IP(src=src_host.ip4, dst=dst_host.ip4,
359 p /= IP(src=src_host.ip4, dst=dst_host.ip4)
360 if traffic_type == self.ICMP:
362 p /= ICMPv6EchoRequest(type=self.icmp6_type,
363 code=self.icmp6_code)
365 p /= ICMP(type=self.icmp4_type,
366 code=self.icmp4_code)
368 p /= self.create_upper_layer(i, pkt_info.proto, ports)
371 pkt_info.data = p.copy()
373 size = random.choice(packet_sizes)
374 self.extend_packet(p, size)
378 def verify_capture(self, pg_if, capture,
379 traffic_type=0, ip_type=0, etype=-1):
381 Verify captured input packet stream for defined interface.
383 :param object pg_if: Interface to verify captured packet stream for.
384 :param list capture: Captured packet stream.
385 :param traffic_type: 1: ICMP packet, 2: IPv6 with EH, 0: otherwise.
388 for i in self.pg_interfaces:
389 last_info[i.sw_if_index] = None
390 dst_sw_if_index = pg_if.sw_if_index
391 for packet in capture:
393 if packet[Ether].type != etype:
394 self.logger.error(ppp("Unexpected ethertype in packet:",
399 # Raw data for ICMPv6 are stored in ICMPv6EchoRequest.data
400 if traffic_type == self.ICMP and ip_type == self.IPV6:
401 payload_info = self.payload_to_info(
402 packet[ICMPv6EchoRequest].data)
403 payload = packet[ICMPv6EchoRequest]
405 payload_info = self.payload_to_info(packet[Raw])
406 payload = packet[self.proto_map[payload_info.proto]]
408 self.logger.error(ppp("Unexpected or invalid packet "
409 "(outside network):", packet))
413 self.assertEqual(payload_info.ip, ip_type)
414 if traffic_type == self.ICMP:
416 if payload_info.ip == 0:
417 self.assertEqual(payload.type, self.icmp4_type)
418 self.assertEqual(payload.code, self.icmp4_code)
420 self.assertEqual(payload.type, self.icmp6_type)
421 self.assertEqual(payload.code, self.icmp6_code)
423 self.logger.error(ppp("Unexpected or invalid packet "
424 "(outside network):", packet))
428 ip_version = IPv6 if payload_info.ip == 1 else IP
430 ip = packet[ip_version]
431 packet_index = payload_info.index
433 self.assertEqual(payload_info.dst, dst_sw_if_index)
434 self.logger.debug("Got packet on port %s: src=%u (id=%u)" %
435 (pg_if.name, payload_info.src,
437 next_info = self.get_next_packet_info_for_interface2(
438 payload_info.src, dst_sw_if_index,
439 last_info[payload_info.src])
440 last_info[payload_info.src] = next_info
441 self.assertTrue(next_info is not None)
442 self.assertEqual(packet_index, next_info.index)
443 saved_packet = next_info.data
444 # Check standard fields
445 self.assertEqual(ip.src, saved_packet[ip_version].src)
446 self.assertEqual(ip.dst, saved_packet[ip_version].dst)
447 p = self.proto_map[payload_info.proto]
450 self.assertEqual(tcp.sport, saved_packet[
452 self.assertEqual(tcp.dport, saved_packet[
456 self.assertEqual(udp.sport, saved_packet[
458 self.assertEqual(udp.dport, saved_packet[
461 self.logger.error(ppp("Unexpected or invalid packet:",
464 for i in self.pg_interfaces:
465 remaining_packet = self.get_next_packet_info_for_interface2(
466 i, dst_sw_if_index, last_info[i.sw_if_index])
468 remaining_packet is None,
469 "Port %u: Packet expected from source %u didn't arrive" %
470 (dst_sw_if_index, i.sw_if_index))
472 def run_traffic_no_check(self):
474 # Create incoming packet streams for packet-generator interfaces
475 for i in self.pg_interfaces:
476 if self.flows.__contains__(i):
477 pkts = self.create_stream(i, self.pg_if_packet_sizes)
481 # Enable packet capture and start packet sending
482 self.pg_enable_capture(self.pg_interfaces)
485 def run_verify_test(self, traffic_type=0, ip_type=0, proto=-1, ports=0,
486 frags=False, pkt_raw=True, etype=-1):
488 # Create incoming packet streams for packet-generator interfaces
490 for i in self.pg_interfaces:
491 if self.flows.__contains__(i):
492 pkts = self.create_stream(i, self.pg_if_packet_sizes,
493 traffic_type, ip_type, proto, ports,
494 frags, pkt_raw, etype)
497 pkts_cnt += len(pkts)
499 # Enable packet capture and start packet sendingself.IPV
500 self.pg_enable_capture(self.pg_interfaces)
504 # Verify outgoing packet streams per packet-generator interface
505 for src_if in self.pg_interfaces:
506 if self.flows.__contains__(src_if):
507 for dst_if in self.flows[src_if]:
508 capture = dst_if.get_capture(pkts_cnt)
509 self.logger.info("Verifying capture on interface %s" %
511 self.verify_capture(dst_if, capture,
512 traffic_type, ip_type, etype)
514 def run_verify_negat_test(self, traffic_type=0, ip_type=0, proto=-1,
515 ports=0, frags=False, etype=-1):
517 self.reset_packet_infos()
518 for i in self.pg_interfaces:
519 if self.flows.__contains__(i):
520 pkts = self.create_stream(i, self.pg_if_packet_sizes,
521 traffic_type, ip_type, proto, ports,
526 # Enable packet capture and start packet sending
527 self.pg_enable_capture(self.pg_interfaces)
531 # Verify outgoing packet streams per packet-generator interface
532 for src_if in self.pg_interfaces:
533 if self.flows.__contains__(src_if):
534 for dst_if in self.flows[src_if]:
535 self.logger.info("Verifying capture on interface %s" %
537 capture = dst_if.get_capture(0)
538 self.assertEqual(len(capture), 0)
540 def build_classify_table(self, src_mac='', dst_mac='', ether_type='',
541 etype='', key='mac', hit_next_index=0xffffffff):
543 a_mask = self.build_mac_mask(src_mac=src_mac, dst_mac=dst_mac,
544 ether_type=ether_type)
545 self.create_classify_table(key, a_mask)
546 for host in self.hosts_by_pg_idx[self.pg0.sw_if_index]:
547 s_mac = host.mac if src_mac else ''
549 for dst_if in self.flows[self.pg0]:
550 for dst_host in self.hosts_by_pg_idx[dst_if.sw_if_index]:
551 self.create_classify_session(
552 self.pg0, self.acl_tbl_idx.get(key),
553 self.build_mac_match(src_mac=s_mac,
554 dst_mac=dst_host.mac,
556 hit_next_index=hit_next_index)
558 self.create_classify_session(
559 self.pg0, self.acl_tbl_idx.get(key),
560 self.build_mac_match(src_mac=s_mac, dst_mac='',
562 hit_next_index=hit_next_index)
564 def test_0000_warmup_test(self):
565 """ Learn the MAC addresses
568 self.run_traffic_no_check()
570 def test_0010_inacl_permit_src_mac(self):
571 """ Input L2 ACL test - permit source MAC
573 Test scenario for basic IP ACL with source IP
574 - Create IPv4 stream for pg0 -> pg1 interface.
575 - Create ACL with source MAC address.
576 - Send and verify received packets on pg1 interface.
579 self.build_classify_table(src_mac='ffffffffffff', key=key)
580 self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key))
581 self.acl_active_table = key
582 self.run_verify_test(self.IP, self.IPV4, -1)
584 def test_0011_inacl_permit_dst_mac(self):
585 """ Input L2 ACL test - permit destination MAC
587 Test scenario for basic IP ACL with source IP
588 - Create IPv4 stream for pg0 -> pg1 interface.
589 - Create ACL with destination MAC address.
590 - Send and verify received packets on pg1 interface.
593 self.build_classify_table(dst_mac='ffffffffffff', key=key)
594 self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key))
595 self.acl_active_table = key
596 self.run_verify_test(self.IP, self.IPV4, -1)
598 def test_0012_inacl_permit_src_dst_mac(self):
599 """ Input L2 ACL test - permit source and destination MAC
601 Test scenario for basic IP ACL with source IP
602 - Create IPv4 stream for pg0 -> pg1 interface.
603 - Create ACL with source and destination MAC addresses.
604 - Send and verify received packets on pg1 interface.
607 self.build_classify_table(
608 src_mac='ffffffffffff', dst_mac='ffffffffffff', key=key)
609 self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key))
610 self.acl_active_table = key
611 self.run_verify_test(self.IP, self.IPV4, -1)
613 def test_0013_inacl_permit_ether_type(self):
614 """ Input L2 ACL test - permit ether_type
616 Test scenario for basic IP ACL with source IP
617 - Create IPv4 stream for pg0 -> pg1 interface.
618 - Create ACL with destination MAC address.
619 - Send and verify received packets on pg1 interface.
622 self.build_classify_table(
623 ether_type='ffff', etype=hex(ETH_P_IP)[2:], key=key)
624 self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key))
625 self.acl_active_table = key
626 self.run_verify_test(self.IP, self.IPV4, -1)
628 def test_0015_inacl_deny(self):
629 """ Input L2 ACL test - deny
631 Test scenario for basic IP ACL with source IP
632 - Create IPv4 stream for pg0 -> pg1 interface.
634 - Create ACL with source MAC address.
635 - Send and verify no received packets on pg1 interface.
638 self.build_classify_table(
639 src_mac='ffffffffffff', hit_next_index=0, key=key)
640 self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key))
641 self.acl_active_table = key
642 self.run_verify_negat_test(self.IP, self.IPV4, -1)
644 def test_0020_outacl_permit(self):
645 """ Output L2 ACL test - permit
647 Test scenario for basic IP ACL with source IP
648 - Create IPv4 stream for pg0 -> pg1 interface.
649 - Create ACL with source MAC address.
650 - Send and verify received packets on pg1 interface.
653 self.build_classify_table(src_mac='ffffffffffff', key=key)
654 self.output_acl_set_interface(self.pg1, self.acl_tbl_idx.get(key))
655 self.acl_active_table = key
656 self.run_verify_test(self.IP, self.IPV4, -1)
658 def test_0025_outacl_deny(self):
659 """ Output L2 ACL test - deny
661 Test scenario for basic IP ACL with source IP
662 - Create IPv4 stream for pg0 -> pg1 interface.
663 - Create ACL with source MAC address.
664 - Send and verify no received packets on pg1 interface.
667 self.build_classify_table(
668 src_mac='ffffffffffff', hit_next_index=0, key=key)
669 self.output_acl_set_interface(self.pg1, self.acl_tbl_idx.get(key))
670 self.acl_active_table = key
671 self.run_verify_negat_test(self.IP, self.IPV4, -1)
673 def test_0030_inoutacl_permit(self):
674 """ Input+Output L2 ACL test - permit
676 Test scenario for basic IP ACL with source IP
677 - Create IPv4 stream for pg0 -> pg1 interface.
678 - Create ACLs with source MAC address.
679 - Send and verify received packets on pg1 interface.
682 self.build_classify_table(src_mac='ffffffffffff', key=key)
683 self.output_acl_set_interface(self.pg1, self.acl_tbl_idx.get(key))
684 self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key))
685 self.acl_active_table = key
686 self.run_verify_test(self.IP, self.IPV4, -1)
688 if __name__ == '__main__':
689 unittest.main(testRunner=VppTestRunner)