X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftest_acl_plugin_l2l3.py;h=48faafb7398f2366c41ffd08cf0795c09abc968e;hb=0c373fc928316ef9928f4171b7c602c259c60d77;hp=26b562e8ad0c7335b7dcdbc59b291eeef04aaf54;hpb=b9ef2739dd1e681143e78ada1ffc342ae2fb89b0;p=vpp.git diff --git a/test/test_acl_plugin_l2l3.py b/test/test_acl_plugin_l2l3.py index 26b562e8ad0..48faafb7398 100644 --- a/test/test_acl_plugin_l2l3.py +++ b/test/test_acl_plugin_l2l3.py @@ -1,4 +1,4 @@ -#!/usr/bin/env python +#!/usr/bin/env python3 """ACL IRB Test Case HLD: **config** @@ -23,11 +23,14 @@ """ +import copy import unittest from socket import inet_pton, AF_INET, AF_INET6 from random import choice, shuffle from pprint import pprint +from ipaddress import ip_network +import scapy.compat from scapy.packet import Raw from scapy.layers.l2 import Ether from scapy.layers.inet import IP, UDP, ICMP, TCP @@ -36,11 +39,14 @@ from scapy.layers.inet6 import ICMPv6EchoReply, IPv6ExtHdrRouting from scapy.layers.inet6 import IPv6ExtHdrFragment from framework import VppTestCase, VppTestRunner +from vpp_l2 import L2_PORT_TYPE import time +from vpp_acl import AclRule, VppAcl, VppAclInterface -class TestIpIrb(VppTestCase): - """IRB Test Case""" + +class TestACLpluginL2L3(VppTestCase): + """TestACLpluginL2L3 Test Case""" @classmethod def setUpClass(cls): @@ -52,7 +58,7 @@ class TestIpIrb(VppTestCase): #. Loopback BVI interface has remote hosts, one half of hosts are behind pg0 second behind pg1. """ - super(TestIpIrb, cls).setUpClass() + super(TestACLpluginL2L3, cls).setUpClass() cls.pg_if_packet_sizes = [64, 512, 1518, 9018] # packet sizes cls.bd_id = 10 @@ -70,11 +76,12 @@ class TestIpIrb(VppTestCase): # Create BD with MAC learning enabled and put interfaces to this BD cls.vapi.sw_interface_set_l2_bridge( - cls.loop0.sw_if_index, bd_id=cls.bd_id, bvi=1) - cls.vapi.sw_interface_set_l2_bridge( - cls.pg0.sw_if_index, bd_id=cls.bd_id) - cls.vapi.sw_interface_set_l2_bridge( - cls.pg1.sw_if_index, bd_id=cls.bd_id) + rx_sw_if_index=cls.loop0.sw_if_index, bd_id=cls.bd_id, + port_type=L2_PORT_TYPE.BVI) + cls.vapi.sw_interface_set_l2_bridge(rx_sw_if_index=cls.pg0.sw_if_index, + bd_id=cls.bd_id) + cls.vapi.sw_interface_set_l2_bridge(rx_sw_if_index=cls.pg1.sw_if_index, + bd_id=cls.bd_id) # Configure IPv4 addresses on loopback interface and routed interface cls.loop0.config_ip4() @@ -92,36 +99,44 @@ class TestIpIrb(VppTestCase): cls.WITHOUT_EH = False cls.WITH_EH = True + cls.STATELESS_ICMP = False + cls.STATEFUL_ICMP = True # Loopback BVI interface has remote hosts, one half of hosts are behind # pg0 second behind pg1 half = cls.remote_hosts_count // 2 cls.pg0.remote_hosts = cls.loop0.remote_hosts[:half] cls.pg1.remote_hosts = cls.loop0.remote_hosts[half:] + reply = cls.vapi.papi.acl_stats_intf_counters_enable(enable=1) + + @classmethod + def tearDownClass(cls): + reply = cls.vapi.papi.acl_stats_intf_counters_enable(enable=0) + super(TestACLpluginL2L3, cls).tearDownClass() def tearDown(self): """Run standard test teardown and log ``show l2patch``, ``show l2fib verbose``,``show bridge-domain detail``, - ``show ip arp``. + ``show ip neighbors``. """ - super(TestIpIrb, self).tearDown() - if not self.vpp_dead: - self.logger.info(self.vapi.cli("show l2patch")) - self.logger.info(self.vapi.cli("show classify tables")) - self.logger.info(self.vapi.cli("show vlib graph")) - self.logger.info(self.vapi.cli("show l2fib verbose")) - self.logger.info(self.vapi.cli("show bridge-domain %s detail" % - self.bd_id)) - self.logger.info(self.vapi.cli("show ip arp")) - self.logger.info(self.vapi.cli("show ip6 neighbors")) - self.logger.info(self.vapi.cli("show acl-plugin sessions")) - self.logger.info(self.vapi.cli("show acl-plugin acl")) - self.logger.info(self.vapi.cli("show acl-plugin interface")) - self.logger.info(self.vapi.cli("show acl-plugin tables")) + super(TestACLpluginL2L3, self).tearDown() + + def show_commands_at_teardown(self): + self.logger.info(self.vapi.cli("show l2patch")) + self.logger.info(self.vapi.cli("show classify tables")) + self.logger.info(self.vapi.cli("show l2fib verbose")) + self.logger.info(self.vapi.cli("show bridge-domain %s detail" % + self.bd_id)) + self.logger.info(self.vapi.cli("show ip neighbors")) + cmd = "show acl-plugin sessions verbose 1" + self.logger.info(self.vapi.cli(cmd)) + self.logger.info(self.vapi.cli("show acl-plugin acl")) + self.logger.info(self.vapi.cli("show acl-plugin interface")) + self.logger.info(self.vapi.cli("show acl-plugin tables")) def create_stream(self, src_ip_if, dst_ip_if, reverse, packet_sizes, is_ip6, expect_blocked, expect_established, - add_extension_header): + add_extension_header, icmp_stateful=False): pkts = [] rules = [] permit_rules = [] @@ -129,7 +144,15 @@ class TestIpIrb(VppTestCase): total_packet_count = 8 for i in range(0, total_packet_count): modulo = (i//2) % 2 - can_reflect_this_packet = (modulo == 0) + icmp_type_delta = i % 2 + icmp_code = i + is_udp_packet = (modulo == 0) + if is_udp_packet and icmp_stateful: + continue + is_reflectable_icmp = (icmp_stateful and icmp_type_delta == 0 and + not is_udp_packet) + is_reflected_icmp = is_reflectable_icmp and expect_established + can_reflect_this_packet = is_udp_packet or is_reflectable_icmp is_permit = i % 2 remote_dst_index = i % len(dst_ip_if.remote_hosts) remote_dst_host = dst_ip_if.remote_hosts[remote_dst_index] @@ -165,13 +188,15 @@ class TestIpIrb(VppTestCase): dst_ip4 = remote_dst_host.ip4 src_l4 = 1234 + i dst_l4 = 4321 + i + if is_reflected_icmp: + icmp_type_delta = 1 # default ULP should be something we do not use in tests ulp_l4 = TCP(sport=src_l4, dport=dst_l4) # potentially a chain of protocols leading to ULP ulp = ulp_l4 - if can_reflect_this_packet: + if is_udp_packet: if is_ip6: ulp_l4 = UDP(sport=src_l4, dport=dst_l4) if add_extension_header: @@ -198,14 +223,15 @@ class TestIpIrb(VppTestCase): Raw(payload)) elif modulo == 1: if is_ip6: - ulp_l4 = ICMPv6Unknown(type=128 + (i % 2), code=i % 2) + ulp_l4 = ICMPv6Unknown(type=128 + icmp_type_delta, + code=icmp_code) ulp = ulp_l4 p = (Ether(dst=dst_mac, src=src_mac) / IPv6(src=src_ip6, dst=dst_ip6) / ulp / Raw(payload)) else: - ulp_l4 = ICMP(type=8 + (i % 2), code=i % 2) + ulp_l4 = ICMP(type=8 - 8*icmp_type_delta, code=icmp_code) ulp = ulp_l4 p = (Ether(dst=dst_mac, src=src_mac) / IP(src=src_ip4, dst=dst_ip4) / @@ -237,32 +263,29 @@ class TestIpIrb(VppTestCase): else: rule_l4_proto = p[IP].proto - new_rule = { - 'is_permit': is_permit, - 'is_ipv6': p.haslayer(IPv6), - 'src_ip_addr': inet_pton(rule_family, - p[rule_l3_layer].src), - 'src_ip_prefix_len': rule_prefix_len, - 'dst_ip_addr': inet_pton(rule_family, - p[rule_l3_layer].dst), - 'dst_ip_prefix_len': rule_prefix_len, - 'srcport_or_icmptype_first': rule_l4_sport, - 'srcport_or_icmptype_last': rule_l4_sport, - 'dstport_or_icmpcode_first': rule_l4_dport, - 'dstport_or_icmpcode_last': rule_l4_dport, - 'proto': rule_l4_proto, - } + new_rule = AclRule(is_permit=is_permit, proto=rule_l4_proto, + src_prefix=ip_network( + (p[rule_l3_layer].src, rule_prefix_len)), + dst_prefix=ip_network( + (p[rule_l3_layer].dst, rule_prefix_len)), + sport_from=rule_l4_sport, + sport_to=rule_l4_sport, + dport_from=rule_l4_dport, + dport_to=rule_l4_dport) + rules.append(new_rule) - new_rule_permit = new_rule.copy() - new_rule_permit['is_permit'] = 1 + new_rule_permit = copy.copy(new_rule) + new_rule_permit.is_permit = 1 permit_rules.append(new_rule_permit) - new_rule_permit_and_reflect = new_rule.copy() + new_rule_permit_and_reflect = copy.copy(new_rule) if can_reflect_this_packet: - new_rule_permit_and_reflect['is_permit'] = 2 + new_rule_permit_and_reflect.is_permit = 2 else: - new_rule_permit_and_reflect['is_permit'] = is_permit + new_rule_permit_and_reflect.is_permit = is_permit + permit_and_reflect_rules.append(new_rule_permit_and_reflect) + self.logger.info("create_stream pkt#%d: %s" % (i, payload)) return {'stream': pkts, 'rules': rules, @@ -275,7 +298,6 @@ class TestIpIrb(VppTestCase): last_info[i.sw_if_index] = None dst_ip_sw_if_index = dst_ip_if.sw_if_index - return for packet in capture: l3 = IP if packet.haslayer(IP) else IPv6 @@ -291,14 +313,18 @@ class TestIpIrb(VppTestCase): # Scapy IPv6 stuff is too smart for its own good. # So we do this and coerce the ICMP into unknown type if packet.haslayer(UDP): - data = str(packet[UDP][Raw]) + data = scapy.compat.raw(packet[UDP][Raw]) else: if l3 == IP: - data = str(ICMP(str(packet[l3].payload))[Raw]) + data = scapy.compat.raw(ICMP( + scapy.compat.raw(packet[l3].payload))[Raw]) else: - data = str(ICMPv6Unknown(str(packet[l3].payload)).msgbody) + data = scapy.compat.raw(ICMPv6Unknown( + scapy.compat.raw(packet[l3].payload)).msgbody) udp_or_icmp = packet[l3].payload - payload_info = self.payload_to_info(data) + data_obj = Raw(data) + # FIXME: make framework believe we are on object + payload_info = self.payload_to_info(data_obj) packet_index = payload_info.index self.assertEqual(payload_info.dst, dst_ip_sw_if_index) @@ -325,85 +351,71 @@ class TestIpIrb(VppTestCase): if l4 == UDP: self.assertEqual(udp_or_icmp.sport, saved_packet[l4].sport) self.assertEqual(udp_or_icmp.dport, saved_packet[l4].dport) - else: - print("Saved packet is none") # self.assertEqual(ip.dst, host.ip4) # UDP: - def applied_acl_shuffle(self, sw_if_index): - # first collect what ACLs are applied and what they look like - r = self.vapi.acl_interface_list_dump(sw_if_index=sw_if_index) - orig_applied_acls = r[0] - - # we will collect these just to save and generate additional rulesets - orig_acls = [] - for acl_num in orig_applied_acls.acls: - rr = self.vapi.acl_dump(acl_num) - orig_acls.append(rr[0]) + def applied_acl_shuffle(self, acl_if): + saved_n_input = acl_if.n_input + # TOTO: maybe copy each one?? + saved_acls = acl_if.acls # now create a list of all the rules in all ACLs all_rules = [] - for old_acl in orig_acls: - for rule in old_acl.r: - all_rules.append(dict(rule._asdict())) + for old_acl in saved_acls: + for rule in old_acl.rules: + all_rules.append(rule) # Add a few ACLs made from shuffled rules shuffle(all_rules) - reply = self.vapi.acl_add_replace(acl_index=4294967295, - r=all_rules[::2], - tag="shuffle 1. acl") - shuffle_acl_1 = reply.acl_index + acl1 = VppAcl(self, rules=all_rules[::2], tag="shuffle 1. acl") + acl1.add_vpp_config() + shuffle(all_rules) - reply = self.vapi.acl_add_replace(acl_index=4294967295, - r=all_rules[::3], - tag="shuffle 2. acl") - shuffle_acl_2 = reply.acl_index + acl2 = VppAcl(self, rules=all_rules[::3], tag="shuffle 2. acl") + acl2.add_vpp_config() + shuffle(all_rules) - reply = self.vapi.acl_add_replace(acl_index=4294967295, - r=all_rules[::2], - tag="shuffle 3. acl") - shuffle_acl_3 = reply.acl_index + acl3 = VppAcl(self, rules=all_rules[::2], tag="shuffle 3. acl") + acl3.add_vpp_config() # apply the shuffle ACLs in front - input_acls = [shuffle_acl_1, shuffle_acl_2] - output_acls = [shuffle_acl_1, shuffle_acl_2] + input_acls = [acl1, acl2] + output_acls = [acl1, acl2] # add the currently applied ACLs - n_input = orig_applied_acls.n_input - input_acls.extend(orig_applied_acls.acls[:n_input]) - output_acls.extend(orig_applied_acls.acls[n_input:]) + n_input = acl_if.n_input + input_acls.extend(saved_acls[:n_input]) + output_acls.extend(saved_acls[n_input:]) # and the trailing shuffle ACL(s) - input_acls.extend([shuffle_acl_3]) - output_acls.extend([shuffle_acl_3]) + input_acls.extend([acl3]) + output_acls.extend([acl3]) # set the interface ACL list to the result - self.vapi.acl_interface_set_acl_list(sw_if_index=sw_if_index, - n_input=len(input_acls), - acls=input_acls + output_acls) + acl_if.n_input = len(input_acls) + acl_if.acls = input_acls + output_acls + acl_if.add_vpp_config() + # change the ACLs a few times for i in range(1, 10): shuffle(all_rules) - reply = self.vapi.acl_add_replace(acl_index=shuffle_acl_1, - r=all_rules[::1+(i % 2)], - tag="shuffle 1. acl") + acl1.modify_vpp_config(all_rules[::1+(i % 2)]) + shuffle(all_rules) - reply = self.vapi.acl_add_replace(acl_index=shuffle_acl_2, - r=all_rules[::1+(i % 3)], - tag="shuffle 2. acl") + acl2.modify_vpp_config(all_rules[::1+(i % 3)]) + shuffle(all_rules) - reply = self.vapi.acl_add_replace(acl_index=shuffle_acl_2, - r=all_rules[::1+(i % 5)], - tag="shuffle 3. acl") + acl3.modify_vpp_config(all_rules[::1+(i % 5)]) # restore to how it was before and clean up - self.vapi.acl_interface_set_acl_list(sw_if_index=sw_if_index, - n_input=orig_applied_acls.n_input, - acls=orig_applied_acls.acls) - reply = self.vapi.acl_del(acl_index=shuffle_acl_1) - reply = self.vapi.acl_del(acl_index=shuffle_acl_2) - reply = self.vapi.acl_del(acl_index=shuffle_acl_3) + acl_if.n_input = saved_n_input + acl_if.acls = saved_acls + acl_if.add_vpp_config() + + acl1.remove_vpp_config() + acl2.remove_vpp_config() + acl3.remove_vpp_config() def create_acls_for_a_stream(self, stream_dict, test_l2_action, is_reflect): @@ -411,15 +423,14 @@ class TestIpIrb(VppTestCase): r_permit = stream_dict['permit_rules'] r_permit_reflect = stream_dict['permit_and_reflect_rules'] r_action = r_permit_reflect if is_reflect else r - reply = self.vapi.acl_add_replace(acl_index=4294967295, r=r_action, - tag="act. acl") - action_acl_index = reply.acl_index - reply = self.vapi.acl_add_replace(acl_index=4294967295, r=r_permit, - tag="perm. acl") - permit_acl_index = reply.acl_index - return {'L2': action_acl_index if test_l2_action else permit_acl_index, - 'L3': permit_acl_index if test_l2_action else action_acl_index, - 'permit': permit_acl_index, 'action': action_acl_index} + action_acl = VppAcl(self, rules=r_action, tag="act. acl") + action_acl.add_vpp_config() + permit_acl = VppAcl(self, rules=r_permit, tag="perm. acl") + permit_acl.add_vpp_config() + + return {'L2': action_acl if test_l2_action else permit_acl, + 'L3': permit_acl if test_l2_action else action_acl, + 'permit': permit_acl, 'action': action_acl} def apply_acl_ip46_x_to_y(self, bridged_to_routed, test_l2_deny, is_ip6, is_reflect, add_eh): @@ -427,43 +438,50 @@ class TestIpIrb(VppTestCase): """ self.reset_packet_infos() stream_dict = self.create_stream( - self.pg2, self.loop0, - bridged_to_routed, - self.pg_if_packet_sizes, is_ip6, - not is_reflect, False, add_eh) + self.pg2, self.loop0, + bridged_to_routed, + self.pg_if_packet_sizes, is_ip6, + not is_reflect, False, add_eh) stream = stream_dict['stream'] acl_idx = self.create_acls_for_a_stream(stream_dict, test_l2_deny, is_reflect) n_input_l3 = 0 if bridged_to_routed else 1 n_input_l2 = 1 if bridged_to_routed else 0 - self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg2.sw_if_index, - n_input=n_input_l3, - acls=[acl_idx['L3']]) - self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg0.sw_if_index, - n_input=n_input_l2, - acls=[acl_idx['L2']]) - self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg1.sw_if_index, - n_input=n_input_l2, - acls=[acl_idx['L2']]) - self.applied_acl_shuffle(self.pg0.sw_if_index) - self.applied_acl_shuffle(self.pg2.sw_if_index) + + acl_if_pg2 = VppAclInterface(self, sw_if_index=self.pg2.sw_if_index, + n_input=n_input_l3, acls=[acl_idx['L3']]) + acl_if_pg2.add_vpp_config() + + acl_if_pg0 = VppAclInterface(self, sw_if_index=self.pg0.sw_if_index, + n_input=n_input_l2, acls=[acl_idx['L2']]) + acl_if_pg0.add_vpp_config() + + acl_if_pg1 = VppAclInterface(self, sw_if_index=self.pg1.sw_if_index, + n_input=n_input_l2, acls=[acl_idx['L2']]) + acl_if_pg1.add_vpp_config() + + self.applied_acl_shuffle(acl_if_pg0) + self.applied_acl_shuffle(acl_if_pg1) + return {'L2': acl_idx['L2'], 'L3': acl_idx['L3']} def apply_acl_ip46_both_directions_reflect(self, primary_is_bridged_to_routed, - reflect_on_l2, is_ip6, add_eh): + reflect_on_l2, is_ip6, add_eh, + stateful_icmp): primary_is_routed_to_bridged = not primary_is_bridged_to_routed self.reset_packet_infos() stream_dict_fwd = self.create_stream(self.pg2, self.loop0, primary_is_bridged_to_routed, self.pg_if_packet_sizes, is_ip6, - False, False, add_eh) + False, False, add_eh, + stateful_icmp) acl_idx_fwd = self.create_acls_for_a_stream(stream_dict_fwd, reflect_on_l2, True) stream_dict_rev = self.create_stream(self.pg2, self.loop0, not primary_is_bridged_to_routed, self.pg_if_packet_sizes, is_ip6, - True, True, add_eh) + True, True, add_eh, stateful_icmp) # We want the primary action to be "deny" rather than reflect acl_idx_rev = self.create_acls_for_a_stream(stream_dict_rev, reflect_on_l2, False) @@ -488,40 +506,53 @@ class TestIpIrb(VppTestCase): else: outbound_l3_acl = acl_idx_rev['L3'] - self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg2.sw_if_index, - n_input=1, - acls=[inbound_l3_acl, - outbound_l3_acl]) - self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg0.sw_if_index, - n_input=1, - acls=[inbound_l2_acl, - outbound_l2_acl]) - self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg1.sw_if_index, - n_input=1, - acls=[inbound_l2_acl, - outbound_l2_acl]) - self.applied_acl_shuffle(self.pg0.sw_if_index) - self.applied_acl_shuffle(self.pg2.sw_if_index) + acl_if_pg2 = VppAclInterface(self, sw_if_index=self.pg2.sw_if_index, + n_input=1, + acls=[inbound_l3_acl, outbound_l3_acl]) + acl_if_pg2.add_vpp_config() + + acl_if_pg0 = VppAclInterface(self, sw_if_index=self.pg0.sw_if_index, + n_input=1, + acls=[inbound_l2_acl, outbound_l2_acl]) + acl_if_pg0.add_vpp_config() + + acl_if_pg1 = VppAclInterface(self, sw_if_index=self.pg1.sw_if_index, + n_input=1, + acls=[inbound_l2_acl, outbound_l2_acl]) + acl_if_pg1.add_vpp_config() + + self.applied_acl_shuffle(acl_if_pg0) + self.applied_acl_shuffle(acl_if_pg2) def apply_acl_ip46_routed_to_bridged(self, test_l2_deny, is_ip6, is_reflect, add_eh): - self.apply_acl_ip46_x_to_y(False, test_l2_deny, is_ip6, - is_reflect, add_eh) + return self.apply_acl_ip46_x_to_y(False, test_l2_deny, is_ip6, + is_reflect, add_eh) def apply_acl_ip46_bridged_to_routed(self, test_l2_deny, is_ip6, is_reflect, add_eh): - self.apply_acl_ip46_x_to_y(True, test_l2_deny, is_ip6, - is_reflect, add_eh) + return self.apply_acl_ip46_x_to_y(True, test_l2_deny, is_ip6, + is_reflect, add_eh) + + def verify_acl_packet_count(self, acl_idx, packet_count): + matches = self.statistics.get_counter('/acl/%d/matches' % acl_idx) + self.logger.info("stat seg for ACL %d: %s" % (acl_idx, repr(matches))) + total_count = 0 + for m in matches: + for p in m: + total_count = total_count + p['packets'] + self.assertEqual(total_count, packet_count) def run_traffic_ip46_x_to_y(self, bridged_to_routed, test_l2_deny, is_ip6, - is_reflect, is_established, add_eh): + is_reflect, is_established, add_eh, + stateful_icmp=False): self.reset_packet_infos() stream_dict = self.create_stream(self.pg2, self.loop0, bridged_to_routed, self.pg_if_packet_sizes, is_ip6, not is_reflect, is_established, - add_eh) + add_eh, stateful_icmp) stream = stream_dict['stream'] tx_if = self.pg0 if bridged_to_routed else self.pg2 @@ -533,48 +564,67 @@ class TestIpIrb(VppTestCase): packet_count = self.get_packet_count_for_if_idx(self.loop0.sw_if_index) rcvd1 = rx_if.get_capture(packet_count) self.verify_capture(self.loop0, self.pg2, rcvd1, bridged_to_routed) + return len(stream) def run_traffic_ip46_routed_to_bridged(self, test_l2_deny, is_ip6, - is_reflect, is_established, add_eh): - self.run_traffic_ip46_x_to_y(False, test_l2_deny, is_ip6, - is_reflect, is_established, add_eh) + is_reflect, is_established, add_eh, + stateful_icmp=False): + return self.run_traffic_ip46_x_to_y(False, test_l2_deny, is_ip6, + is_reflect, is_established, add_eh, + stateful_icmp) def run_traffic_ip46_bridged_to_routed(self, test_l2_deny, is_ip6, - is_reflect, is_established, add_eh): - self.run_traffic_ip46_x_to_y(True, test_l2_deny, is_ip6, - is_reflect, is_established, add_eh) + is_reflect, is_established, add_eh, + stateful_icmp=False): + return self.run_traffic_ip46_x_to_y(True, test_l2_deny, is_ip6, + is_reflect, is_established, add_eh, + stateful_icmp) def run_test_ip46_routed_to_bridged(self, test_l2_deny, is_ip6, is_reflect, add_eh): - self.apply_acl_ip46_routed_to_bridged(test_l2_deny, - is_ip6, is_reflect, add_eh) - self.run_traffic_ip46_routed_to_bridged(test_l2_deny, is_ip6, - is_reflect, False, add_eh) + acls = self.apply_acl_ip46_routed_to_bridged(test_l2_deny, + is_ip6, is_reflect, + add_eh) + pkts = self.run_traffic_ip46_routed_to_bridged(test_l2_deny, is_ip6, + is_reflect, False, + add_eh) + self.verify_acl_packet_count(acls['L3'].acl_index, pkts) def run_test_ip46_bridged_to_routed(self, test_l2_deny, is_ip6, is_reflect, add_eh): - self.apply_acl_ip46_bridged_to_routed(test_l2_deny, - is_ip6, is_reflect, add_eh) - self.run_traffic_ip46_bridged_to_routed(test_l2_deny, is_ip6, - is_reflect, False, add_eh) + acls = self.apply_acl_ip46_bridged_to_routed(test_l2_deny, + is_ip6, is_reflect, + add_eh) + pkts = self.run_traffic_ip46_bridged_to_routed(test_l2_deny, is_ip6, + is_reflect, False, + add_eh) + self.verify_acl_packet_count(acls['L2'].acl_index, pkts) def run_test_ip46_routed_to_bridged_and_back(self, test_l2_action, - is_ip6, add_eh): + is_ip6, add_eh, + stateful_icmp=False): self.apply_acl_ip46_both_directions_reflect(False, test_l2_action, - is_ip6, add_eh) + is_ip6, add_eh, + stateful_icmp) self.run_traffic_ip46_routed_to_bridged(test_l2_action, is_ip6, - True, False, add_eh) + True, False, add_eh, + stateful_icmp) self.run_traffic_ip46_bridged_to_routed(test_l2_action, is_ip6, - False, True, add_eh) + False, True, add_eh, + stateful_icmp) def run_test_ip46_bridged_to_routed_and_back(self, test_l2_action, - is_ip6, add_eh): + is_ip6, add_eh, + stateful_icmp=False): self.apply_acl_ip46_both_directions_reflect(True, test_l2_action, - is_ip6, add_eh) + is_ip6, add_eh, + stateful_icmp) self.run_traffic_ip46_bridged_to_routed(test_l2_action, is_ip6, - True, False, add_eh) + True, False, add_eh, + stateful_icmp) self.run_traffic_ip46_routed_to_bridged(test_l2_action, is_ip6, - False, True, add_eh) + False, True, add_eh, + stateful_icmp) def test_0000_ip6_irb_1(self): """ ACL plugin prepare""" @@ -759,6 +809,56 @@ class TestIpIrb(VppTestCase): """ ACL IPv4+MF bridged -> routed, L3 ACL permit+reflect""" self.run_test_ip46_bridged_to_routed_and_back(False, False, self.WITH_EH) + # Stateful ACL tests with stateful ICMP + + def test_1401_ip6_irb_1(self): + """ IPv6 routed -> bridged, L2 ACL permit+reflect, ICMP reflect""" + self.run_test_ip46_routed_to_bridged_and_back(True, True, + self.WITHOUT_EH, + self.STATEFUL_ICMP) + + def test_1402_ip6_irb_1(self): + """ IPv6 bridged -> routed, L2 ACL permit+reflect, ICMP reflect""" + self.run_test_ip46_bridged_to_routed_and_back(True, True, + self.WITHOUT_EH, + self.STATEFUL_ICMP) + + def test_1403_ip4_irb_1(self): + """ IPv4 routed -> bridged, L2 ACL permit+reflect, ICMP reflect""" + self.run_test_ip46_routed_to_bridged_and_back(True, False, + self.WITHOUT_EH, + self.STATEFUL_ICMP) + + def test_1404_ip4_irb_1(self): + """ IPv4 bridged -> routed, L2 ACL permit+reflect, ICMP reflect""" + self.run_test_ip46_bridged_to_routed_and_back(True, False, + self.WITHOUT_EH, + self.STATEFUL_ICMP) + + def test_1411_ip6_irb_1(self): + """ IPv6 routed -> bridged, L3 ACL permit+reflect, ICMP reflect""" + self.run_test_ip46_routed_to_bridged_and_back(False, True, + self.WITHOUT_EH, + self.STATEFUL_ICMP) + + def test_1412_ip6_irb_1(self): + """ IPv6 bridged -> routed, L3 ACL permit+reflect, ICMP reflect""" + self.run_test_ip46_bridged_to_routed_and_back(False, True, + self.WITHOUT_EH, + self.STATEFUL_ICMP) + + def test_1413_ip4_irb_1(self): + """ IPv4 routed -> bridged, L3 ACL permit+reflect, ICMP reflect""" + self.run_test_ip46_routed_to_bridged_and_back(False, False, + self.WITHOUT_EH, + self.STATEFUL_ICMP) + + def test_1414_ip4_irb_1(self): + """ IPv4 bridged -> routed, L3 ACL permit+reflect, ICMP reflect""" + self.run_test_ip46_bridged_to_routed_and_back(False, False, + self.WITHOUT_EH, + self.STATEFUL_ICMP) + if __name__ == '__main__': unittest.main(testRunner=VppTestRunner)