-#!/usr/bin/env python
+#!/usr/bin/env python3
"""ACL IRB Test Case HLD:
**config**
"""
+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
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):
#. 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
# 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()
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 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 = []
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]
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:
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) /
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,
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
# 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)
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):
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):
"""
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)
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
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"""
""" 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)