X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftest_acl_plugin_macip.py;h=5edd7b0325859d3caef66a09455dc95d4bfb01ad;hb=bd46907b3f27b480e5f8b0be9cb4c5b5f1fa08ab;hp=a8df833cd263ccf2075fa1ae0c08a8ab6f50487b;hpb=563a8531cbc3e7c222e3e4841637b94b65637caa;p=vpp.git diff --git a/test/test_acl_plugin_macip.py b/test/test_acl_plugin_macip.py index a8df833cd26..5edd7b03258 100644 --- a/test/test_acl_plugin_macip.py +++ b/test/test_acl_plugin_macip.py @@ -1,12 +1,17 @@ -#!/usr/bin/env python +#!/usr/bin/env python3 +from __future__ import print_function """ACL plugin - MACIP tests """ +import binascii +import ipaddress import random +from socket import inet_ntop, inet_pton, AF_INET, AF_INET6 +from struct import pack, unpack import re import unittest +from ipaddress import ip_network, IPv4Network, IPv6Network -from socket import inet_ntop, inet_pton, AF_INET, AF_INET6 -from struct import * +import scapy.compat from scapy.packet import Raw from scapy.layers.l2 import Ether from scapy.layers.inet import IP, UDP @@ -14,8 +19,12 @@ from scapy.layers.inet6 import IPv6 from framework import VppTestCase, VppTestRunner, running_extended_tests from vpp_lo_interface import VppLoInterface -from vpp_papi_provider import L2_VTR_OP -from vpp_sub_interface import VppSubInterface, VppDot1QSubint, VppDot1ADSubint +from vpp_l2 import L2_PORT_TYPE +from vpp_sub_interface import L2_VTR_OP, VppSubInterface, VppDot1QSubint, \ + VppDot1ADSubint +from vpp_acl import AclRule, VppAcl, VppAclInterface, VppEtypeWhitelist, \ + VppMacipAclInterface, VppMacipAcl, MacipRule +from vpp_papi import MACAddress class MethodHolder(VppTestCase): @@ -63,14 +72,14 @@ class MethodHolder(VppTestCase): try: # create 4 pg interfaces, 1 loopback interface cls.create_pg_interfaces(range(4)) - cls.create_loopback_interfaces(range(1)) + cls.create_loopback_interfaces(1) # create 2 subinterfaces cls.subifs = [ - VppDot1QSubint(cls, cls.pg1, 10), - VppDot1ADSubint(cls, cls.pg2, 20, 300, 400), - VppDot1QSubint(cls, cls.pg3, 30), - VppDot1ADSubint(cls, cls.pg3, 40, 600, 700)] + VppDot1QSubint(cls, cls.pg1, 10), + VppDot1ADSubint(cls, cls.pg2, 20, 300, 400), + VppDot1QSubint(cls, cls.pg3, 30), + VppDot1ADSubint(cls, cls.pg3, 40, 600, 700)] cls.subifs[0].set_vtr(L2_VTR_OP.L2_POP_1, inner=10, push1q=1) @@ -90,15 +99,16 @@ class MethodHolder(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) + 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( - cls.pg0.sw_if_index, bd_id=cls.bd_id) + rx_sw_if_index=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.pg1.sw_if_index, bd_id=cls.bd_id) cls.vapi.sw_interface_set_l2_bridge( - cls.subifs[0].sw_if_index, bd_id=cls.bd_id) + rx_sw_if_index=cls.subifs[0].sw_if_index, bd_id=cls.bd_id) cls.vapi.sw_interface_set_l2_bridge( - cls.subifs[1].sw_if_index, bd_id=cls.bd_id) + rx_sw_if_index=cls.subifs[1].sw_if_index, bd_id=cls.bd_id) # Configure IPv4/6 addresses on loop interface and routed interface cls.loop0.config_ip4() @@ -142,56 +152,57 @@ class MethodHolder(VppTestCase): cls.subifs[3].remote_hosts = cls.loop0.remote_hosts[175:] except Exception: - super(TestMACIP, cls).tearDownClass() + super(MethodHolder, cls).tearDownClass() raise + @classmethod + def tearDownClass(cls): + super(MethodHolder, cls).tearDownClass() + def setUp(self): super(MethodHolder, self).setUp() self.reset_packet_infos() - del self.ACLS[:] - def tearDown(self): - """ - Show various debug prints after each test. - """ - super(MethodHolder, self).tearDown() - if not self.vpp_dead: - self.logger.info(self.vapi.ppcli("show interface address")) - self.logger.info(self.vapi.ppcli("show hardware")) - self.logger.info(self.vapi.ppcli("sh acl-plugin macip acl")) - self.logger.info(self.vapi.ppcli("sh acl-plugin macip interface")) - self.logger.info(self.vapi.ppcli("sh classify tables verbose")) - self.logger.info(self.vapi.ppcli("sh acl-plugin acl")) - self.logger.info(self.vapi.ppcli("sh acl-plugin interface")) - self.logger.info(self.vapi.ppcli("sh acl-plugin tables")) - # print self.vapi.ppcli("show interface address") - # print self.vapi.ppcli("show hardware") - # print self.vapi.ppcli("sh acl-plugin macip interface") - # print self.vapi.ppcli("sh acl-plugin macip acl") - self.delete_acls() + def show_commands_at_teardown(self): + self.logger.info(self.vapi.ppcli("show interface address")) + self.logger.info(self.vapi.ppcli("show hardware")) + self.logger.info(self.vapi.ppcli("sh acl-plugin macip acl")) + self.logger.info(self.vapi.ppcli("sh acl-plugin macip interface")) + self.logger.info(self.vapi.ppcli("sh classify tables verbose")) + self.logger.info(self.vapi.ppcli("sh acl-plugin acl")) + self.logger.info(self.vapi.ppcli("sh acl-plugin interface")) + self.logger.info(self.vapi.ppcli("sh acl-plugin tables")) + # print(self.vapi.ppcli("show interface address")) + # print(self.vapi.ppcli("show hardware")) + # print(self.vapi.ppcli("sh acl-plugin macip interface")) + # print(self.vapi.ppcli("sh acl-plugin macip acl")) def macip_acl_dump_debug(self): acls = self.vapi.macip_acl_dump() if self.DEBUG: for acl in acls: - print "ACL #"+str(acl.acl_index) + # print("ACL #"+str(acl.acl_index)) for r in acl.r: rule = "ACTION" if r.is_permit == 1: rule = "PERMIT" elif r.is_permit == 0: rule = "DENY " - print " IP6" if r.is_ipv6 else " IP4", \ - rule, \ - r.src_mac.encode('hex'), \ - r.src_mac_mask.encode('hex'),\ - unpack('<16B', r.src_ip_addr), \ - r.src_ip_prefix_len + """ + print(" IP6" if r.is_ipv6 else " IP4", + rule, + binascii.hexlify(r.src_mac), + binascii.hexlify(r.src_mac_mask), + unpack('<16B', r.src_ip_addr), + r.src_ip_prefix_len) + """ return acls def create_rules(self, mac_type=EXACT_MAC, ip_type=EXACT_IP, - acl_count=1, rules_count=[1]): + acl_count=1, rules_count=None): acls = [] + if rules_count is None: + rules_count = [1] src_mac = int("220000dead00", 16) for acl in range(2, (acl_count+1) * 2): rules = [] @@ -208,7 +219,7 @@ class MethodHolder(VppTestCase): ip6 = [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0] prefix_len4 = 0 prefix_len6 = 0 - rules_count[(acl / 2) - 1] = 1 + rules_count[int((acl / 2) - 1)] = 1 else: prefix_len4 = 24 prefix_len6 = 64 @@ -225,7 +236,7 @@ class MethodHolder(VppTestCase): ip = ip6 if is_ip6 else ip4 ip_len = prefix_len6 if is_ip6 else prefix_len4 - for i in range(0, rules_count[(acl / 2) - 1]): + for i in range(0, (rules_count[int((acl / 2) - 1)])): src_mac += 16777217 if mac_type == self.WILD_MAC: mac = "00:00:00:00:00:00" @@ -233,7 +244,8 @@ class MethodHolder(VppTestCase): mac = ':'.join(re.findall('..', '{:02x}'.format( src_mac))[:3])+":00:00:00" else: - mac = ':'.join(re.findall('..', '{:02x}'.format(src_mac))) + mac = ':'.join(re.findall( + '..', '{:02x}'.format(src_mac))) if ip_type == self.EXACT_IP: ip4[3] = random.randint(100, 200) @@ -241,18 +253,16 @@ class MethodHolder(VppTestCase): elif ip_type == self.SUBNET_IP: ip4[2] = random.randint(100, 200) ip4[3] = 0 - ip6[8] = random.randint(100, 200) + ip6[7] = random.randint(100, 200) ip6[15] = 0 - ip_pack = '' + ip_pack = b'' for j in range(0, len(ip)): ip_pack += pack(' 0: continue if is_permit: - macip_rule = ({ - 'is_permit': is_permit, - 'is_ipv6': is_ip6, - 'src_ip_addr': ip_rule, - 'src_ip_prefix_len': prefix_len, - 'src_mac': mac_rule.replace(':', '').decode('hex'), - 'src_mac_mask': mac_mask.replace(':', '').decode('hex')}) + macip_rule = MacipRule( + is_permit=is_permit, + src_prefix=ip_network( + (ip_rule, prefix_len)), + src_mac=MACAddress(mac_rule).packed, + src_mac_mask=MACAddress(mac_mask).packed) macip_rules.append(macip_rule) # deny all other packets if not (mac_type == self.WILD_MAC and ip_type == self.WILD_IP): - macip_rule = ({'is_permit': 0, - 'is_ipv6': is_ip6, - 'src_ip_addr': "", - 'src_ip_prefix_len': 0, - 'src_mac': "", - 'src_mac_mask': ""}) + network = IPv6Network((0, 0)) if is_ip6 else IPv4Network((0, 0)) + macip_rule = MacipRule( + is_permit=0, + src_prefix=network, + src_mac=MACAddress("00:00:00:00:00:00").packed, + src_mac_mask=MACAddress("00:00:00:00:00:00").packed) macip_rules.append(macip_rule) - acl_rule = {'is_permit': 0, - 'is_ipv6': is_ip6} + network = IPv6Network((0, 0)) if is_ip6 else IPv4Network((0, 0)) + acl_rule = AclRule(is_permit=0, src_prefix=network, dst_prefix=network, + sport_from=0, sport_to=0, dport_from=0, dport_to=0) acl_rules.append(acl_rule) return {'stream': packets, 'macip_rules': macip_rules, @@ -566,22 +565,22 @@ class MethodHolder(VppTestCase): # p_l3 = IPv6 if is_ip6 else IP # if self.DEBUG: # for p in stream: - # print p[Ether].src, p[Ether].dst, p[p_l3].src, p[p_l3].dst + # print(p[Ether].src, p[Ether].dst, p[p_l3].src, p[p_l3].dst) # # acls = self.macip_acl_dump_debug() # TODO : verify # for acl in acls: # for r in acl.r: - # print r.src_mac.encode('hex'), \ - # r.src_mac_mask.encode('hex'),\ + # print(binascii.hexlify(r.src_mac), \ + # binascii.hexlify(r.src_mac_mask),\ # unpack('<16B', r.src_ip_addr), \ - # r.src_ip_prefix_len + # r.src_ip_prefix_len) # # for p in capture: - # print p[Ether].src, p[Ether].dst, p[p_l3].src, p[p_l3].dst - # data = p[Raw].load.split(':',1)[1] - # print p[p_l3].src, data + # print(p[Ether].src, p[Ether].dst, p[p_l3].src, p[p_l3].dst + # data = p[Raw].load.split(':',1)[1]) + # print(p[p_l3].src, data) def run_traffic(self, mac_type, ip_type, traffic, is_ip6, packets, do_not_expected_capture=False, tags=None, @@ -627,36 +626,32 @@ class MethodHolder(VppTestCase): if apply_rules: if isMACIP: - reply = self.vapi.macip_acl_add(test_dict['macip_rules']) + self.acl = VppMacipAcl(self, rules=test_dict['macip_rules']) else: - reply = self.vapi.acl_add_replace(acl_index=4294967295, - r=test_dict['acl_rules']) - self.assertEqual(reply.retval, 0) - acl_index = reply.acl_index + self.acl = VppAcl(self, rules=test_dict['acl_rules']) + self.acl.add_vpp_config() if isMACIP: - self.vapi.macip_acl_interface_add_del( - sw_if_index=tx_if.sw_if_index, - acl_index=acl_index) - reply = self.vapi.macip_acl_interface_get() - self.assertEqual(reply.acls[tx_if.sw_if_index], acl_index) - self.ACLS.append(reply.acls[tx_if.sw_if_index]) + self.acl_if = VppMacipAclInterface( + self, sw_if_index=tx_if.sw_if_index, acls=[self.acl]) + self.acl_if.add_vpp_config() + + dump = self.acl_if.dump() + self.assertTrue(dump) + self.assertEqual(dump[0].acls[0], self.acl.acl_index) else: - self.vapi.acl_interface_add_del( - sw_if_index=tx_if.sw_if_index, acl_index=acl_index) + self.acl_if = VppAclInterface( + self, sw_if_index=tx_if.sw_if_index, n_input=1, + acls=[self.acl]) + self.acl_if.add_vpp_config() else: - self.vapi.macip_acl_interface_add_del( - sw_if_index=tx_if.sw_if_index, - acl_index=0) - if try_replace: + if hasattr(self, "acl_if"): + self.acl_if.remove_vpp_config() + if try_replace and hasattr(self, "acl"): if isMACIP: - reply = self.vapi.macip_acl_add_replace( - test_dict['macip_rules'], - acl_index) + self.acl.modify_vpp_config(test_dict['macip_rules']) else: - reply = self.vapi.acl_add_replace(acl_index=acl_index, - r=test_dict['acl_rules']) - self.assertEqual(reply.retval, 0) + self.acl.modify_vpp_config(test_dict['acl_rules']) if not isinstance(src_if, VppSubInterface): tx_if.add_stream(test_dict['stream']) @@ -676,9 +671,10 @@ class MethodHolder(VppTestCase): self.get_packet_count_for_if_idx(dst_if.sw_if_index)) self.verify_capture(test_dict['stream'], capture, is_ip6) if not isMACIP: - self.vapi.acl_interface_add_del(sw_if_index=tx_if.sw_if_index, - acl_index=acl_index, is_add=0) - self.vapi.acl_del(acl_index) + if hasattr(self, "acl_if"): + self.acl_if.remove_vpp_config() + if hasattr(self, "acl"): + self.acl.remove_vpp_config() def run_test_acls(self, mac_type, ip_type, acl_count, rules_count, traffic=None, ip=None): @@ -693,6 +689,14 @@ class MethodHolder(VppTestCase): class TestMACIP_IP4(MethodHolder): """MACIP with IP4 traffic""" + @classmethod + def setUpClass(cls): + super(TestMACIP_IP4, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(TestMACIP_IP4, cls).tearDownClass() + def test_acl_bridged_ip4_exactMAC_exactIP(self): """ IP4 MACIP exactMAC|exactIP ACL bridged traffic """ @@ -827,6 +831,14 @@ class TestMACIP_IP4(MethodHolder): class TestMACIP_IP6(MethodHolder): """MACIP with IP6 traffic""" + @classmethod + def setUpClass(cls): + super(TestMACIP_IP6, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(TestMACIP_IP6, cls).tearDownClass() + def test_acl_bridged_ip6_exactMAC_exactIP(self): """ IP6 MACIP exactMAC|exactIP ACL bridged traffic """ @@ -965,6 +977,14 @@ class TestMACIP_IP6(MethodHolder): class TestMACIP(MethodHolder): """MACIP Tests""" + @classmethod + def setUpClass(cls): + super(TestMACIP, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(TestMACIP, cls).tearDownClass() + def test_acl_1_2(self): """ MACIP ACL with 2 entries """ @@ -1036,17 +1056,13 @@ class TestMACIP(MethodHolder): r1 = self.create_rules(acl_count=3, rules_count=[2, 2, 2]) r2 = self.create_rules(mac_type=self.OUI_MAC, ip_type=self.SUBNET_IP) - self.apply_macip_rules(r1) + macip_acls = self.apply_macip_rules(r1) acls_before = self.macip_acl_dump_debug() # replace acls #2, #3 with new - reply = self.vapi.macip_acl_add_replace(r2[0], 2) - self.assertEqual(reply.retval, 0) - self.assertEqual(reply.acl_index, 2) - reply = self.vapi.macip_acl_add_replace(r2[1], 3) - self.assertEqual(reply.retval, 0) - self.assertEqual(reply.acl_index, 3) + macip_acls[2].modify_vpp_config(r2[0]) + macip_acls[3].modify_vpp_config(r2[1]) acls_after = self.macip_acl_dump_debug() @@ -1077,21 +1093,25 @@ class TestMACIP(MethodHolder): intf_count = len(self.interfaces)+1 intf = [] - self.apply_macip_rules(self.create_rules(acl_count=3, - rules_count=[3, 5, 4])) + macip_alcs = self.apply_macip_rules( + self.create_rules(acl_count=3, rules_count=[3, 5, 4])) - intf.append(VppLoInterface(self, 0)) - intf.append(VppLoInterface(self, 1)) + intf.append(VppLoInterface(self)) + intf.append(VppLoInterface(self)) sw_if_index0 = intf[0].sw_if_index - self.vapi.macip_acl_interface_add_del(sw_if_index0, 1) + macip_acl_if0 = VppMacipAclInterface( + self, sw_if_index=sw_if_index0, acls=[macip_alcs[1]]) + macip_acl_if0.add_vpp_config() reply = self.vapi.macip_acl_interface_get() self.assertEqual(reply.count, intf_count+1) self.assertEqual(reply.acls[sw_if_index0], 1) sw_if_index1 = intf[1].sw_if_index - self.vapi.macip_acl_interface_add_del(sw_if_index1, 0) + macip_acl_if1 = VppMacipAclInterface( + self, sw_if_index=sw_if_index1, acls=[macip_alcs[0]]) + macip_acl_if1.add_vpp_config() reply = self.vapi.macip_acl_interface_get() self.assertEqual(reply.count, intf_count+2) @@ -1103,12 +1123,16 @@ class TestMACIP(MethodHolder): self.assertEqual(reply.acls[sw_if_index0], 4294967295) self.assertEqual(reply.acls[sw_if_index1], 0) - intf.append(VppLoInterface(self, 2)) - intf.append(VppLoInterface(self, 3)) + intf.append(VppLoInterface(self)) + intf.append(VppLoInterface(self)) sw_if_index2 = intf[2].sw_if_index sw_if_index3 = intf[3].sw_if_index - self.vapi.macip_acl_interface_add_del(sw_if_index2, 1) - self.vapi.macip_acl_interface_add_del(sw_if_index3, 1) + macip_acl_if2 = VppMacipAclInterface( + self, sw_if_index=sw_if_index2, acls=[macip_alcs[1]]) + macip_acl_if2.add_vpp_config() + macip_acl_if3 = VppMacipAclInterface( + self, sw_if_index=sw_if_index3, acls=[macip_alcs[1]]) + macip_acl_if3.add_vpp_config() reply = self.vapi.macip_acl_interface_get() self.assertEqual(reply.count, intf_count+3) @@ -1141,6 +1165,14 @@ class TestMACIP(MethodHolder): class TestACL_dot1q_bridged(MethodHolder): """ACL on dot1q bridged subinterfaces Tests""" + @classmethod + def setUpClass(cls): + super(TestACL_dot1q_bridged, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(TestACL_dot1q_bridged, cls).tearDownClass() + def test_acl_bridged_ip4_subif_dot1q(self): """ IP4 ACL SubIf Dot1Q bridged traffic""" self.run_traffic(self.EXACT_MAC, self.EXACT_IP, self.BRIDGED, @@ -1155,6 +1187,14 @@ class TestACL_dot1q_bridged(MethodHolder): class TestACL_dot1ad_bridged(MethodHolder): """ACL on dot1ad bridged subinterfaces Tests""" + @classmethod + def setUpClass(cls): + super(TestACL_dot1ad_bridged, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(TestACL_dot1ad_bridged, cls).tearDownClass() + def test_acl_bridged_ip4_subif_dot1ad(self): """ IP4 ACL SubIf Dot1AD bridged traffic""" self.run_traffic(self.EXACT_MAC, self.EXACT_IP, self.BRIDGED, @@ -1169,6 +1209,14 @@ class TestACL_dot1ad_bridged(MethodHolder): class TestACL_dot1q_routed(MethodHolder): """ACL on dot1q routed subinterfaces Tests""" + @classmethod + def setUpClass(cls): + super(TestACL_dot1q_routed, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(TestACL_dot1q_routed, cls).tearDownClass() + def test_acl_routed_ip4_subif_dot1q(self): """ IP4 ACL SubIf Dot1Q routed traffic""" self.run_traffic(self.EXACT_MAC, self.EXACT_IP, self.ROUTED, @@ -1195,6 +1243,14 @@ class TestACL_dot1q_routed(MethodHolder): class TestACL_dot1ad_routed(MethodHolder): """ACL on dot1ad routed subinterfaces Tests""" + @classmethod + def setUpClass(cls): + super(TestACL_dot1ad_routed, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(TestACL_dot1ad_routed, cls).tearDownClass() + def test_acl_routed_ip6_subif_dot1ad(self): """ IP6 ACL SubIf Dot1AD routed traffic""" self.run_traffic(self.EXACT_MAC, self.EXACT_IP, self.ROUTED,