| #!/usr/bin/env python3 |
| """ACL IRB Test Case HLD: |
| |
| **config** |
| - L2 MAC learning enabled in l2bd |
| - 2 routed interfaces untagged, bvi (Bridge Virtual Interface) |
| - 2 bridged interfaces in l2bd with bvi |
| |
| **test** |
| - sending ip4 eth pkts between routed interfaces |
| - 2 routed interfaces |
| - 2 bridged interfaces |
| |
| - 64B, 512B, 1518B, 9200B (ether_size) |
| |
| - burst of pkts per interface |
| - 257pkts per burst |
| - routed pkts hitting different FIB entries |
| - bridged pkts hitting different MAC entries |
| |
| **verify** |
| - all packets received correctly |
| |
| """ |
| |
| import copy |
| import unittest |
| from socket import AF_INET, AF_INET6 |
| from random import shuffle |
| from ipaddress import ip_network |
| from config import config |
| |
| import scapy.compat |
| from scapy.packet import Raw |
| from scapy.layers.l2 import Ether |
| from scapy.layers.inet import IP, UDP, ICMP, TCP |
| from scapy.layers.inet6 import IPv6, ICMPv6Unknown |
| from scapy.layers.inet6 import IPv6ExtHdrRouting |
| from scapy.layers.inet6 import IPv6ExtHdrFragment |
| |
| from framework import VppTestCase |
| from asfframework import VppTestRunner |
| from vpp_l2 import L2_PORT_TYPE |
| |
| from vpp_acl import AclRule, VppAcl, VppAclInterface |
| |
| |
| @unittest.skipIf("acl" in config.excluded_plugins, "Exclude ACL plugin tests") |
| class TestACLpluginL2L3(VppTestCase): |
| """TestACLpluginL2L3 Test Case""" |
| |
| @classmethod |
| def setUpClass(cls): |
| """ |
| #. Create BD with MAC learning enabled and put interfaces to this BD. |
| #. Configure IPv4 addresses on loopback interface and routed interface. |
| #. Configure MAC address binding to IPv4 neighbors on loop0. |
| #. Configure MAC address on pg2. |
| #. Loopback BVI interface has remote hosts, one half of hosts are |
| behind pg0 second behind pg1. |
| """ |
| super(TestACLpluginL2L3, cls).setUpClass() |
| |
| cls.pg_if_packet_sizes = [64, 512, 1518, 9018] # packet sizes |
| cls.bd_id = 10 |
| cls.remote_hosts_count = 250 |
| |
| # create 3 pg interfaces, 1 loopback interface |
| cls.create_pg_interfaces(range(3)) |
| cls.create_loopback_interfaces(1) |
| |
| cls.interfaces = list(cls.pg_interfaces) |
| cls.interfaces.extend(cls.lo_interfaces) |
| |
| for i in cls.interfaces: |
| i.admin_up() |
| |
| # Create BD with MAC learning enabled and put interfaces to this BD |
| cls.vapi.sw_interface_set_l2_bridge( |
| 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() |
| cls.loop0.config_ip6() |
| cls.pg2.config_ip4() |
| cls.pg2.config_ip6() |
| |
| # Configure MAC address binding to IPv4 neighbors on loop0 |
| cls.loop0.generate_remote_hosts(cls.remote_hosts_count) |
| cls.loop0.configure_ipv4_neighbors() |
| cls.loop0.configure_ipv6_neighbors() |
| # configure MAC address on pg2 |
| cls.pg2.resolve_arp() |
| cls.pg2.resolve_ndp() |
| |
| 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 <bd_id> detail``, |
| ``show ip neighbors``. |
| """ |
| 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, |
| icmp_stateful=False, |
| ): |
| pkts = [] |
| rules = [] |
| permit_rules = [] |
| permit_and_reflect_rules = [] |
| total_packet_count = 8 |
| for i in range(0, total_packet_count): |
| modulo = (i // 2) % 2 |
| 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] |
| if is_permit == 1: |
| info = self.create_packet_info(src_ip_if, dst_ip_if) |
| payload = self.info_to_payload(info) |
| else: |
| to_be_blocked = False |
| if expect_blocked and not expect_established: |
| to_be_blocked = True |
| if not can_reflect_this_packet: |
| to_be_blocked = True |
| if to_be_blocked: |
| payload = "to be blocked" |
| else: |
| info = self.create_packet_info(src_ip_if, dst_ip_if) |
| payload = self.info_to_payload(info) |
| if reverse: |
| dst_mac = "de:ad:00:00:00:00" |
| src_mac = remote_dst_host._mac |
| dst_ip6 = src_ip_if.remote_ip6 |
| src_ip6 = remote_dst_host.ip6 |
| dst_ip4 = src_ip_if.remote_ip4 |
| src_ip4 = remote_dst_host.ip4 |
| dst_l4 = 1234 + i |
| src_l4 = 4321 + i |
| else: |
| dst_mac = src_ip_if.local_mac |
| src_mac = src_ip_if.remote_mac |
| src_ip6 = src_ip_if.remote_ip6 |
| dst_ip6 = remote_dst_host.ip6 |
| src_ip4 = src_ip_if.remote_ip4 |
| 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 is_udp_packet: |
| if is_ip6: |
| ulp_l4 = UDP(sport=src_l4, dport=dst_l4) |
| if add_extension_header: |
| # prepend some extension headers |
| ulp = ( |
| IPv6ExtHdrRouting() |
| / IPv6ExtHdrRouting() |
| / IPv6ExtHdrFragment(offset=0, m=1) |
| / ulp_l4 |
| ) |
| # uncomment below to test invalid ones |
| # ulp = IPv6ExtHdrRouting(len = 200) / ulp_l4 |
| else: |
| ulp = ulp_l4 |
| p = ( |
| Ether(dst=dst_mac, src=src_mac) |
| / IPv6(src=src_ip6, dst=dst_ip6) |
| / ulp |
| / Raw(payload) |
| ) |
| else: |
| ulp_l4 = UDP(sport=src_l4, dport=dst_l4) |
| # IPv4 does not allow extension headers, |
| # but we rather make it a first fragment |
| flags = 1 if add_extension_header else 0 |
| ulp = ulp_l4 |
| p = ( |
| Ether(dst=dst_mac, src=src_mac) |
| / IP(src=src_ip4, dst=dst_ip4, frag=0, flags=flags) |
| / ulp |
| / Raw(payload) |
| ) |
| elif modulo == 1: |
| if is_ip6: |
| 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 - 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) |
| / ulp |
| / Raw(payload) |
| ) |
| |
| if i % 2 == 1: |
| info.data = p.copy() |
| size = packet_sizes[(i // 2) % len(packet_sizes)] |
| self.extend_packet(p, size) |
| pkts.append(p) |
| |
| rule_family = AF_INET6 if p.haslayer(IPv6) else AF_INET |
| rule_prefix_len = 128 if p.haslayer(IPv6) else 32 |
| rule_l3_layer = IPv6 if p.haslayer(IPv6) else IP |
| |
| if p.haslayer(UDP): |
| rule_l4_sport = p[UDP].sport |
| rule_l4_dport = p[UDP].dport |
| else: |
| if p.haslayer(ICMP): |
| rule_l4_sport = p[ICMP].type |
| rule_l4_dport = p[ICMP].code |
| else: |
| rule_l4_sport = p[ICMPv6Unknown].type |
| rule_l4_dport = p[ICMPv6Unknown].code |
| if p.haslayer(IPv6): |
| rule_l4_proto = ulp_l4.overload_fields[IPv6]["nh"] |
| else: |
| rule_l4_proto = p[IP].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 = copy.copy(new_rule) |
| new_rule_permit.is_permit = 1 |
| permit_rules.append(new_rule_permit) |
| |
| new_rule_permit_and_reflect = copy.copy(new_rule) |
| if can_reflect_this_packet: |
| new_rule_permit_and_reflect.is_permit = 2 |
| else: |
| 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, |
| "permit_rules": permit_rules, |
| "permit_and_reflect_rules": permit_and_reflect_rules, |
| } |
| |
| def verify_capture(self, dst_ip_if, src_ip_if, capture, reverse): |
| last_info = dict() |
| for i in self.interfaces: |
| last_info[i.sw_if_index] = None |
| |
| dst_ip_sw_if_index = dst_ip_if.sw_if_index |
| |
| for packet in capture: |
| l3 = IP if packet.haslayer(IP) else IPv6 |
| ip = packet[l3] |
| if packet.haslayer(UDP): |
| l4 = UDP |
| else: |
| if packet.haslayer(ICMP): |
| l4 = ICMP |
| else: |
| l4 = ICMPv6Unknown |
| |
| # 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 = scapy.compat.raw(packet[UDP][Raw]) |
| else: |
| if l3 == IP: |
| data = scapy.compat.raw( |
| ICMP(scapy.compat.raw(packet[l3].payload))[Raw] |
| ) |
| else: |
| data = scapy.compat.raw( |
| ICMPv6Unknown(scapy.compat.raw(packet[l3].payload)).msgbody |
| ) |
| udp_or_icmp = packet[l3].payload |
| 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) |
| |
| next_info = self.get_next_packet_info_for_interface2( |
| payload_info.src, dst_ip_sw_if_index, last_info[payload_info.src] |
| ) |
| last_info[payload_info.src] = next_info |
| self.assertTrue(next_info is not None) |
| self.assertEqual(packet_index, next_info.index) |
| saved_packet = next_info.data |
| self.assertTrue(next_info is not None) |
| |
| # MAC: src, dst |
| if not reverse: |
| self.assertEqual(packet.src, dst_ip_if.local_mac) |
| host = dst_ip_if.host_by_mac(packet.dst) |
| |
| # IP: src, dst |
| # self.assertEqual(ip.src, src_ip_if.remote_ip4) |
| if saved_packet is not None: |
| self.assertEqual(ip.src, saved_packet[l3].src) |
| self.assertEqual(ip.dst, saved_packet[l3].dst) |
| if l4 == UDP: |
| self.assertEqual(udp_or_icmp.sport, saved_packet[l4].sport) |
| self.assertEqual(udp_or_icmp.dport, saved_packet[l4].dport) |
| # self.assertEqual(ip.dst, host.ip4) |
| |
| # UDP: |
| |
| 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 saved_acls: |
| for rule in old_acl.rules: |
| all_rules.append(rule) |
| |
| # Add a few ACLs made from shuffled rules |
| shuffle(all_rules) |
| acl1 = VppAcl(self, rules=all_rules[::2], tag="shuffle 1. acl") |
| acl1.add_vpp_config() |
| |
| shuffle(all_rules) |
| acl2 = VppAcl(self, rules=all_rules[::3], tag="shuffle 2. acl") |
| acl2.add_vpp_config() |
| |
| shuffle(all_rules) |
| acl3 = VppAcl(self, rules=all_rules[::2], tag="shuffle 3. acl") |
| acl3.add_vpp_config() |
| |
| # apply the shuffle ACLs in front |
| input_acls = [acl1, acl2] |
| output_acls = [acl1, acl2] |
| |
| # add the currently applied ACLs |
| 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([acl3]) |
| output_acls.extend([acl3]) |
| |
| # set the interface ACL list to the result |
| 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) |
| acl1.modify_vpp_config(all_rules[:: 1 + (i % 2)]) |
| |
| shuffle(all_rules) |
| acl2.modify_vpp_config(all_rules[:: 1 + (i % 3)]) |
| |
| shuffle(all_rules) |
| acl3.modify_vpp_config(all_rules[:: 1 + (i % 5)]) |
| |
| # restore to how it was before and clean up |
| 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): |
| r = stream_dict["rules"] |
| 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 |
| 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 |
| ): |
| """Apply the ACLs""" |
| 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, |
| ) |
| 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 |
| |
| 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, 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, |
| 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, |
| 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 |
| ) |
| |
| if primary_is_bridged_to_routed: |
| inbound_l2_acl = acl_idx_fwd["L2"] |
| else: |
| inbound_l2_acl = acl_idx_rev["L2"] |
| |
| if primary_is_routed_to_bridged: |
| outbound_l2_acl = acl_idx_fwd["L2"] |
| else: |
| outbound_l2_acl = acl_idx_rev["L2"] |
| |
| if primary_is_routed_to_bridged: |
| inbound_l3_acl = acl_idx_fwd["L3"] |
| else: |
| inbound_l3_acl = acl_idx_rev["L3"] |
| |
| if primary_is_bridged_to_routed: |
| outbound_l3_acl = acl_idx_fwd["L3"] |
| else: |
| outbound_l3_acl = acl_idx_rev["L3"] |
| |
| 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 |
| ): |
| 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 |
| ): |
| 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, |
| 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, |
| stateful_icmp, |
| ) |
| stream = stream_dict["stream"] |
| |
| tx_if = self.pg0 if bridged_to_routed else self.pg2 |
| rx_if = self.pg2 if bridged_to_routed else self.pg0 |
| |
| tx_if.add_stream(stream) |
| self.pg_enable_capture(self.pg_interfaces) |
| self.pg_start() |
| 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, |
| 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, |
| 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): |
| 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): |
| 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, stateful_icmp=False |
| ): |
| self.apply_acl_ip46_both_directions_reflect( |
| False, test_l2_action, is_ip6, add_eh, stateful_icmp |
| ) |
| self.run_traffic_ip46_routed_to_bridged( |
| test_l2_action, is_ip6, True, False, add_eh, stateful_icmp |
| ) |
| self.run_traffic_ip46_bridged_to_routed( |
| test_l2_action, is_ip6, False, True, add_eh, stateful_icmp |
| ) |
| |
| def run_test_ip46_bridged_to_routed_and_back( |
| self, test_l2_action, is_ip6, add_eh, stateful_icmp=False |
| ): |
| self.apply_acl_ip46_both_directions_reflect( |
| True, test_l2_action, is_ip6, add_eh, stateful_icmp |
| ) |
| self.run_traffic_ip46_bridged_to_routed( |
| test_l2_action, is_ip6, True, False, add_eh, stateful_icmp |
| ) |
| self.run_traffic_ip46_routed_to_bridged( |
| test_l2_action, is_ip6, False, True, add_eh, stateful_icmp |
| ) |
| |
| def test_0000_ip6_irb_1(self): |
| """ACL plugin prepare""" |
| if not self.vpp_dead: |
| cmd = "set acl-plugin session timeout udp idle 2000" |
| self.logger.info(self.vapi.ppcli(cmd)) |
| # uncomment to not skip past the routing header |
| # and watch the EH tests fail |
| # self.logger.info(self.vapi.ppcli( |
| # "set acl-plugin skip-ipv6-extension-header 43 0")) |
| # uncomment to test the session limit (stateful tests will fail) |
| # self.logger.info(self.vapi.ppcli( |
| # "set acl-plugin session table max-entries 1")) |
| # new datapath is the default, but just in case |
| # self.logger.info(self.vapi.ppcli( |
| # "set acl-plugin l2-datapath new")) |
| # If you want to see some tests fail, uncomment the next line |
| # self.logger.info(self.vapi.ppcli( |
| # "set acl-plugin l2-datapath old")) |
| |
| def test_0001_ip6_irb_1(self): |
| """ACL IPv6 routed -> bridged, L2 ACL deny""" |
| self.run_test_ip46_routed_to_bridged(True, True, False, self.WITHOUT_EH) |
| |
| def test_0002_ip6_irb_1(self): |
| """ACL IPv6 routed -> bridged, L3 ACL deny""" |
| self.run_test_ip46_routed_to_bridged(False, True, False, self.WITHOUT_EH) |
| |
| def test_0003_ip4_irb_1(self): |
| """ACL IPv4 routed -> bridged, L2 ACL deny""" |
| self.run_test_ip46_routed_to_bridged(True, False, False, self.WITHOUT_EH) |
| |
| def test_0004_ip4_irb_1(self): |
| """ACL IPv4 routed -> bridged, L3 ACL deny""" |
| self.run_test_ip46_routed_to_bridged(False, False, False, self.WITHOUT_EH) |
| |
| def test_0005_ip6_irb_1(self): |
| """ACL IPv6 bridged -> routed, L2 ACL deny""" |
| self.run_test_ip46_bridged_to_routed(True, True, False, self.WITHOUT_EH) |
| |
| def test_0006_ip6_irb_1(self): |
| """ACL IPv6 bridged -> routed, L3 ACL deny""" |
| self.run_test_ip46_bridged_to_routed(False, True, False, self.WITHOUT_EH) |
| |
| def test_0007_ip6_irb_1(self): |
| """ACL IPv4 bridged -> routed, L2 ACL deny""" |
| self.run_test_ip46_bridged_to_routed(True, False, False, self.WITHOUT_EH) |
| |
| def test_0008_ip6_irb_1(self): |
| """ACL IPv4 bridged -> routed, L3 ACL deny""" |
| self.run_test_ip46_bridged_to_routed(False, False, False, self.WITHOUT_EH) |
| |
| # Stateful ACL tests |
| def test_0101_ip6_irb_1(self): |
| """ACL IPv6 routed -> bridged, L2 ACL permit+reflect""" |
| self.run_test_ip46_routed_to_bridged_and_back(True, True, self.WITHOUT_EH) |
| |
| def test_0102_ip6_irb_1(self): |
| """ACL IPv6 bridged -> routed, L2 ACL permit+reflect""" |
| self.run_test_ip46_bridged_to_routed_and_back(True, True, self.WITHOUT_EH) |
| |
| def test_0103_ip6_irb_1(self): |
| """ACL IPv4 routed -> bridged, L2 ACL permit+reflect""" |
| self.run_test_ip46_routed_to_bridged_and_back(True, False, self.WITHOUT_EH) |
| |
| def test_0104_ip6_irb_1(self): |
| """ACL IPv4 bridged -> routed, L2 ACL permit+reflect""" |
| self.run_test_ip46_bridged_to_routed_and_back(True, False, self.WITHOUT_EH) |
| |
| def test_0111_ip6_irb_1(self): |
| """ACL IPv6 routed -> bridged, L3 ACL permit+reflect""" |
| self.run_test_ip46_routed_to_bridged_and_back(False, True, self.WITHOUT_EH) |
| |
| def test_0112_ip6_irb_1(self): |
| """ACL IPv6 bridged -> routed, L3 ACL permit+reflect""" |
| self.run_test_ip46_bridged_to_routed_and_back(False, True, self.WITHOUT_EH) |
| |
| def test_0113_ip6_irb_1(self): |
| """ACL IPv4 routed -> bridged, L3 ACL permit+reflect""" |
| self.run_test_ip46_routed_to_bridged_and_back(False, False, self.WITHOUT_EH) |
| |
| def test_0114_ip6_irb_1(self): |
| """ACL IPv4 bridged -> routed, L3 ACL permit+reflect""" |
| self.run_test_ip46_bridged_to_routed_and_back(False, False, self.WITHOUT_EH) |
| |
| # A block of tests with extension headers |
| |
| def test_1001_ip6_irb_1(self): |
| """ACL IPv6+EH routed -> bridged, L2 ACL deny""" |
| self.run_test_ip46_routed_to_bridged(True, True, False, self.WITH_EH) |
| |
| def test_1002_ip6_irb_1(self): |
| """ACL IPv6+EH routed -> bridged, L3 ACL deny""" |
| self.run_test_ip46_routed_to_bridged(False, True, False, self.WITH_EH) |
| |
| def test_1005_ip6_irb_1(self): |
| """ACL IPv6+EH bridged -> routed, L2 ACL deny""" |
| self.run_test_ip46_bridged_to_routed(True, True, False, self.WITH_EH) |
| |
| def test_1006_ip6_irb_1(self): |
| """ACL IPv6+EH bridged -> routed, L3 ACL deny""" |
| self.run_test_ip46_bridged_to_routed(False, True, False, self.WITH_EH) |
| |
| def test_1101_ip6_irb_1(self): |
| """ACL IPv6+EH routed -> bridged, L2 ACL permit+reflect""" |
| self.run_test_ip46_routed_to_bridged_and_back(True, True, self.WITH_EH) |
| |
| def test_1102_ip6_irb_1(self): |
| """ACL IPv6+EH bridged -> routed, L2 ACL permit+reflect""" |
| self.run_test_ip46_bridged_to_routed_and_back(True, True, self.WITH_EH) |
| |
| def test_1111_ip6_irb_1(self): |
| """ACL IPv6+EH routed -> bridged, L3 ACL permit+reflect""" |
| self.run_test_ip46_routed_to_bridged_and_back(False, True, self.WITH_EH) |
| |
| def test_1112_ip6_irb_1(self): |
| """ACL IPv6+EH bridged -> routed, L3 ACL permit+reflect""" |
| self.run_test_ip46_bridged_to_routed_and_back(False, True, self.WITH_EH) |
| |
| # IPv4 with "MF" bit set |
| |
| def test_1201_ip6_irb_1(self): |
| """ACL IPv4+MF routed -> bridged, L2 ACL deny""" |
| self.run_test_ip46_routed_to_bridged(True, False, False, self.WITH_EH) |
| |
| def test_1202_ip6_irb_1(self): |
| """ACL IPv4+MF routed -> bridged, L3 ACL deny""" |
| self.run_test_ip46_routed_to_bridged(False, False, False, self.WITH_EH) |
| |
| def test_1205_ip6_irb_1(self): |
| """ACL IPv4+MF bridged -> routed, L2 ACL deny""" |
| self.run_test_ip46_bridged_to_routed(True, False, False, self.WITH_EH) |
| |
| def test_1206_ip6_irb_1(self): |
| """ACL IPv4+MF bridged -> routed, L3 ACL deny""" |
| self.run_test_ip46_bridged_to_routed(False, False, False, self.WITH_EH) |
| |
| def test_1301_ip6_irb_1(self): |
| """ACL IPv4+MF routed -> bridged, L2 ACL permit+reflect""" |
| self.run_test_ip46_routed_to_bridged_and_back(True, False, self.WITH_EH) |
| |
| def test_1302_ip6_irb_1(self): |
| """ACL IPv4+MF bridged -> routed, L2 ACL permit+reflect""" |
| self.run_test_ip46_bridged_to_routed_and_back(True, False, self.WITH_EH) |
| |
| def test_1311_ip6_irb_1(self): |
| """ACL IPv4+MF routed -> bridged, L3 ACL permit+reflect""" |
| self.run_test_ip46_routed_to_bridged_and_back(False, False, self.WITH_EH) |
| |
| def test_1312_ip6_irb_1(self): |
| """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) |