from scapy.layers.inet import IP, UDP, ICMP, TCP
from scapy.layers.inet6 import IPv6, ICMPv6Unknown, ICMPv6EchoRequest
from scapy.layers.inet6 import ICMPv6EchoReply, IPv6ExtHdrRouting
+from scapy.layers.inet6 import IPv6ExtHdrFragment
from framework import VppTestCase, VppTestRunner
import time
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"))
-
- def api_acl_add_replace(self, acl_index, r, count, tag="",
- expected_retval=0):
- """Add/replace an ACL
-
- :param int acl_index: ACL index to replace, 4294967295 to create new.
- :param acl_rule r: ACL rules array.
- :param str tag: symbolic tag (description) for this ACL.
- :param int count: number of rules.
- """
- return self.vapi.api(self.vapi.papi.acl_add_replace,
- {'acl_index': acl_index,
- 'r': r,
- 'count': count,
- 'tag': tag
- }, expected_retval=expected_retval)
-
- def api_acl_interface_set_acl_list(self, sw_if_index, count, n_input, acls,
- expected_retval=0):
- return self.vapi.api(self.vapi.papi.acl_interface_set_acl_list,
- {'sw_if_index': sw_if_index,
- 'count': count,
- 'n_input': n_input,
- 'acls': acls
- }, expected_retval=expected_retval)
-
- def api_acl_dump(self, acl_index, expected_retval=0):
- return self.vapi.api(self.vapi.papi.acl_dump,
- {'acl_index': acl_index},
- expected_retval=expected_retval)
+ 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,
if add_extension_header:
# prepend some extension headers
ulp = (IPv6ExtHdrRouting() / IPv6ExtHdrRouting() /
- IPv6ExtHdrRouting() / ulp_l4)
+ IPv6ExtHdrFragment(offset=0, m=1) / ulp_l4)
# uncomment below to test invalid ones
# ulp = IPv6ExtHdrRouting(len = 200) / ulp_l4
else:
Raw(payload))
else:
ulp_l4 = UDP(sport=src_l4, dport=dst_l4)
- # IPv4 does not allow extension headers
+ # 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) /
+ IP(src=src_ip4, dst=dst_ip4, frag=0, flags=flags) /
ulp /
Raw(payload))
elif modulo == 1:
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.api_acl_add_replace(acl_index=4294967295, r=r_action,
- count=len(r_action), tag="action acl")
+ reply = self.vapi.acl_add_replace(acl_index=4294967295, r=r_action,
+ tag="act. acl")
action_acl_index = reply.acl_index
- reply = self.api_acl_add_replace(acl_index=4294967295, r=r_permit,
- count=len(r_permit), tag="permit acl")
+ 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,
is_reflect)
n_input_l3 = 0 if bridged_to_routed else 1
n_input_l2 = 1 if bridged_to_routed else 0
- self.api_acl_interface_set_acl_list(sw_if_index=self.pg2.sw_if_index,
- count=1,
- n_input=n_input_l3,
- acls=[acl_idx['L3']])
- self.api_acl_interface_set_acl_list(sw_if_index=self.pg0.sw_if_index,
- count=1,
- n_input=n_input_l2,
- acls=[acl_idx['L2']])
- self.api_acl_interface_set_acl_list(sw_if_index=self.pg1.sw_if_index,
- count=1,
- n_input=n_input_l2,
- acls=[acl_idx['L2']])
+ 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']])
def apply_acl_ip46_both_directions_reflect(self,
primary_is_bridged_to_routed,
else:
outbound_l3_acl = acl_idx_rev['L3']
- self.api_acl_interface_set_acl_list(sw_if_index=self.pg2.sw_if_index,
- count=2,
- n_input=1,
- acls=[inbound_l3_acl,
- outbound_l3_acl])
- self.api_acl_interface_set_acl_list(sw_if_index=self.pg0.sw_if_index,
- count=2,
- n_input=1,
- acls=[inbound_l2_acl,
- outbound_l2_acl])
- self.api_acl_interface_set_acl_list(sw_if_index=self.pg1.sw_if_index,
- count=2,
- n_input=1,
- acls=[inbound_l2_acl,
- outbound_l2_acl])
+ 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])
def apply_acl_ip46_routed_to_bridged(self, test_l2_deny, is_ip6,
is_reflect, add_eh):
self.run_test_ip46_bridged_to_routed_and_back(False, True,
self.WITH_EH)
- # Old datapath group
- def test_8900_ip6_irb_1(self):
- """ ACL plugin set old L2 datapath"""
- if not self.vpp_dead:
- cmd = "set acl-plugin l2-datapath old"
- self.logger.info(self.vapi.ppcli(cmd))
-
- def test_8901_ip6_irb_1(self):
- """ ACL IPv6 routed -> bridged, L2 ACL deny"""
- self.run_test_ip46_routed_to_bridged(True, True, False,
- self.WITHOUT_EH)
+ # IPv4 with "MF" bit set
- def test_8902_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_8903_ip4_irb_1(self):
- """ ACL IPv4 routed -> bridged, L2 ACL deny"""
+ 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.WITHOUT_EH)
+ self.WITH_EH)
- def test_8904_ip4_irb_1(self):
- """ ACL IPv4 routed -> bridged, L3 ACL deny"""
+ 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.WITHOUT_EH)
-
- def test_8905_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_8906_ip6_irb_1(self):
- """ ACL IPv6 bridged -> routed, L3 ACL deny """
- self.run_test_ip46_bridged_to_routed(False, True, False,
- self.WITHOUT_EH)
+ self.WITH_EH)
- def test_8907_ip6_irb_1(self):
- """ ACL IPv4 bridged -> routed, L2 ACL deny """
+ 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.WITHOUT_EH)
+ self.WITH_EH)
- def test_8908_ip6_irb_1(self):
- """ ACL IPv4 bridged -> routed, L3 ACL deny """
+ 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.WITHOUT_EH)
+ 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)
if __name__ == '__main__':
unittest.main(testRunner=VppTestRunner)