2 """ACL plugin Test Case HLD:
8 from scapy.packet import Raw
9 from scapy.layers.l2 import Ether
10 from scapy.layers.inet import IP, TCP, UDP, ICMP
11 from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest
12 from scapy.layers.inet6 import IPv6ExtHdrFragment
13 from framework import VppTestCase, VppTestRunner
14 from util import Host, ppp
16 from vpp_lo_interface import VppLoInterface
19 class TestACLplugin(VppTestCase):
20 """ ACL plugin Test Case """
36 proto = [[6, 17], [1, 58]]
37 proto_map = {1: 'ICMP', 58: 'ICMPv6EchoRequest', 6: 'TCP', 17: 'UDP'}
49 udp_sport_to = udp_sport_from + 5
50 udp_dport_from = 20000
51 udp_dport_to = udp_dport_from + 5000
53 tcp_sport_to = tcp_sport_from + 5
54 tcp_dport_from = 40000
55 tcp_dport_to = tcp_dport_from + 5000
58 udp_sport_to_2 = udp_sport_from_2 + 5
59 udp_dport_from_2 = 30000
60 udp_dport_to_2 = udp_dport_from_2 + 5000
61 tcp_sport_from_2 = 130
62 tcp_sport_to_2 = tcp_sport_from_2 + 5
63 tcp_dport_from_2 = 20000
64 tcp_dport_to_2 = tcp_dport_from_2 + 5000
66 icmp4_type = 8 # echo request
68 icmp6_type = 128 # echo request
84 Perform standard class setup (defined by class method setUpClass in
85 class VppTestCase) before running the test case, set test case related
86 variables and configure VPP.
88 super(TestACLplugin, cls).setUpClass()
91 # Create 2 pg interfaces
92 cls.create_pg_interfaces(range(2))
94 # Packet flows mapping pg0 -> pg1, pg2 etc.
96 cls.flows[cls.pg0] = [cls.pg1]
99 cls.pg_if_packet_sizes = [64, 512, 1518, 9018]
101 # Create BD with MAC learning and unknown unicast flooding disabled
102 # and put interfaces to this BD
103 cls.vapi.bridge_domain_add_del(bd_id=cls.bd_id, uu_flood=1,
105 for pg_if in cls.pg_interfaces:
106 cls.vapi.sw_interface_set_l2_bridge(
107 rx_sw_if_index=pg_if.sw_if_index, bd_id=cls.bd_id)
109 # Set up all interfaces
110 for i in cls.pg_interfaces:
113 # Mapping between packet-generator index and lists of test hosts
114 cls.hosts_by_pg_idx = dict()
115 for pg_if in cls.pg_interfaces:
116 cls.hosts_by_pg_idx[pg_if.sw_if_index] = []
118 # Create list of deleted hosts
119 cls.deleted_hosts_by_pg_idx = dict()
120 for pg_if in cls.pg_interfaces:
121 cls.deleted_hosts_by_pg_idx[pg_if.sw_if_index] = []
123 # warm-up the mac address tables
127 n_int = len(cls.pg_interfaces)
128 macs_per_if = count / n_int
130 for pg_if in cls.pg_interfaces:
132 start_nr = macs_per_if * i + start
133 end_nr = count + start if i == (n_int - 1) \
134 else macs_per_if * (i + 1) + start
135 hosts = cls.hosts_by_pg_idx[pg_if.sw_if_index]
136 for j in range(start_nr, end_nr):
138 "00:00:00:ff:%02x:%02x" % (pg_if.sw_if_index, j),
139 "172.17.1%02x.%u" % (pg_if.sw_if_index, j),
140 "2017:dead:%02x::%u" % (pg_if.sw_if_index, j))
144 super(TestACLplugin, cls).tearDownClass()
148 def tearDownClass(cls):
149 super(TestACLplugin, cls).tearDownClass()
152 super(TestACLplugin, self).setUp()
153 self.reset_packet_infos()
157 Show various debug prints after each test.
159 super(TestACLplugin, self).tearDown()
161 def show_commands_at_teardown(self):
162 cli = "show vlib graph l2-input-feat-arc"
163 self.logger.info(self.vapi.ppcli(cli))
164 cli = "show vlib graph l2-input-feat-arc-end"
165 self.logger.info(self.vapi.ppcli(cli))
166 cli = "show vlib graph l2-output-feat-arc"
167 self.logger.info(self.vapi.ppcli(cli))
168 cli = "show vlib graph l2-output-feat-arc-end"
169 self.logger.info(self.vapi.ppcli(cli))
170 self.logger.info(self.vapi.ppcli("show l2fib verbose"))
171 self.logger.info(self.vapi.ppcli("show acl-plugin acl"))
172 self.logger.info(self.vapi.ppcli("show acl-plugin interface"))
173 self.logger.info(self.vapi.ppcli("show acl-plugin tables"))
174 self.logger.info(self.vapi.ppcli("show bridge-domain %s detail"
177 def create_rule(self, ip=0, permit_deny=0, ports=PORTS_ALL, proto=-1,
178 s_prefix=0, s_ip='\x00\x00\x00\x00',
179 d_prefix=0, d_ip='\x00\x00\x00\x00'):
182 if ports == self.PORTS_ALL:
185 sport_to = 65535 if proto != 1 and proto != 58 else 255
187 elif ports == self.PORTS_RANGE:
189 sport_from = self.icmp4_type
190 sport_to = self.icmp4_type
191 dport_from = self.icmp4_code
192 dport_to = self.icmp4_code
194 sport_from = self.icmp6_type
195 sport_to = self.icmp6_type
196 dport_from = self.icmp6_code
197 dport_to = self.icmp6_code
198 elif proto == self.proto[self.IP][self.TCP]:
199 sport_from = self.tcp_sport_from
200 sport_to = self.tcp_sport_to
201 dport_from = self.tcp_dport_from
202 dport_to = self.tcp_dport_to
203 elif proto == self.proto[self.IP][self.UDP]:
204 sport_from = self.udp_sport_from
205 sport_to = self.udp_sport_to
206 dport_from = self.udp_dport_from
207 dport_to = self.udp_dport_to
208 elif ports == self.PORTS_RANGE_2:
210 sport_from = self.icmp4_type_2
211 sport_to = self.icmp4_type_2
212 dport_from = self.icmp4_code_from_2
213 dport_to = self.icmp4_code_to_2
215 sport_from = self.icmp6_type_2
216 sport_to = self.icmp6_type_2
217 dport_from = self.icmp6_code_from_2
218 dport_to = self.icmp6_code_to_2
219 elif proto == self.proto[self.IP][self.TCP]:
220 sport_from = self.tcp_sport_from_2
221 sport_to = self.tcp_sport_to_2
222 dport_from = self.tcp_dport_from_2
223 dport_to = self.tcp_dport_to_2
224 elif proto == self.proto[self.IP][self.UDP]:
225 sport_from = self.udp_sport_from_2
226 sport_to = self.udp_sport_to_2
227 dport_from = self.udp_dport_from_2
228 dport_to = self.udp_dport_to_2
235 rule = ({'is_permit': permit_deny, 'is_ipv6': ip, 'proto': proto,
236 'srcport_or_icmptype_first': sport_from,
237 'srcport_or_icmptype_last': sport_to,
238 'src_ip_prefix_len': s_prefix,
240 'dstport_or_icmpcode_first': dport_from,
241 'dstport_or_icmpcode_last': dport_to,
242 'dst_ip_prefix_len': d_prefix,
243 'dst_ip_addr': d_ip})
246 def apply_rules(self, rules, tag=b''):
247 reply = self.vapi.acl_add_replace(acl_index=4294967295, r=rules,
249 self.logger.info("Dumped ACL: " + str(
250 self.vapi.acl_dump(reply.acl_index)))
251 # Apply a ACL on the interface as inbound
252 for i in self.pg_interfaces:
253 self.vapi.acl_interface_set_acl_list(sw_if_index=i.sw_if_index,
255 acls=[reply.acl_index])
258 def apply_rules_to(self, rules, tag=b'', sw_if_index=0xFFFFFFFF):
259 reply = self.vapi.acl_add_replace(acl_index=4294967295, r=rules,
261 self.logger.info("Dumped ACL: " + str(
262 self.vapi.acl_dump(reply.acl_index)))
263 # Apply a ACL on the interface as inbound
264 self.vapi.acl_interface_set_acl_list(sw_if_index=sw_if_index,
266 acls=[reply.acl_index])
269 def etype_whitelist(self, whitelist, n_input):
270 # Apply whitelists on all the interfaces
271 for i in self.pg_interfaces:
272 # checkstyle can't read long names. Help them.
273 fun = self.vapi.acl_interface_set_etype_whitelist
274 fun(sw_if_index=i.sw_if_index, n_input=n_input,
278 def create_upper_layer(self, packet_index, proto, ports=0):
279 p = self.proto_map[proto]
282 return UDP(sport=random.randint(self.udp_sport_from,
284 dport=random.randint(self.udp_dport_from,
287 return UDP(sport=ports, dport=ports)
290 return TCP(sport=random.randint(self.tcp_sport_from,
292 dport=random.randint(self.tcp_dport_from,
295 return TCP(sport=ports, dport=ports)
298 def create_stream(self, src_if, packet_sizes, traffic_type=0, ipv6=0,
299 proto=-1, ports=0, fragments=False,
300 pkt_raw=True, etype=-1):
302 Create input packet stream for defined interface using hosts or
305 :param object src_if: Interface to create packet stream for.
306 :param list packet_sizes: List of required packet sizes.
307 :param traffic_type: 1: ICMP packet, 2: IPv6 with EH, 0: otherwise.
308 :return: Stream of packets.
311 if self.flows.__contains__(src_if):
312 src_hosts = self.hosts_by_pg_idx[src_if.sw_if_index]
313 for dst_if in self.flows[src_if]:
314 dst_hosts = self.hosts_by_pg_idx[dst_if.sw_if_index]
315 n_int = len(dst_hosts) * len(src_hosts)
316 for i in range(0, n_int):
317 dst_host = dst_hosts[i / len(src_hosts)]
318 src_host = src_hosts[i % len(src_hosts)]
319 pkt_info = self.create_packet_info(src_if, dst_if)
325 pkt_info.ip = random.choice([0, 1])
327 pkt_info.proto = random.choice(self.proto[self.IP])
329 pkt_info.proto = proto
330 payload = self.info_to_payload(pkt_info)
331 p = Ether(dst=dst_host.mac, src=src_host.mac)
333 p = Ether(dst=dst_host.mac,
337 p /= IPv6(dst=dst_host.ip6, src=src_host.ip6)
339 p /= IPv6ExtHdrFragment(offset=64, m=1)
342 p /= IP(src=src_host.ip4, dst=dst_host.ip4,
345 p /= IP(src=src_host.ip4, dst=dst_host.ip4)
346 if traffic_type == self.ICMP:
348 p /= ICMPv6EchoRequest(type=self.icmp6_type,
349 code=self.icmp6_code)
351 p /= ICMP(type=self.icmp4_type,
352 code=self.icmp4_code)
354 p /= self.create_upper_layer(i, pkt_info.proto, ports)
357 pkt_info.data = p.copy()
359 size = random.choice(packet_sizes)
360 self.extend_packet(p, size)
364 def verify_capture(self, pg_if, capture,
365 traffic_type=0, ip_type=0, etype=-1):
367 Verify captured input packet stream for defined interface.
369 :param object pg_if: Interface to verify captured packet stream for.
370 :param list capture: Captured packet stream.
371 :param traffic_type: 1: ICMP packet, 2: IPv6 with EH, 0: otherwise.
374 for i in self.pg_interfaces:
375 last_info[i.sw_if_index] = None
376 dst_sw_if_index = pg_if.sw_if_index
377 for packet in capture:
379 if packet[Ether].type != etype:
380 self.logger.error(ppp("Unexpected ethertype in packet:",
385 # Raw data for ICMPv6 are stored in ICMPv6EchoRequest.data
386 if traffic_type == self.ICMP and ip_type == self.IPV6:
387 payload_info = self.payload_to_info(
388 packet[ICMPv6EchoRequest], 'data')
389 payload = packet[ICMPv6EchoRequest]
391 payload_info = self.payload_to_info(packet[Raw])
392 payload = packet[self.proto_map[payload_info.proto]]
394 self.logger.error(ppp("Unexpected or invalid packet "
395 "(outside network):", packet))
399 self.assertEqual(payload_info.ip, ip_type)
400 if traffic_type == self.ICMP:
402 if payload_info.ip == 0:
403 self.assertEqual(payload.type, self.icmp4_type)
404 self.assertEqual(payload.code, self.icmp4_code)
406 self.assertEqual(payload.type, self.icmp6_type)
407 self.assertEqual(payload.code, self.icmp6_code)
409 self.logger.error(ppp("Unexpected or invalid packet "
410 "(outside network):", packet))
414 ip_version = IPv6 if payload_info.ip == 1 else IP
416 ip = packet[ip_version]
417 packet_index = payload_info.index
419 self.assertEqual(payload_info.dst, dst_sw_if_index)
420 self.logger.debug("Got packet on port %s: src=%u (id=%u)" %
421 (pg_if.name, payload_info.src,
423 next_info = self.get_next_packet_info_for_interface2(
424 payload_info.src, dst_sw_if_index,
425 last_info[payload_info.src])
426 last_info[payload_info.src] = next_info
427 self.assertTrue(next_info is not None)
428 self.assertEqual(packet_index, next_info.index)
429 saved_packet = next_info.data
430 # Check standard fields
431 self.assertEqual(ip.src, saved_packet[ip_version].src)
432 self.assertEqual(ip.dst, saved_packet[ip_version].dst)
433 p = self.proto_map[payload_info.proto]
436 self.assertEqual(tcp.sport, saved_packet[
438 self.assertEqual(tcp.dport, saved_packet[
442 self.assertEqual(udp.sport, saved_packet[
444 self.assertEqual(udp.dport, saved_packet[
447 self.logger.error(ppp("Unexpected or invalid packet:",
450 for i in self.pg_interfaces:
451 remaining_packet = self.get_next_packet_info_for_interface2(
452 i, dst_sw_if_index, last_info[i.sw_if_index])
454 remaining_packet is None,
455 "Port %u: Packet expected from source %u didn't arrive" %
456 (dst_sw_if_index, i.sw_if_index))
458 def run_traffic_no_check(self):
460 # Create incoming packet streams for packet-generator interfaces
461 for i in self.pg_interfaces:
462 if self.flows.__contains__(i):
463 pkts = self.create_stream(i, self.pg_if_packet_sizes)
467 # Enable packet capture and start packet sending
468 self.pg_enable_capture(self.pg_interfaces)
471 def run_verify_test(self, traffic_type=0, ip_type=0, proto=-1, ports=0,
472 frags=False, pkt_raw=True, etype=-1):
474 # Create incoming packet streams for packet-generator interfaces
476 for i in self.pg_interfaces:
477 if self.flows.__contains__(i):
478 pkts = self.create_stream(i, self.pg_if_packet_sizes,
479 traffic_type, ip_type, proto, ports,
480 frags, pkt_raw, etype)
483 pkts_cnt += len(pkts)
485 # Enable packet capture and start packet sendingself.IPV
486 self.pg_enable_capture(self.pg_interfaces)
488 self.logger.info("sent packets count: %d" % pkts_cnt)
491 # Verify outgoing packet streams per packet-generator interface
492 for src_if in self.pg_interfaces:
493 if self.flows.__contains__(src_if):
494 for dst_if in self.flows[src_if]:
495 capture = dst_if.get_capture(pkts_cnt)
496 self.logger.info("Verifying capture on interface %s" %
498 self.verify_capture(dst_if, capture,
499 traffic_type, ip_type, etype)
501 def run_verify_negat_test(self, traffic_type=0, ip_type=0, proto=-1,
502 ports=0, frags=False, etype=-1):
505 self.reset_packet_infos()
506 for i in self.pg_interfaces:
507 if self.flows.__contains__(i):
508 pkts = self.create_stream(i, self.pg_if_packet_sizes,
509 traffic_type, ip_type, proto, ports,
513 pkts_cnt += len(pkts)
515 # Enable packet capture and start packet sending
516 self.pg_enable_capture(self.pg_interfaces)
518 self.logger.info("sent packets count: %d" % pkts_cnt)
521 # Verify outgoing packet streams per packet-generator interface
522 for src_if in self.pg_interfaces:
523 if self.flows.__contains__(src_if):
524 for dst_if in self.flows[src_if]:
525 self.logger.info("Verifying capture on interface %s" %
527 capture = dst_if.get_capture(0)
528 self.assertEqual(len(capture), 0)
530 def test_0000_warmup_test(self):
531 """ ACL plugin version check; learn MACs
533 reply = self.vapi.papi.acl_plugin_get_version()
534 self.assertEqual(reply.major, 1)
535 self.logger.info("Working with ACL plugin version: %d.%d" % (
536 reply.major, reply.minor))
537 # minor version changes are non breaking
538 # self.assertEqual(reply.minor, 0)
540 def test_0001_acl_create(self):
541 """ ACL create/delete test
544 self.logger.info("ACLP_TEST_START_0001")
546 r = [{'is_permit': 1, 'is_ipv6': 0, 'proto': 17,
547 'srcport_or_icmptype_first': 1234,
548 'srcport_or_icmptype_last': 1235,
549 'src_ip_prefix_len': 0,
550 'src_ip_addr': b'\x00\x00\x00\x00',
551 'dstport_or_icmpcode_first': 1234,
552 'dstport_or_icmpcode_last': 1234,
553 'dst_ip_addr': b'\x00\x00\x00\x00',
554 'dst_ip_prefix_len': 0}]
555 # Test 1: add a new ACL
556 reply = self.vapi.acl_add_replace(acl_index=4294967295, r=r,
558 self.assertEqual(reply.retval, 0)
559 # The very first ACL gets #0
560 self.assertEqual(reply.acl_index, 0)
561 first_acl = reply.acl_index
562 rr = self.vapi.acl_dump(reply.acl_index)
563 self.logger.info("Dumped ACL: " + str(rr))
564 self.assertEqual(len(rr), 1)
565 # We should have the same number of ACL entries as we had asked
566 self.assertEqual(len(rr[0].r), len(r))
567 # The rules should be the same. But because the submitted and returned
568 # are different types, we need to iterate over rules and keys to get
570 for i_rule in range(0, len(r) - 1):
571 for rule_key in r[i_rule]:
572 self.assertEqual(rr[0].r[i_rule][rule_key],
575 # Add a deny-1234 ACL
576 r_deny = [{'is_permit': 0, 'is_ipv6': 0, 'proto': 17,
577 'srcport_or_icmptype_first': 1234,
578 'srcport_or_icmptype_last': 1235,
579 'src_ip_prefix_len': 0,
580 'src_ip_addr': b'\x00\x00\x00\x00',
581 'dstport_or_icmpcode_first': 1234,
582 'dstport_or_icmpcode_last': 1234,
583 'dst_ip_addr': b'\x00\x00\x00\x00',
584 'dst_ip_prefix_len': 0},
585 {'is_permit': 1, 'is_ipv6': 0, 'proto': 17,
586 'srcport_or_icmptype_first': 0,
587 'srcport_or_icmptype_last': 0,
588 'src_ip_prefix_len': 0,
589 'src_ip_addr': b'\x00\x00\x00\x00',
590 'dstport_or_icmpcode_first': 0,
591 'dstport_or_icmpcode_last': 0,
592 'dst_ip_addr': b'\x00\x00\x00\x00',
593 'dst_ip_prefix_len': 0}]
595 reply = self.vapi.acl_add_replace(acl_index=4294967295, r=r_deny,
596 tag=b"deny 1234;permit all")
597 self.assertEqual(reply.retval, 0)
598 # The second ACL gets #1
599 self.assertEqual(reply.acl_index, 1)
600 second_acl = reply.acl_index
602 # Test 2: try to modify a nonexistent ACL
603 reply = self.vapi.acl_add_replace(acl_index=432, r=r,
604 tag=b"FFFF:FFFF", expected_retval=-6)
605 self.assertEqual(reply.retval, -6)
606 # The ACL number should pass through
607 self.assertEqual(reply.acl_index, 432)
608 # apply an ACL on an interface inbound, try to delete ACL, must fail
609 self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg0.sw_if_index,
612 reply = self.vapi.acl_del(acl_index=first_acl, expected_retval=-142)
613 # Unapply an ACL and then try to delete it - must be ok
614 self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg0.sw_if_index,
617 reply = self.vapi.acl_del(acl_index=first_acl, expected_retval=0)
619 # apply an ACL on an interface outbound, try to delete ACL, must fail
620 self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg0.sw_if_index,
623 reply = self.vapi.acl_del(acl_index=second_acl, expected_retval=-143)
624 # Unapply the ACL and then try to delete it - must be ok
625 self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg0.sw_if_index,
628 reply = self.vapi.acl_del(acl_index=second_acl, expected_retval=0)
630 # try to apply a nonexistent ACL - must fail
631 self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg0.sw_if_index,
636 self.logger.info("ACLP_TEST_FINISH_0001")
638 def test_0002_acl_permit_apply(self):
639 """ permit ACL apply test
641 self.logger.info("ACLP_TEST_START_0002")
644 rules.append(self.create_rule(self.IPV4, self.PERMIT,
645 0, self.proto[self.IP][self.UDP]))
646 rules.append(self.create_rule(self.IPV4, self.PERMIT,
647 0, self.proto[self.IP][self.TCP]))
650 self.apply_rules(rules, b"permit per-flow")
652 # Traffic should still pass
653 self.run_verify_test(self.IP, self.IPV4, -1)
654 self.logger.info("ACLP_TEST_FINISH_0002")
656 def test_0003_acl_deny_apply(self):
657 """ deny ACL apply test
659 self.logger.info("ACLP_TEST_START_0003")
660 # Add a deny-flows ACL
662 rules.append(self.create_rule(self.IPV4, self.DENY,
663 self.PORTS_ALL, self.proto[self.IP][self.UDP]))
664 # Permit ip any any in the end
665 rules.append(self.create_rule(self.IPV4, self.PERMIT,
669 self.apply_rules(rules, b"deny per-flow;permit all")
671 # Traffic should not pass
672 self.run_verify_negat_test(self.IP, self.IPV4,
673 self.proto[self.IP][self.UDP])
674 self.logger.info("ACLP_TEST_FINISH_0003")
675 # self.assertEqual(1, 0)
677 def test_0004_vpp624_permit_icmpv4(self):
678 """ VPP_624 permit ICMPv4
680 self.logger.info("ACLP_TEST_START_0004")
684 rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE,
685 self.proto[self.ICMP][self.ICMPv4]))
686 # deny ip any any in the end
687 rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
690 self.apply_rules(rules, b"permit icmpv4")
692 # Traffic should still pass
693 self.run_verify_test(self.ICMP, self.IPV4,
694 self.proto[self.ICMP][self.ICMPv4])
696 self.logger.info("ACLP_TEST_FINISH_0004")
698 def test_0005_vpp624_permit_icmpv6(self):
699 """ VPP_624 permit ICMPv6
701 self.logger.info("ACLP_TEST_START_0005")
705 rules.append(self.create_rule(self.IPV6, self.PERMIT, self.PORTS_RANGE,
706 self.proto[self.ICMP][self.ICMPv6]))
707 # deny ip any any in the end
708 rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_ALL, 0))
711 self.apply_rules(rules, b"permit icmpv6")
713 # Traffic should still pass
714 self.run_verify_test(self.ICMP, self.IPV6,
715 self.proto[self.ICMP][self.ICMPv6])
717 self.logger.info("ACLP_TEST_FINISH_0005")
719 def test_0006_vpp624_deny_icmpv4(self):
720 """ VPP_624 deny ICMPv4
722 self.logger.info("ACLP_TEST_START_0006")
725 rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE,
726 self.proto[self.ICMP][self.ICMPv4]))
727 # permit ip any any in the end
728 rules.append(self.create_rule(self.IPV4, self.PERMIT,
732 self.apply_rules(rules, b"deny icmpv4")
734 # Traffic should not pass
735 self.run_verify_negat_test(self.ICMP, self.IPV4, 0)
737 self.logger.info("ACLP_TEST_FINISH_0006")
739 def test_0007_vpp624_deny_icmpv6(self):
740 """ VPP_624 deny ICMPv6
742 self.logger.info("ACLP_TEST_START_0007")
745 rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_RANGE,
746 self.proto[self.ICMP][self.ICMPv6]))
747 # deny ip any any in the end
748 rules.append(self.create_rule(self.IPV6, self.PERMIT,
752 self.apply_rules(rules, b"deny icmpv6")
754 # Traffic should not pass
755 self.run_verify_negat_test(self.ICMP, self.IPV6, 0)
757 self.logger.info("ACLP_TEST_FINISH_0007")
759 def test_0008_tcp_permit_v4(self):
762 self.logger.info("ACLP_TEST_START_0008")
766 rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE,
767 self.proto[self.IP][self.TCP]))
768 # deny ip any any in the end
769 rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
772 self.apply_rules(rules, b"permit ipv4 tcp")
774 # Traffic should still pass
775 self.run_verify_test(self.IP, self.IPV4, self.proto[self.IP][self.TCP])
777 self.logger.info("ACLP_TEST_FINISH_0008")
779 def test_0009_tcp_permit_v6(self):
782 self.logger.info("ACLP_TEST_START_0009")
786 rules.append(self.create_rule(self.IPV6, self.PERMIT, self.PORTS_RANGE,
787 self.proto[self.IP][self.TCP]))
788 # deny ip any any in the end
789 rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_ALL, 0))
792 self.apply_rules(rules, b"permit ip6 tcp")
794 # Traffic should still pass
795 self.run_verify_test(self.IP, self.IPV6, self.proto[self.IP][self.TCP])
797 self.logger.info("ACLP_TEST_FINISH_0008")
799 def test_0010_udp_permit_v4(self):
802 self.logger.info("ACLP_TEST_START_0010")
806 rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE,
807 self.proto[self.IP][self.UDP]))
808 # deny ip any any in the end
809 rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
812 self.apply_rules(rules, b"permit ipv udp")
814 # Traffic should still pass
815 self.run_verify_test(self.IP, self.IPV4, self.proto[self.IP][self.UDP])
817 self.logger.info("ACLP_TEST_FINISH_0010")
819 def test_0011_udp_permit_v6(self):
822 self.logger.info("ACLP_TEST_START_0011")
826 rules.append(self.create_rule(self.IPV6, self.PERMIT, self.PORTS_RANGE,
827 self.proto[self.IP][self.UDP]))
828 # deny ip any any in the end
829 rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_ALL, 0))
832 self.apply_rules(rules, b"permit ip6 udp")
834 # Traffic should still pass
835 self.run_verify_test(self.IP, self.IPV6, self.proto[self.IP][self.UDP])
837 self.logger.info("ACLP_TEST_FINISH_0011")
839 def test_0012_tcp_deny(self):
842 self.logger.info("ACLP_TEST_START_0012")
846 rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE,
847 self.proto[self.IP][self.TCP]))
848 rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_RANGE,
849 self.proto[self.IP][self.TCP]))
850 # permit ip any any in the end
851 rules.append(self.create_rule(self.IPV4, self.PERMIT,
853 rules.append(self.create_rule(self.IPV6, self.PERMIT,
857 self.apply_rules(rules, b"deny ip4/ip6 tcp")
859 # Traffic should not pass
860 self.run_verify_negat_test(self.IP, self.IPRANDOM,
861 self.proto[self.IP][self.TCP])
863 self.logger.info("ACLP_TEST_FINISH_0012")
865 def test_0013_udp_deny(self):
868 self.logger.info("ACLP_TEST_START_0013")
872 rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE,
873 self.proto[self.IP][self.UDP]))
874 rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_RANGE,
875 self.proto[self.IP][self.UDP]))
876 # permit ip any any in the end
877 rules.append(self.create_rule(self.IPV4, self.PERMIT,
879 rules.append(self.create_rule(self.IPV6, self.PERMIT,
883 self.apply_rules(rules, b"deny ip4/ip6 udp")
885 # Traffic should not pass
886 self.run_verify_negat_test(self.IP, self.IPRANDOM,
887 self.proto[self.IP][self.UDP])
889 self.logger.info("ACLP_TEST_FINISH_0013")
891 def test_0014_acl_dump(self):
892 """ verify add/dump acls
894 self.logger.info("ACLP_TEST_START_0014")
896 r = [[self.IPV4, self.PERMIT, 1234, self.proto[self.IP][self.TCP]],
897 [self.IPV4, self.PERMIT, 2345, self.proto[self.IP][self.UDP]],
898 [self.IPV4, self.PERMIT, 0, self.proto[self.IP][self.TCP]],
899 [self.IPV4, self.PERMIT, 0, self.proto[self.IP][self.UDP]],
900 [self.IPV4, self.PERMIT, 5, self.proto[self.ICMP][self.ICMPv4]],
901 [self.IPV6, self.PERMIT, 4321, self.proto[self.IP][self.TCP]],
902 [self.IPV6, self.PERMIT, 5432, self.proto[self.IP][self.UDP]],
903 [self.IPV6, self.PERMIT, 0, self.proto[self.IP][self.TCP]],
904 [self.IPV6, self.PERMIT, 0, self.proto[self.IP][self.UDP]],
905 [self.IPV6, self.PERMIT, 6, self.proto[self.ICMP][self.ICMPv6]],
906 [self.IPV4, self.DENY, self.PORTS_ALL, 0],
907 [self.IPV4, self.DENY, 1234, self.proto[self.IP][self.TCP]],
908 [self.IPV4, self.DENY, 2345, self.proto[self.IP][self.UDP]],
909 [self.IPV4, self.DENY, 5, self.proto[self.ICMP][self.ICMPv4]],
910 [self.IPV6, self.DENY, 4321, self.proto[self.IP][self.TCP]],
911 [self.IPV6, self.DENY, 5432, self.proto[self.IP][self.UDP]],
912 [self.IPV6, self.DENY, 6, self.proto[self.ICMP][self.ICMPv6]],
913 [self.IPV6, self.DENY, self.PORTS_ALL, 0]
916 # Add and verify new ACLs
918 for i in range(len(r)):
919 rules.append(self.create_rule(r[i][0], r[i][1], r[i][2], r[i][3]))
921 reply = self.vapi.acl_add_replace(acl_index=4294967295, r=rules)
922 result = self.vapi.acl_dump(reply.acl_index)
925 for drules in result:
927 self.assertEqual(dr.is_ipv6, r[i][0])
928 self.assertEqual(dr.is_permit, r[i][1])
929 self.assertEqual(dr.proto, r[i][3])
932 self.assertEqual(dr.srcport_or_icmptype_first, r[i][2])
935 self.assertEqual(dr.srcport_or_icmptype_first, 0)
936 self.assertEqual(dr.srcport_or_icmptype_last, 65535)
938 if dr.proto == self.proto[self.IP][self.TCP]:
939 self.assertGreater(dr.srcport_or_icmptype_first,
940 self.tcp_sport_from-1)
941 self.assertLess(dr.srcport_or_icmptype_first,
943 self.assertGreater(dr.dstport_or_icmpcode_last,
944 self.tcp_dport_from-1)
945 self.assertLess(dr.dstport_or_icmpcode_last,
947 elif dr.proto == self.proto[self.IP][self.UDP]:
948 self.assertGreater(dr.srcport_or_icmptype_first,
949 self.udp_sport_from-1)
950 self.assertLess(dr.srcport_or_icmptype_first,
952 self.assertGreater(dr.dstport_or_icmpcode_last,
953 self.udp_dport_from-1)
954 self.assertLess(dr.dstport_or_icmpcode_last,
958 self.logger.info("ACLP_TEST_FINISH_0014")
960 def test_0015_tcp_permit_port_v4(self):
961 """ permit single TCPv4
963 self.logger.info("ACLP_TEST_START_0015")
965 port = random.randint(0, 65535)
968 rules.append(self.create_rule(self.IPV4, self.PERMIT, port,
969 self.proto[self.IP][self.TCP]))
970 # deny ip any any in the end
971 rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
974 self.apply_rules(rules, b"permit ip4 tcp %d" % port)
976 # Traffic should still pass
977 self.run_verify_test(self.IP, self.IPV4,
978 self.proto[self.IP][self.TCP], port)
980 self.logger.info("ACLP_TEST_FINISH_0015")
982 def test_0016_udp_permit_port_v4(self):
983 """ permit single UDPv4
985 self.logger.info("ACLP_TEST_START_0016")
987 port = random.randint(0, 65535)
990 rules.append(self.create_rule(self.IPV4, self.PERMIT, port,
991 self.proto[self.IP][self.UDP]))
992 # deny ip any any in the end
993 rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
996 self.apply_rules(rules, b"permit ip4 tcp %d" % port)
998 # Traffic should still pass
999 self.run_verify_test(self.IP, self.IPV4,
1000 self.proto[self.IP][self.UDP], port)
1002 self.logger.info("ACLP_TEST_FINISH_0016")
1004 def test_0017_tcp_permit_port_v6(self):
1005 """ permit single TCPv6
1007 self.logger.info("ACLP_TEST_START_0017")
1009 port = random.randint(0, 65535)
1012 rules.append(self.create_rule(self.IPV6, self.PERMIT, port,
1013 self.proto[self.IP][self.TCP]))
1014 # deny ip any any in the end
1015 rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_ALL, 0))
1018 self.apply_rules(rules, b"permit ip4 tcp %d" % port)
1020 # Traffic should still pass
1021 self.run_verify_test(self.IP, self.IPV6,
1022 self.proto[self.IP][self.TCP], port)
1024 self.logger.info("ACLP_TEST_FINISH_0017")
1026 def test_0018_udp_permit_port_v6(self):
1027 """ permit single UPPv6
1029 self.logger.info("ACLP_TEST_START_0018")
1031 port = random.randint(0, 65535)
1034 rules.append(self.create_rule(self.IPV6, self.PERMIT, port,
1035 self.proto[self.IP][self.UDP]))
1036 # deny ip any any in the end
1037 rules.append(self.create_rule(self.IPV6, self.DENY,
1041 self.apply_rules(rules, b"permit ip4 tcp %d" % port)
1043 # Traffic should still pass
1044 self.run_verify_test(self.IP, self.IPV6,
1045 self.proto[self.IP][self.UDP], port)
1047 self.logger.info("ACLP_TEST_FINISH_0018")
1049 def test_0019_udp_deny_port(self):
1050 """ deny single TCPv4/v6
1052 self.logger.info("ACLP_TEST_START_0019")
1054 port = random.randint(0, 65535)
1057 rules.append(self.create_rule(self.IPV4, self.DENY, port,
1058 self.proto[self.IP][self.TCP]))
1059 rules.append(self.create_rule(self.IPV6, self.DENY, port,
1060 self.proto[self.IP][self.TCP]))
1061 # Permit ip any any in the end
1062 rules.append(self.create_rule(self.IPV4, self.PERMIT,
1064 rules.append(self.create_rule(self.IPV6, self.PERMIT,
1068 self.apply_rules(rules, b"deny ip4/ip6 udp %d" % port)
1070 # Traffic should not pass
1071 self.run_verify_negat_test(self.IP, self.IPRANDOM,
1072 self.proto[self.IP][self.TCP], port)
1074 self.logger.info("ACLP_TEST_FINISH_0019")
1076 def test_0020_udp_deny_port(self):
1077 """ deny single UDPv4/v6
1079 self.logger.info("ACLP_TEST_START_0020")
1081 port = random.randint(0, 65535)
1084 rules.append(self.create_rule(self.IPV4, self.DENY, port,
1085 self.proto[self.IP][self.UDP]))
1086 rules.append(self.create_rule(self.IPV6, self.DENY, port,
1087 self.proto[self.IP][self.UDP]))
1088 # Permit ip any any in the end
1089 rules.append(self.create_rule(self.IPV4, self.PERMIT,
1091 rules.append(self.create_rule(self.IPV6, self.PERMIT,
1095 self.apply_rules(rules, b"deny ip4/ip6 udp %d" % port)
1097 # Traffic should not pass
1098 self.run_verify_negat_test(self.IP, self.IPRANDOM,
1099 self.proto[self.IP][self.UDP], port)
1101 self.logger.info("ACLP_TEST_FINISH_0020")
1103 def test_0021_udp_deny_port_verify_fragment_deny(self):
1104 """ deny single UDPv4/v6, permit ip any, verify non-initial fragment
1107 self.logger.info("ACLP_TEST_START_0021")
1109 port = random.randint(0, 65535)
1112 rules.append(self.create_rule(self.IPV4, self.DENY, port,
1113 self.proto[self.IP][self.UDP]))
1114 rules.append(self.create_rule(self.IPV6, self.DENY, port,
1115 self.proto[self.IP][self.UDP]))
1116 # deny ip any any in the end
1117 rules.append(self.create_rule(self.IPV4, self.PERMIT,
1119 rules.append(self.create_rule(self.IPV6, self.PERMIT,
1123 self.apply_rules(rules, b"deny ip4/ip6 udp %d" % port)
1125 # Traffic should not pass
1126 self.run_verify_negat_test(self.IP, self.IPRANDOM,
1127 self.proto[self.IP][self.UDP], port, True)
1129 self.logger.info("ACLP_TEST_FINISH_0021")
1131 def test_0022_zero_length_udp_ipv4(self):
1132 """ VPP-687 zero length udp ipv4 packet"""
1133 self.logger.info("ACLP_TEST_START_0022")
1135 port = random.randint(0, 65535)
1138 rules.append(self.create_rule(self.IPV4, self.PERMIT, port,
1139 self.proto[self.IP][self.UDP]))
1140 # deny ip any any in the end
1142 self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
1145 self.apply_rules(rules, b"permit empty udp ip4 %d" % port)
1147 # Traffic should still pass
1148 # Create incoming packet streams for packet-generator interfaces
1150 pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes,
1152 self.proto[self.IP][self.UDP], port,
1155 self.pg0.add_stream(pkts)
1156 pkts_cnt += len(pkts)
1158 # Enable packet capture and start packet sendingself.IPV
1159 self.pg_enable_capture(self.pg_interfaces)
1162 self.pg1.get_capture(pkts_cnt)
1164 self.logger.info("ACLP_TEST_FINISH_0022")
1166 def test_0023_zero_length_udp_ipv6(self):
1167 """ VPP-687 zero length udp ipv6 packet"""
1168 self.logger.info("ACLP_TEST_START_0023")
1170 port = random.randint(0, 65535)
1173 rules.append(self.create_rule(self.IPV6, self.PERMIT, port,
1174 self.proto[self.IP][self.UDP]))
1175 # deny ip any any in the end
1176 rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_ALL, 0))
1179 self.apply_rules(rules, b"permit empty udp ip6 %d" % port)
1181 # Traffic should still pass
1182 # Create incoming packet streams for packet-generator interfaces
1184 pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes,
1186 self.proto[self.IP][self.UDP], port,
1189 self.pg0.add_stream(pkts)
1190 pkts_cnt += len(pkts)
1192 # Enable packet capture and start packet sendingself.IPV
1193 self.pg_enable_capture(self.pg_interfaces)
1196 # Verify outgoing packet streams per packet-generator interface
1197 self.pg1.get_capture(pkts_cnt)
1199 self.logger.info("ACLP_TEST_FINISH_0023")
1201 def test_0108_tcp_permit_v4(self):
1202 """ permit TCPv4 + non-match range
1204 self.logger.info("ACLP_TEST_START_0108")
1208 rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE_2,
1209 self.proto[self.IP][self.TCP]))
1210 rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE,
1211 self.proto[self.IP][self.TCP]))
1212 # deny ip any any in the end
1213 rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
1216 self.apply_rules(rules, b"permit ipv4 tcp")
1218 # Traffic should still pass
1219 self.run_verify_test(self.IP, self.IPV4, self.proto[self.IP][self.TCP])
1221 self.logger.info("ACLP_TEST_FINISH_0108")
1223 def test_0109_tcp_permit_v6(self):
1224 """ permit TCPv6 + non-match range
1226 self.logger.info("ACLP_TEST_START_0109")
1230 rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_RANGE_2,
1231 self.proto[self.IP][self.TCP]))
1232 rules.append(self.create_rule(self.IPV6, self.PERMIT, self.PORTS_RANGE,
1233 self.proto[self.IP][self.TCP]))
1234 # deny ip any any in the end
1235 rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_ALL, 0))
1238 self.apply_rules(rules, b"permit ip6 tcp")
1240 # Traffic should still pass
1241 self.run_verify_test(self.IP, self.IPV6, self.proto[self.IP][self.TCP])
1243 self.logger.info("ACLP_TEST_FINISH_0109")
1245 def test_0110_udp_permit_v4(self):
1246 """ permit UDPv4 + non-match range
1248 self.logger.info("ACLP_TEST_START_0110")
1252 rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE_2,
1253 self.proto[self.IP][self.UDP]))
1254 rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE,
1255 self.proto[self.IP][self.UDP]))
1256 # deny ip any any in the end
1257 rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
1260 self.apply_rules(rules, b"permit ipv4 udp")
1262 # Traffic should still pass
1263 self.run_verify_test(self.IP, self.IPV4, self.proto[self.IP][self.UDP])
1265 self.logger.info("ACLP_TEST_FINISH_0110")
1267 def test_0111_udp_permit_v6(self):
1268 """ permit UDPv6 + non-match range
1270 self.logger.info("ACLP_TEST_START_0111")
1274 rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_RANGE_2,
1275 self.proto[self.IP][self.UDP]))
1276 rules.append(self.create_rule(self.IPV6, self.PERMIT, self.PORTS_RANGE,
1277 self.proto[self.IP][self.UDP]))
1278 # deny ip any any in the end
1279 rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_ALL, 0))
1282 self.apply_rules(rules, b"permit ip6 udp")
1284 # Traffic should still pass
1285 self.run_verify_test(self.IP, self.IPV6, self.proto[self.IP][self.UDP])
1287 self.logger.info("ACLP_TEST_FINISH_0111")
1289 def test_0112_tcp_deny(self):
1290 """ deny TCPv4/v6 + non-match range
1292 self.logger.info("ACLP_TEST_START_0112")
1296 rules.append(self.create_rule(self.IPV4, self.PERMIT,
1298 self.proto[self.IP][self.TCP]))
1299 rules.append(self.create_rule(self.IPV6, self.PERMIT,
1301 self.proto[self.IP][self.TCP]))
1302 rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE,
1303 self.proto[self.IP][self.TCP]))
1304 rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_RANGE,
1305 self.proto[self.IP][self.TCP]))
1306 # permit ip any any in the end
1307 rules.append(self.create_rule(self.IPV4, self.PERMIT,
1309 rules.append(self.create_rule(self.IPV6, self.PERMIT,
1313 self.apply_rules(rules, b"deny ip4/ip6 tcp")
1315 # Traffic should not pass
1316 self.run_verify_negat_test(self.IP, self.IPRANDOM,
1317 self.proto[self.IP][self.TCP])
1319 self.logger.info("ACLP_TEST_FINISH_0112")
1321 def test_0113_udp_deny(self):
1322 """ deny UDPv4/v6 + non-match range
1324 self.logger.info("ACLP_TEST_START_0113")
1328 rules.append(self.create_rule(self.IPV4, self.PERMIT,
1330 self.proto[self.IP][self.UDP]))
1331 rules.append(self.create_rule(self.IPV6, self.PERMIT,
1333 self.proto[self.IP][self.UDP]))
1334 rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE,
1335 self.proto[self.IP][self.UDP]))
1336 rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_RANGE,
1337 self.proto[self.IP][self.UDP]))
1338 # permit ip any any in the end
1339 rules.append(self.create_rule(self.IPV4, self.PERMIT,
1341 rules.append(self.create_rule(self.IPV6, self.PERMIT,
1345 self.apply_rules(rules, b"deny ip4/ip6 udp")
1347 # Traffic should not pass
1348 self.run_verify_negat_test(self.IP, self.IPRANDOM,
1349 self.proto[self.IP][self.UDP])
1351 self.logger.info("ACLP_TEST_FINISH_0113")
1353 def test_0300_tcp_permit_v4_etype_aaaa(self):
1354 """ permit TCPv4, send 0xAAAA etype
1356 self.logger.info("ACLP_TEST_START_0300")
1360 rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE_2,
1361 self.proto[self.IP][self.TCP]))
1362 rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE,
1363 self.proto[self.IP][self.TCP]))
1364 # deny ip any any in the end
1365 rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
1368 self.apply_rules(rules, b"permit ipv4 tcp")
1370 # Traffic should still pass also for an odd ethertype
1371 self.run_verify_test(self.IP, self.IPV4, self.proto[self.IP][self.TCP],
1372 0, False, True, 0xaaaa)
1373 self.logger.info("ACLP_TEST_FINISH_0300")
1375 def test_0305_tcp_permit_v4_etype_blacklist_aaaa(self):
1376 """ permit TCPv4, whitelist 0x0BBB ethertype, send 0xAAAA-blocked
1378 self.logger.info("ACLP_TEST_START_0305")
1382 rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE_2,
1383 self.proto[self.IP][self.TCP]))
1384 rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE,
1385 self.proto[self.IP][self.TCP]))
1386 # deny ip any any in the end
1387 rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
1390 self.apply_rules(rules, b"permit ipv4 tcp")
1392 # whitelist the 0xbbbb etype - so the 0xaaaa should be blocked
1393 self.etype_whitelist([0xbbb], 1)
1395 # The oddball ethertype should be blocked
1396 self.run_verify_negat_test(self.IP, self.IPV4,
1397 self.proto[self.IP][self.TCP],
1400 # remove the whitelist
1401 self.etype_whitelist([], 0)
1403 self.logger.info("ACLP_TEST_FINISH_0305")
1405 def test_0306_tcp_permit_v4_etype_blacklist_aaaa(self):
1406 """ permit TCPv4, whitelist 0x0BBB ethertype, send 0x0BBB - pass
1408 self.logger.info("ACLP_TEST_START_0306")
1412 rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE_2,
1413 self.proto[self.IP][self.TCP]))
1414 rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE,
1415 self.proto[self.IP][self.TCP]))
1416 # deny ip any any in the end
1417 rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
1420 self.apply_rules(rules, b"permit ipv4 tcp")
1422 # whitelist the 0xbbbb etype - so the 0xaaaa should be blocked
1423 self.etype_whitelist([0xbbb], 1)
1425 # The whitelisted traffic, should pass
1426 self.run_verify_test(self.IP, self.IPV4, self.proto[self.IP][self.TCP],
1427 0, False, True, 0x0bbb)
1429 # remove the whitelist, the previously blocked 0xAAAA should pass now
1430 self.etype_whitelist([], 0)
1432 self.logger.info("ACLP_TEST_FINISH_0306")
1434 def test_0307_tcp_permit_v4_etype_blacklist_aaaa(self):
1435 """ permit TCPv4, whitelist 0x0BBB, remove, send 0xAAAA - pass
1437 self.logger.info("ACLP_TEST_START_0307")
1441 rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE_2,
1442 self.proto[self.IP][self.TCP]))
1443 rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE,
1444 self.proto[self.IP][self.TCP]))
1445 # deny ip any any in the end
1446 rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
1449 self.apply_rules(rules, b"permit ipv4 tcp")
1451 # whitelist the 0xbbbb etype - so the 0xaaaa should be blocked
1452 self.etype_whitelist([0xbbb], 1)
1453 # remove the whitelist, the previously blocked 0xAAAA should pass now
1454 self.etype_whitelist([], 0)
1456 # The whitelisted traffic, should pass
1457 self.run_verify_test(self.IP, self.IPV4, self.proto[self.IP][self.TCP],
1458 0, False, True, 0xaaaa)
1460 self.logger.info("ACLP_TEST_FINISH_0306")
1462 def test_0315_del_intf(self):
1463 """ apply an acl and delete the interface
1465 self.logger.info("ACLP_TEST_START_0315")
1469 rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE_2,
1470 self.proto[self.IP][self.TCP]))
1471 rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE,
1472 self.proto[self.IP][self.TCP]))
1473 # deny ip any any in the end
1474 rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
1476 # create an interface
1478 intf.append(VppLoInterface(self))
1481 self.apply_rules_to(rules, b"permit ipv4 tcp", intf[0].sw_if_index)
1483 # Remove the interface
1484 intf[0].remove_vpp_config()
1486 self.logger.info("ACLP_TEST_FINISH_0315")
1488 if __name__ == '__main__':
1489 unittest.main(testRunner=VppTestRunner)