| #!/usr/bin/env python |
| """ACL plugin Test Case HLD: |
| """ |
| |
| import unittest |
| import random |
| |
| from scapy.packet import Raw |
| from scapy.layers.l2 import Ether |
| from scapy.layers.inet import IP, TCP, UDP, ICMP |
| from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest |
| from scapy.layers.inet6 import IPv6ExtHdrFragment |
| from framework import VppTestCase, VppTestRunner |
| from util import Host, ppp |
| |
| from vpp_lo_interface import VppLoInterface |
| |
| |
| class TestACLplugin(VppTestCase): |
| """ ACL plugin Test Case """ |
| |
| # traffic types |
| IP = 0 |
| ICMP = 1 |
| |
| # IP version |
| IPRANDOM = -1 |
| IPV4 = 0 |
| IPV6 = 1 |
| |
| # rule types |
| DENY = 0 |
| PERMIT = 1 |
| |
| # supported protocols |
| proto = [[6, 17], [1, 58]] |
| proto_map = {1: 'ICMP', 58: 'ICMPv6EchoRequest', 6: 'TCP', 17: 'UDP'} |
| ICMPv4 = 0 |
| ICMPv6 = 1 |
| TCP = 0 |
| UDP = 1 |
| PROTO_ALL = 0 |
| |
| # port ranges |
| PORTS_ALL = -1 |
| PORTS_RANGE = 0 |
| PORTS_RANGE_2 = 1 |
| udp_sport_from = 10 |
| udp_sport_to = udp_sport_from + 5 |
| udp_dport_from = 20000 |
| udp_dport_to = udp_dport_from + 5000 |
| tcp_sport_from = 30 |
| tcp_sport_to = tcp_sport_from + 5 |
| tcp_dport_from = 40000 |
| tcp_dport_to = tcp_dport_from + 5000 |
| |
| udp_sport_from_2 = 90 |
| udp_sport_to_2 = udp_sport_from_2 + 5 |
| udp_dport_from_2 = 30000 |
| udp_dport_to_2 = udp_dport_from_2 + 5000 |
| tcp_sport_from_2 = 130 |
| tcp_sport_to_2 = tcp_sport_from_2 + 5 |
| tcp_dport_from_2 = 20000 |
| tcp_dport_to_2 = tcp_dport_from_2 + 5000 |
| |
| icmp4_type = 8 # echo request |
| icmp4_code = 3 |
| icmp6_type = 128 # echo request |
| icmp6_code = 3 |
| |
| icmp4_type_2 = 8 |
| icmp4_code_from_2 = 5 |
| icmp4_code_to_2 = 20 |
| icmp6_type_2 = 128 |
| icmp6_code_from_2 = 8 |
| icmp6_code_to_2 = 42 |
| |
| # Test variables |
| bd_id = 1 |
| |
| @classmethod |
| def setUpClass(cls): |
| """ |
| Perform standard class setup (defined by class method setUpClass in |
| class VppTestCase) before running the test case, set test case related |
| variables and configure VPP. |
| """ |
| super(TestACLplugin, cls).setUpClass() |
| |
| try: |
| # Create 2 pg interfaces |
| cls.create_pg_interfaces(range(2)) |
| |
| # Packet flows mapping pg0 -> pg1, pg2 etc. |
| cls.flows = dict() |
| cls.flows[cls.pg0] = [cls.pg1] |
| |
| # Packet sizes |
| cls.pg_if_packet_sizes = [64, 512, 1518, 9018] |
| |
| # Create BD with MAC learning and unknown unicast flooding disabled |
| # and put interfaces to this BD |
| cls.vapi.bridge_domain_add_del(bd_id=cls.bd_id, uu_flood=1, |
| learn=1) |
| for pg_if in cls.pg_interfaces: |
| cls.vapi.sw_interface_set_l2_bridge(pg_if.sw_if_index, |
| bd_id=cls.bd_id) |
| |
| # Set up all interfaces |
| for i in cls.pg_interfaces: |
| i.admin_up() |
| |
| # Mapping between packet-generator index and lists of test hosts |
| cls.hosts_by_pg_idx = dict() |
| for pg_if in cls.pg_interfaces: |
| cls.hosts_by_pg_idx[pg_if.sw_if_index] = [] |
| |
| # Create list of deleted hosts |
| cls.deleted_hosts_by_pg_idx = dict() |
| for pg_if in cls.pg_interfaces: |
| cls.deleted_hosts_by_pg_idx[pg_if.sw_if_index] = [] |
| |
| # warm-up the mac address tables |
| # self.warmup_test() |
| count = 16 |
| start = 0 |
| n_int = len(cls.pg_interfaces) |
| macs_per_if = count / n_int |
| i = -1 |
| for pg_if in cls.pg_interfaces: |
| i += 1 |
| start_nr = macs_per_if * i + start |
| end_nr = count + start if i == (n_int - 1) \ |
| else macs_per_if * (i + 1) + start |
| hosts = cls.hosts_by_pg_idx[pg_if.sw_if_index] |
| for j in range(start_nr, end_nr): |
| host = Host( |
| "00:00:00:ff:%02x:%02x" % (pg_if.sw_if_index, j), |
| "172.17.1%02x.%u" % (pg_if.sw_if_index, j), |
| "2017:dead:%02x::%u" % (pg_if.sw_if_index, j)) |
| hosts.append(host) |
| |
| except Exception: |
| super(TestACLplugin, cls).tearDownClass() |
| raise |
| |
| def setUp(self): |
| super(TestACLplugin, self).setUp() |
| self.reset_packet_infos() |
| |
| def tearDown(self): |
| """ |
| Show various debug prints after each test. |
| """ |
| super(TestACLplugin, self).tearDown() |
| if not self.vpp_dead: |
| cli = "show vlib graph l2-input-feat-arc" |
| self.logger.info(self.vapi.ppcli(cli)) |
| cli = "show vlib graph l2-input-feat-arc-end" |
| self.logger.info(self.vapi.ppcli(cli)) |
| cli = "show vlib graph l2-output-feat-arc" |
| self.logger.info(self.vapi.ppcli(cli)) |
| cli = "show vlib graph l2-output-feat-arc-end" |
| self.logger.info(self.vapi.ppcli(cli)) |
| self.logger.info(self.vapi.ppcli("show l2fib verbose")) |
| self.logger.info(self.vapi.ppcli("show acl-plugin acl")) |
| self.logger.info(self.vapi.ppcli("show acl-plugin interface")) |
| self.logger.info(self.vapi.ppcli("show acl-plugin tables")) |
| self.logger.info(self.vapi.ppcli("show bridge-domain %s detail" |
| % self.bd_id)) |
| |
| def create_rule(self, ip=0, permit_deny=0, ports=PORTS_ALL, proto=-1, |
| s_prefix=0, s_ip='\x00\x00\x00\x00', |
| d_prefix=0, d_ip='\x00\x00\x00\x00'): |
| if proto == -1: |
| return |
| if ports == self.PORTS_ALL: |
| sport_from = 0 |
| dport_from = 0 |
| sport_to = 65535 if proto != 1 and proto != 58 else 255 |
| dport_to = sport_to |
| elif ports == self.PORTS_RANGE: |
| if proto == 1: |
| sport_from = self.icmp4_type |
| sport_to = self.icmp4_type |
| dport_from = self.icmp4_code |
| dport_to = self.icmp4_code |
| elif proto == 58: |
| sport_from = self.icmp6_type |
| sport_to = self.icmp6_type |
| dport_from = self.icmp6_code |
| dport_to = self.icmp6_code |
| elif proto == self.proto[self.IP][self.TCP]: |
| sport_from = self.tcp_sport_from |
| sport_to = self.tcp_sport_to |
| dport_from = self.tcp_dport_from |
| dport_to = self.tcp_dport_to |
| elif proto == self.proto[self.IP][self.UDP]: |
| sport_from = self.udp_sport_from |
| sport_to = self.udp_sport_to |
| dport_from = self.udp_dport_from |
| dport_to = self.udp_dport_to |
| elif ports == self.PORTS_RANGE_2: |
| if proto == 1: |
| sport_from = self.icmp4_type_2 |
| sport_to = self.icmp4_type_2 |
| dport_from = self.icmp4_code_from_2 |
| dport_to = self.icmp4_code_to_2 |
| elif proto == 58: |
| sport_from = self.icmp6_type_2 |
| sport_to = self.icmp6_type_2 |
| dport_from = self.icmp6_code_from_2 |
| dport_to = self.icmp6_code_to_2 |
| elif proto == self.proto[self.IP][self.TCP]: |
| sport_from = self.tcp_sport_from_2 |
| sport_to = self.tcp_sport_to_2 |
| dport_from = self.tcp_dport_from_2 |
| dport_to = self.tcp_dport_to_2 |
| elif proto == self.proto[self.IP][self.UDP]: |
| sport_from = self.udp_sport_from_2 |
| sport_to = self.udp_sport_to_2 |
| dport_from = self.udp_dport_from_2 |
| dport_to = self.udp_dport_to_2 |
| else: |
| sport_from = ports |
| sport_to = ports |
| dport_from = ports |
| dport_to = ports |
| |
| rule = ({'is_permit': permit_deny, 'is_ipv6': ip, 'proto': proto, |
| 'srcport_or_icmptype_first': sport_from, |
| 'srcport_or_icmptype_last': sport_to, |
| 'src_ip_prefix_len': s_prefix, |
| 'src_ip_addr': s_ip, |
| 'dstport_or_icmpcode_first': dport_from, |
| 'dstport_or_icmpcode_last': dport_to, |
| 'dst_ip_prefix_len': d_prefix, |
| 'dst_ip_addr': d_ip}) |
| return rule |
| |
| def apply_rules(self, rules, tag=''): |
| reply = self.vapi.acl_add_replace(acl_index=4294967295, r=rules, |
| tag=tag) |
| self.logger.info("Dumped ACL: " + str( |
| self.vapi.acl_dump(reply.acl_index))) |
| # Apply a ACL on the interface as inbound |
| for i in self.pg_interfaces: |
| self.vapi.acl_interface_set_acl_list(sw_if_index=i.sw_if_index, |
| n_input=1, |
| acls=[reply.acl_index]) |
| return |
| |
| def apply_rules_to(self, rules, tag='', sw_if_index=0xFFFFFFFF): |
| reply = self.vapi.acl_add_replace(acl_index=4294967295, r=rules, |
| tag=tag) |
| self.logger.info("Dumped ACL: " + str( |
| self.vapi.acl_dump(reply.acl_index))) |
| # Apply a ACL on the interface as inbound |
| self.vapi.acl_interface_set_acl_list(sw_if_index=sw_if_index, |
| n_input=1, |
| acls=[reply.acl_index]) |
| return |
| |
| def etype_whitelist(self, whitelist, n_input): |
| # Apply whitelists on all the interfaces |
| for i in self.pg_interfaces: |
| # checkstyle can't read long names. Help them. |
| fun = self.vapi.acl_interface_set_etype_whitelist |
| fun(sw_if_index=i.sw_if_index, n_input=n_input, |
| whitelist=whitelist) |
| return |
| |
| def create_upper_layer(self, packet_index, proto, ports=0): |
| p = self.proto_map[proto] |
| if p == 'UDP': |
| if ports == 0: |
| return UDP(sport=random.randint(self.udp_sport_from, |
| self.udp_sport_to), |
| dport=random.randint(self.udp_dport_from, |
| self.udp_dport_to)) |
| else: |
| return UDP(sport=ports, dport=ports) |
| elif p == 'TCP': |
| if ports == 0: |
| return TCP(sport=random.randint(self.tcp_sport_from, |
| self.tcp_sport_to), |
| dport=random.randint(self.tcp_dport_from, |
| self.tcp_dport_to)) |
| else: |
| return TCP(sport=ports, dport=ports) |
| return '' |
| |
| def create_stream(self, src_if, packet_sizes, traffic_type=0, ipv6=0, |
| proto=-1, ports=0, fragments=False, |
| pkt_raw=True, etype=-1): |
| """ |
| Create input packet stream for defined interface using hosts or |
| deleted_hosts list. |
| |
| :param object src_if: Interface to create packet stream for. |
| :param list packet_sizes: List of required packet sizes. |
| :param traffic_type: 1: ICMP packet, 2: IPv6 with EH, 0: otherwise. |
| :return: Stream of packets. |
| """ |
| pkts = [] |
| if self.flows.__contains__(src_if): |
| src_hosts = self.hosts_by_pg_idx[src_if.sw_if_index] |
| for dst_if in self.flows[src_if]: |
| dst_hosts = self.hosts_by_pg_idx[dst_if.sw_if_index] |
| n_int = len(dst_hosts) * len(src_hosts) |
| for i in range(0, n_int): |
| dst_host = dst_hosts[i / len(src_hosts)] |
| src_host = src_hosts[i % len(src_hosts)] |
| pkt_info = self.create_packet_info(src_if, dst_if) |
| if ipv6 == 1: |
| pkt_info.ip = 1 |
| elif ipv6 == 0: |
| pkt_info.ip = 0 |
| else: |
| pkt_info.ip = random.choice([0, 1]) |
| if proto == -1: |
| pkt_info.proto = random.choice(self.proto[self.IP]) |
| else: |
| pkt_info.proto = proto |
| payload = self.info_to_payload(pkt_info) |
| p = Ether(dst=dst_host.mac, src=src_host.mac) |
| if etype > 0: |
| p = Ether(dst=dst_host.mac, |
| src=src_host.mac, |
| type=etype) |
| if pkt_info.ip: |
| p /= IPv6(dst=dst_host.ip6, src=src_host.ip6) |
| if fragments: |
| p /= IPv6ExtHdrFragment(offset=64, m=1) |
| else: |
| if fragments: |
| p /= IP(src=src_host.ip4, dst=dst_host.ip4, |
| flags=1, frag=64) |
| else: |
| p /= IP(src=src_host.ip4, dst=dst_host.ip4) |
| if traffic_type == self.ICMP: |
| if pkt_info.ip: |
| p /= ICMPv6EchoRequest(type=self.icmp6_type, |
| code=self.icmp6_code) |
| else: |
| p /= ICMP(type=self.icmp4_type, |
| code=self.icmp4_code) |
| else: |
| p /= self.create_upper_layer(i, pkt_info.proto, ports) |
| if pkt_raw: |
| p /= Raw(payload) |
| pkt_info.data = p.copy() |
| if pkt_raw: |
| size = random.choice(packet_sizes) |
| self.extend_packet(p, size) |
| pkts.append(p) |
| return pkts |
| |
| def verify_capture(self, pg_if, capture, |
| traffic_type=0, ip_type=0, etype=-1): |
| """ |
| Verify captured input packet stream for defined interface. |
| |
| :param object pg_if: Interface to verify captured packet stream for. |
| :param list capture: Captured packet stream. |
| :param traffic_type: 1: ICMP packet, 2: IPv6 with EH, 0: otherwise. |
| """ |
| last_info = dict() |
| for i in self.pg_interfaces: |
| last_info[i.sw_if_index] = None |
| dst_sw_if_index = pg_if.sw_if_index |
| for packet in capture: |
| if etype > 0: |
| if packet[Ether].type != etype: |
| self.logger.error(ppp("Unexpected ethertype in packet:", |
| packet)) |
| else: |
| continue |
| try: |
| # Raw data for ICMPv6 are stored in ICMPv6EchoRequest.data |
| if traffic_type == self.ICMP and ip_type == self.IPV6: |
| payload_info = self.payload_to_info( |
| packet[ICMPv6EchoRequest].data) |
| payload = packet[ICMPv6EchoRequest] |
| else: |
| payload_info = self.payload_to_info(str(packet[Raw])) |
| payload = packet[self.proto_map[payload_info.proto]] |
| except: |
| self.logger.error(ppp("Unexpected or invalid packet " |
| "(outside network):", packet)) |
| raise |
| |
| if ip_type != 0: |
| self.assertEqual(payload_info.ip, ip_type) |
| if traffic_type == self.ICMP: |
| try: |
| if payload_info.ip == 0: |
| self.assertEqual(payload.type, self.icmp4_type) |
| self.assertEqual(payload.code, self.icmp4_code) |
| else: |
| self.assertEqual(payload.type, self.icmp6_type) |
| self.assertEqual(payload.code, self.icmp6_code) |
| except: |
| self.logger.error(ppp("Unexpected or invalid packet " |
| "(outside network):", packet)) |
| raise |
| else: |
| try: |
| ip_version = IPv6 if payload_info.ip == 1 else IP |
| |
| ip = packet[ip_version] |
| packet_index = payload_info.index |
| |
| self.assertEqual(payload_info.dst, dst_sw_if_index) |
| self.logger.debug("Got packet on port %s: src=%u (id=%u)" % |
| (pg_if.name, payload_info.src, |
| packet_index)) |
| next_info = self.get_next_packet_info_for_interface2( |
| payload_info.src, dst_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 |
| # Check standard fields |
| self.assertEqual(ip.src, saved_packet[ip_version].src) |
| self.assertEqual(ip.dst, saved_packet[ip_version].dst) |
| p = self.proto_map[payload_info.proto] |
| if p == 'TCP': |
| tcp = packet[TCP] |
| self.assertEqual(tcp.sport, saved_packet[ |
| TCP].sport) |
| self.assertEqual(tcp.dport, saved_packet[ |
| TCP].dport) |
| elif p == 'UDP': |
| udp = packet[UDP] |
| self.assertEqual(udp.sport, saved_packet[ |
| UDP].sport) |
| self.assertEqual(udp.dport, saved_packet[ |
| UDP].dport) |
| except: |
| self.logger.error(ppp("Unexpected or invalid packet:", |
| packet)) |
| raise |
| for i in self.pg_interfaces: |
| remaining_packet = self.get_next_packet_info_for_interface2( |
| i, dst_sw_if_index, last_info[i.sw_if_index]) |
| self.assertTrue( |
| remaining_packet is None, |
| "Port %u: Packet expected from source %u didn't arrive" % |
| (dst_sw_if_index, i.sw_if_index)) |
| |
| def run_traffic_no_check(self): |
| # Test |
| # Create incoming packet streams for packet-generator interfaces |
| for i in self.pg_interfaces: |
| if self.flows.__contains__(i): |
| pkts = self.create_stream(i, self.pg_if_packet_sizes) |
| if len(pkts) > 0: |
| i.add_stream(pkts) |
| |
| # Enable packet capture and start packet sending |
| self.pg_enable_capture(self.pg_interfaces) |
| self.pg_start() |
| |
| def run_verify_test(self, traffic_type=0, ip_type=0, proto=-1, ports=0, |
| frags=False, pkt_raw=True, etype=-1): |
| # Test |
| # Create incoming packet streams for packet-generator interfaces |
| pkts_cnt = 0 |
| for i in self.pg_interfaces: |
| if self.flows.__contains__(i): |
| pkts = self.create_stream(i, self.pg_if_packet_sizes, |
| traffic_type, ip_type, proto, ports, |
| frags, pkt_raw, etype) |
| if len(pkts) > 0: |
| i.add_stream(pkts) |
| pkts_cnt += len(pkts) |
| |
| # Enable packet capture and start packet sendingself.IPV |
| self.pg_enable_capture(self.pg_interfaces) |
| self.pg_start() |
| self.logger.info("sent packets count: %d" % pkts_cnt) |
| |
| # Verify |
| # Verify outgoing packet streams per packet-generator interface |
| for src_if in self.pg_interfaces: |
| if self.flows.__contains__(src_if): |
| for dst_if in self.flows[src_if]: |
| capture = dst_if.get_capture(pkts_cnt) |
| self.logger.info("Verifying capture on interface %s" % |
| dst_if.name) |
| self.verify_capture(dst_if, capture, |
| traffic_type, ip_type, etype) |
| |
| def run_verify_negat_test(self, traffic_type=0, ip_type=0, proto=-1, |
| ports=0, frags=False, etype=-1): |
| # Test |
| pkts_cnt = 0 |
| self.reset_packet_infos() |
| for i in self.pg_interfaces: |
| if self.flows.__contains__(i): |
| pkts = self.create_stream(i, self.pg_if_packet_sizes, |
| traffic_type, ip_type, proto, ports, |
| frags, True, etype) |
| if len(pkts) > 0: |
| i.add_stream(pkts) |
| pkts_cnt += len(pkts) |
| |
| # Enable packet capture and start packet sending |
| self.pg_enable_capture(self.pg_interfaces) |
| self.pg_start() |
| self.logger.info("sent packets count: %d" % pkts_cnt) |
| |
| # Verify |
| # Verify outgoing packet streams per packet-generator interface |
| for src_if in self.pg_interfaces: |
| if self.flows.__contains__(src_if): |
| for dst_if in self.flows[src_if]: |
| self.logger.info("Verifying capture on interface %s" % |
| dst_if.name) |
| capture = dst_if.get_capture(0) |
| self.assertEqual(len(capture), 0) |
| |
| def test_0000_warmup_test(self): |
| """ ACL plugin version check; learn MACs |
| """ |
| reply = self.vapi.papi.acl_plugin_get_version() |
| self.assertEqual(reply.major, 1) |
| self.logger.info("Working with ACL plugin version: %d.%d" % ( |
| reply.major, reply.minor)) |
| # minor version changes are non breaking |
| # self.assertEqual(reply.minor, 0) |
| |
| def test_0001_acl_create(self): |
| """ ACL create/delete test |
| """ |
| |
| self.logger.info("ACLP_TEST_START_0001") |
| # Add an ACL |
| r = [{'is_permit': 1, 'is_ipv6': 0, 'proto': 17, |
| 'srcport_or_icmptype_first': 1234, |
| 'srcport_or_icmptype_last': 1235, |
| 'src_ip_prefix_len': 0, |
| 'src_ip_addr': '\x00\x00\x00\x00', |
| 'dstport_or_icmpcode_first': 1234, |
| 'dstport_or_icmpcode_last': 1234, |
| 'dst_ip_addr': '\x00\x00\x00\x00', |
| 'dst_ip_prefix_len': 0}] |
| # Test 1: add a new ACL |
| reply = self.vapi.acl_add_replace(acl_index=4294967295, r=r, |
| tag="permit 1234") |
| self.assertEqual(reply.retval, 0) |
| # The very first ACL gets #0 |
| self.assertEqual(reply.acl_index, 0) |
| first_acl = reply.acl_index |
| rr = self.vapi.acl_dump(reply.acl_index) |
| self.logger.info("Dumped ACL: " + str(rr)) |
| self.assertEqual(len(rr), 1) |
| # We should have the same number of ACL entries as we had asked |
| self.assertEqual(len(rr[0].r), len(r)) |
| # The rules should be the same. But because the submitted and returned |
| # are different types, we need to iterate over rules and keys to get |
| # to basic values. |
| for i_rule in range(0, len(r) - 1): |
| for rule_key in r[i_rule]: |
| self.assertEqual(rr[0].r[i_rule][rule_key], |
| r[i_rule][rule_key]) |
| |
| # Add a deny-1234 ACL |
| r_deny = [{'is_permit': 0, 'is_ipv6': 0, 'proto': 17, |
| 'srcport_or_icmptype_first': 1234, |
| 'srcport_or_icmptype_last': 1235, |
| 'src_ip_prefix_len': 0, |
| 'src_ip_addr': '\x00\x00\x00\x00', |
| 'dstport_or_icmpcode_first': 1234, |
| 'dstport_or_icmpcode_last': 1234, |
| 'dst_ip_addr': '\x00\x00\x00\x00', |
| 'dst_ip_prefix_len': 0}, |
| {'is_permit': 1, 'is_ipv6': 0, 'proto': 17, |
| 'srcport_or_icmptype_first': 0, |
| 'srcport_or_icmptype_last': 0, |
| 'src_ip_prefix_len': 0, |
| 'src_ip_addr': '\x00\x00\x00\x00', |
| 'dstport_or_icmpcode_first': 0, |
| 'dstport_or_icmpcode_last': 0, |
| 'dst_ip_addr': '\x00\x00\x00\x00', |
| 'dst_ip_prefix_len': 0}] |
| |
| reply = self.vapi.acl_add_replace(acl_index=4294967295, r=r_deny, |
| tag="deny 1234;permit all") |
| self.assertEqual(reply.retval, 0) |
| # The second ACL gets #1 |
| self.assertEqual(reply.acl_index, 1) |
| second_acl = reply.acl_index |
| |
| # Test 2: try to modify a nonexistent ACL |
| reply = self.vapi.acl_add_replace(acl_index=432, r=r, |
| tag="FFFF:FFFF", expected_retval=-6) |
| self.assertEqual(reply.retval, -6) |
| # The ACL number should pass through |
| self.assertEqual(reply.acl_index, 432) |
| # apply an ACL on an interface inbound, try to delete ACL, must fail |
| self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg0.sw_if_index, |
| n_input=1, |
| acls=[first_acl]) |
| reply = self.vapi.acl_del(acl_index=first_acl, expected_retval=-142) |
| # Unapply an ACL and then try to delete it - must be ok |
| self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg0.sw_if_index, |
| n_input=0, |
| acls=[]) |
| reply = self.vapi.acl_del(acl_index=first_acl, expected_retval=0) |
| |
| # apply an ACL on an interface outbound, try to delete ACL, must fail |
| self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg0.sw_if_index, |
| n_input=0, |
| acls=[second_acl]) |
| reply = self.vapi.acl_del(acl_index=second_acl, expected_retval=-143) |
| # Unapply the ACL and then try to delete it - must be ok |
| self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg0.sw_if_index, |
| n_input=0, |
| acls=[]) |
| reply = self.vapi.acl_del(acl_index=second_acl, expected_retval=0) |
| |
| # try to apply a nonexistent ACL - must fail |
| self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg0.sw_if_index, |
| n_input=1, |
| acls=[first_acl], |
| expected_retval=-6) |
| |
| self.logger.info("ACLP_TEST_FINISH_0001") |
| |
| def test_0002_acl_permit_apply(self): |
| """ permit ACL apply test |
| """ |
| self.logger.info("ACLP_TEST_START_0002") |
| |
| rules = [] |
| rules.append(self.create_rule(self.IPV4, self.PERMIT, |
| 0, self.proto[self.IP][self.UDP])) |
| rules.append(self.create_rule(self.IPV4, self.PERMIT, |
| 0, self.proto[self.IP][self.TCP])) |
| |
| # Apply rules |
| self.apply_rules(rules, "permit per-flow") |
| |
| # Traffic should still pass |
| self.run_verify_test(self.IP, self.IPV4, -1) |
| self.logger.info("ACLP_TEST_FINISH_0002") |
| |
| def test_0003_acl_deny_apply(self): |
| """ deny ACL apply test |
| """ |
| self.logger.info("ACLP_TEST_START_0003") |
| # Add a deny-flows ACL |
| rules = [] |
| rules.append(self.create_rule(self.IPV4, self.DENY, |
| self.PORTS_ALL, self.proto[self.IP][self.UDP])) |
| # Permit ip any any in the end |
| rules.append(self.create_rule(self.IPV4, self.PERMIT, |
| self.PORTS_ALL, 0)) |
| |
| # Apply rules |
| self.apply_rules(rules, "deny per-flow;permit all") |
| |
| # Traffic should not pass |
| self.run_verify_negat_test(self.IP, self.IPV4, |
| self.proto[self.IP][self.UDP]) |
| self.logger.info("ACLP_TEST_FINISH_0003") |
| # self.assertEqual(1, 0) |
| |
| def test_0004_vpp624_permit_icmpv4(self): |
| """ VPP_624 permit ICMPv4 |
| """ |
| self.logger.info("ACLP_TEST_START_0004") |
| |
| # Add an ACL |
| rules = [] |
| rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE, |
| self.proto[self.ICMP][self.ICMPv4])) |
| # deny ip any any in the end |
| rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0)) |
| |
| # Apply rules |
| self.apply_rules(rules, "permit icmpv4") |
| |
| # Traffic should still pass |
| self.run_verify_test(self.ICMP, self.IPV4, |
| self.proto[self.ICMP][self.ICMPv4]) |
| |
| self.logger.info("ACLP_TEST_FINISH_0004") |
| |
| def test_0005_vpp624_permit_icmpv6(self): |
| """ VPP_624 permit ICMPv6 |
| """ |
| self.logger.info("ACLP_TEST_START_0005") |
| |
| # Add an ACL |
| rules = [] |
| rules.append(self.create_rule(self.IPV6, self.PERMIT, self.PORTS_RANGE, |
| self.proto[self.ICMP][self.ICMPv6])) |
| # deny ip any any in the end |
| rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_ALL, 0)) |
| |
| # Apply rules |
| self.apply_rules(rules, "permit icmpv6") |
| |
| # Traffic should still pass |
| self.run_verify_test(self.ICMP, self.IPV6, |
| self.proto[self.ICMP][self.ICMPv6]) |
| |
| self.logger.info("ACLP_TEST_FINISH_0005") |
| |
| def test_0006_vpp624_deny_icmpv4(self): |
| """ VPP_624 deny ICMPv4 |
| """ |
| self.logger.info("ACLP_TEST_START_0006") |
| # Add an ACL |
| rules = [] |
| rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE, |
| self.proto[self.ICMP][self.ICMPv4])) |
| # permit ip any any in the end |
| rules.append(self.create_rule(self.IPV4, self.PERMIT, |
| self.PORTS_ALL, 0)) |
| |
| # Apply rules |
| self.apply_rules(rules, "deny icmpv4") |
| |
| # Traffic should not pass |
| self.run_verify_negat_test(self.ICMP, self.IPV4, 0) |
| |
| self.logger.info("ACLP_TEST_FINISH_0006") |
| |
| def test_0007_vpp624_deny_icmpv6(self): |
| """ VPP_624 deny ICMPv6 |
| """ |
| self.logger.info("ACLP_TEST_START_0007") |
| # Add an ACL |
| rules = [] |
| rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_RANGE, |
| self.proto[self.ICMP][self.ICMPv6])) |
| # deny ip any any in the end |
| rules.append(self.create_rule(self.IPV6, self.PERMIT, |
| self.PORTS_ALL, 0)) |
| |
| # Apply rules |
| self.apply_rules(rules, "deny icmpv6") |
| |
| # Traffic should not pass |
| self.run_verify_negat_test(self.ICMP, self.IPV6, 0) |
| |
| self.logger.info("ACLP_TEST_FINISH_0007") |
| |
| def test_0008_tcp_permit_v4(self): |
| """ permit TCPv4 |
| """ |
| self.logger.info("ACLP_TEST_START_0008") |
| |
| # Add an ACL |
| rules = [] |
| rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE, |
| self.proto[self.IP][self.TCP])) |
| # deny ip any any in the end |
| rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0)) |
| |
| # Apply rules |
| self.apply_rules(rules, "permit ipv4 tcp") |
| |
| # Traffic should still pass |
| self.run_verify_test(self.IP, self.IPV4, self.proto[self.IP][self.TCP]) |
| |
| self.logger.info("ACLP_TEST_FINISH_0008") |
| |
| def test_0009_tcp_permit_v6(self): |
| """ permit TCPv6 |
| """ |
| self.logger.info("ACLP_TEST_START_0009") |
| |
| # Add an ACL |
| rules = [] |
| rules.append(self.create_rule(self.IPV6, self.PERMIT, self.PORTS_RANGE, |
| self.proto[self.IP][self.TCP])) |
| # deny ip any any in the end |
| rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_ALL, 0)) |
| |
| # Apply rules |
| self.apply_rules(rules, "permit ip6 tcp") |
| |
| # Traffic should still pass |
| self.run_verify_test(self.IP, self.IPV6, self.proto[self.IP][self.TCP]) |
| |
| self.logger.info("ACLP_TEST_FINISH_0008") |
| |
| def test_0010_udp_permit_v4(self): |
| """ permit UDPv4 |
| """ |
| self.logger.info("ACLP_TEST_START_0010") |
| |
| # Add an ACL |
| rules = [] |
| rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE, |
| self.proto[self.IP][self.UDP])) |
| # deny ip any any in the end |
| rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0)) |
| |
| # Apply rules |
| self.apply_rules(rules, "permit ipv udp") |
| |
| # Traffic should still pass |
| self.run_verify_test(self.IP, self.IPV4, self.proto[self.IP][self.UDP]) |
| |
| self.logger.info("ACLP_TEST_FINISH_0010") |
| |
| def test_0011_udp_permit_v6(self): |
| """ permit UDPv6 |
| """ |
| self.logger.info("ACLP_TEST_START_0011") |
| |
| # Add an ACL |
| rules = [] |
| rules.append(self.create_rule(self.IPV6, self.PERMIT, self.PORTS_RANGE, |
| self.proto[self.IP][self.UDP])) |
| # deny ip any any in the end |
| rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_ALL, 0)) |
| |
| # Apply rules |
| self.apply_rules(rules, "permit ip6 udp") |
| |
| # Traffic should still pass |
| self.run_verify_test(self.IP, self.IPV6, self.proto[self.IP][self.UDP]) |
| |
| self.logger.info("ACLP_TEST_FINISH_0011") |
| |
| def test_0012_tcp_deny(self): |
| """ deny TCPv4/v6 |
| """ |
| self.logger.info("ACLP_TEST_START_0012") |
| |
| # Add an ACL |
| rules = [] |
| rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE, |
| self.proto[self.IP][self.TCP])) |
| rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_RANGE, |
| self.proto[self.IP][self.TCP])) |
| # permit ip any any in the end |
| rules.append(self.create_rule(self.IPV4, self.PERMIT, |
| self.PORTS_ALL, 0)) |
| rules.append(self.create_rule(self.IPV6, self.PERMIT, |
| self.PORTS_ALL, 0)) |
| |
| # Apply rules |
| self.apply_rules(rules, "deny ip4/ip6 tcp") |
| |
| # Traffic should not pass |
| self.run_verify_negat_test(self.IP, self.IPRANDOM, |
| self.proto[self.IP][self.TCP]) |
| |
| self.logger.info("ACLP_TEST_FINISH_0012") |
| |
| def test_0013_udp_deny(self): |
| """ deny UDPv4/v6 |
| """ |
| self.logger.info("ACLP_TEST_START_0013") |
| |
| # Add an ACL |
| rules = [] |
| rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE, |
| self.proto[self.IP][self.UDP])) |
| rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_RANGE, |
| self.proto[self.IP][self.UDP])) |
| # permit ip any any in the end |
| rules.append(self.create_rule(self.IPV4, self.PERMIT, |
| self.PORTS_ALL, 0)) |
| rules.append(self.create_rule(self.IPV6, self.PERMIT, |
| self.PORTS_ALL, 0)) |
| |
| # Apply rules |
| self.apply_rules(rules, "deny ip4/ip6 udp") |
| |
| # Traffic should not pass |
| self.run_verify_negat_test(self.IP, self.IPRANDOM, |
| self.proto[self.IP][self.UDP]) |
| |
| self.logger.info("ACLP_TEST_FINISH_0013") |
| |
| def test_0014_acl_dump(self): |
| """ verify add/dump acls |
| """ |
| self.logger.info("ACLP_TEST_START_0014") |
| |
| r = [[self.IPV4, self.PERMIT, 1234, self.proto[self.IP][self.TCP]], |
| [self.IPV4, self.PERMIT, 2345, self.proto[self.IP][self.UDP]], |
| [self.IPV4, self.PERMIT, 0, self.proto[self.IP][self.TCP]], |
| [self.IPV4, self.PERMIT, 0, self.proto[self.IP][self.UDP]], |
| [self.IPV4, self.PERMIT, 5, self.proto[self.ICMP][self.ICMPv4]], |
| [self.IPV6, self.PERMIT, 4321, self.proto[self.IP][self.TCP]], |
| [self.IPV6, self.PERMIT, 5432, self.proto[self.IP][self.UDP]], |
| [self.IPV6, self.PERMIT, 0, self.proto[self.IP][self.TCP]], |
| [self.IPV6, self.PERMIT, 0, self.proto[self.IP][self.UDP]], |
| [self.IPV6, self.PERMIT, 6, self.proto[self.ICMP][self.ICMPv6]], |
| [self.IPV4, self.DENY, self.PORTS_ALL, 0], |
| [self.IPV4, self.DENY, 1234, self.proto[self.IP][self.TCP]], |
| [self.IPV4, self.DENY, 2345, self.proto[self.IP][self.UDP]], |
| [self.IPV4, self.DENY, 5, self.proto[self.ICMP][self.ICMPv4]], |
| [self.IPV6, self.DENY, 4321, self.proto[self.IP][self.TCP]], |
| [self.IPV6, self.DENY, 5432, self.proto[self.IP][self.UDP]], |
| [self.IPV6, self.DENY, 6, self.proto[self.ICMP][self.ICMPv6]], |
| [self.IPV6, self.DENY, self.PORTS_ALL, 0] |
| ] |
| |
| # Add and verify new ACLs |
| rules = [] |
| for i in range(len(r)): |
| rules.append(self.create_rule(r[i][0], r[i][1], r[i][2], r[i][3])) |
| |
| reply = self.vapi.acl_add_replace(acl_index=4294967295, r=rules) |
| result = self.vapi.acl_dump(reply.acl_index) |
| |
| i = 0 |
| for drules in result: |
| for dr in drules.r: |
| self.assertEqual(dr.is_ipv6, r[i][0]) |
| self.assertEqual(dr.is_permit, r[i][1]) |
| self.assertEqual(dr.proto, r[i][3]) |
| |
| if r[i][2] > 0: |
| self.assertEqual(dr.srcport_or_icmptype_first, r[i][2]) |
| else: |
| if r[i][2] < 0: |
| self.assertEqual(dr.srcport_or_icmptype_first, 0) |
| self.assertEqual(dr.srcport_or_icmptype_last, 65535) |
| else: |
| if dr.proto == self.proto[self.IP][self.TCP]: |
| self.assertGreater(dr.srcport_or_icmptype_first, |
| self.tcp_sport_from-1) |
| self.assertLess(dr.srcport_or_icmptype_first, |
| self.tcp_sport_to+1) |
| self.assertGreater(dr.dstport_or_icmpcode_last, |
| self.tcp_dport_from-1) |
| self.assertLess(dr.dstport_or_icmpcode_last, |
| self.tcp_dport_to+1) |
| elif dr.proto == self.proto[self.IP][self.UDP]: |
| self.assertGreater(dr.srcport_or_icmptype_first, |
| self.udp_sport_from-1) |
| self.assertLess(dr.srcport_or_icmptype_first, |
| self.udp_sport_to+1) |
| self.assertGreater(dr.dstport_or_icmpcode_last, |
| self.udp_dport_from-1) |
| self.assertLess(dr.dstport_or_icmpcode_last, |
| self.udp_dport_to+1) |
| i += 1 |
| |
| self.logger.info("ACLP_TEST_FINISH_0014") |
| |
| def test_0015_tcp_permit_port_v4(self): |
| """ permit single TCPv4 |
| """ |
| self.logger.info("ACLP_TEST_START_0015") |
| |
| port = random.randint(0, 65535) |
| # Add an ACL |
| rules = [] |
| rules.append(self.create_rule(self.IPV4, self.PERMIT, port, |
| self.proto[self.IP][self.TCP])) |
| # deny ip any any in the end |
| rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0)) |
| |
| # Apply rules |
| self.apply_rules(rules, "permit ip4 tcp "+str(port)) |
| |
| # Traffic should still pass |
| self.run_verify_test(self.IP, self.IPV4, |
| self.proto[self.IP][self.TCP], port) |
| |
| self.logger.info("ACLP_TEST_FINISH_0015") |
| |
| def test_0016_udp_permit_port_v4(self): |
| """ permit single UDPv4 |
| """ |
| self.logger.info("ACLP_TEST_START_0016") |
| |
| port = random.randint(0, 65535) |
| # Add an ACL |
| rules = [] |
| rules.append(self.create_rule(self.IPV4, self.PERMIT, port, |
| self.proto[self.IP][self.UDP])) |
| # deny ip any any in the end |
| rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0)) |
| |
| # Apply rules |
| self.apply_rules(rules, "permit ip4 tcp "+str(port)) |
| |
| # Traffic should still pass |
| self.run_verify_test(self.IP, self.IPV4, |
| self.proto[self.IP][self.UDP], port) |
| |
| self.logger.info("ACLP_TEST_FINISH_0016") |
| |
| def test_0017_tcp_permit_port_v6(self): |
| """ permit single TCPv6 |
| """ |
| self.logger.info("ACLP_TEST_START_0017") |
| |
| port = random.randint(0, 65535) |
| # Add an ACL |
| rules = [] |
| rules.append(self.create_rule(self.IPV6, self.PERMIT, port, |
| self.proto[self.IP][self.TCP])) |
| # deny ip any any in the end |
| rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_ALL, 0)) |
| |
| # Apply rules |
| self.apply_rules(rules, "permit ip4 tcp "+str(port)) |
| |
| # Traffic should still pass |
| self.run_verify_test(self.IP, self.IPV6, |
| self.proto[self.IP][self.TCP], port) |
| |
| self.logger.info("ACLP_TEST_FINISH_0017") |
| |
| def test_0018_udp_permit_port_v6(self): |
| """ permit single UPPv6 |
| """ |
| self.logger.info("ACLP_TEST_START_0018") |
| |
| port = random.randint(0, 65535) |
| # Add an ACL |
| rules = [] |
| rules.append(self.create_rule(self.IPV6, self.PERMIT, port, |
| self.proto[self.IP][self.UDP])) |
| # deny ip any any in the end |
| rules.append(self.create_rule(self.IPV6, self.DENY, |
| self.PORTS_ALL, 0)) |
| |
| # Apply rules |
| self.apply_rules(rules, "permit ip4 tcp "+str(port)) |
| |
| # Traffic should still pass |
| self.run_verify_test(self.IP, self.IPV6, |
| self.proto[self.IP][self.UDP], port) |
| |
| self.logger.info("ACLP_TEST_FINISH_0018") |
| |
| def test_0019_udp_deny_port(self): |
| """ deny single TCPv4/v6 |
| """ |
| self.logger.info("ACLP_TEST_START_0019") |
| |
| port = random.randint(0, 65535) |
| # Add an ACL |
| rules = [] |
| rules.append(self.create_rule(self.IPV4, self.DENY, port, |
| self.proto[self.IP][self.TCP])) |
| rules.append(self.create_rule(self.IPV6, self.DENY, port, |
| self.proto[self.IP][self.TCP])) |
| # Permit ip any any in the end |
| rules.append(self.create_rule(self.IPV4, self.PERMIT, |
| self.PORTS_ALL, 0)) |
| rules.append(self.create_rule(self.IPV6, self.PERMIT, |
| self.PORTS_ALL, 0)) |
| |
| # Apply rules |
| self.apply_rules(rules, "deny ip4/ip6 udp "+str(port)) |
| |
| # Traffic should not pass |
| self.run_verify_negat_test(self.IP, self.IPRANDOM, |
| self.proto[self.IP][self.TCP], port) |
| |
| self.logger.info("ACLP_TEST_FINISH_0019") |
| |
| def test_0020_udp_deny_port(self): |
| """ deny single UDPv4/v6 |
| """ |
| self.logger.info("ACLP_TEST_START_0020") |
| |
| port = random.randint(0, 65535) |
| # Add an ACL |
| rules = [] |
| rules.append(self.create_rule(self.IPV4, self.DENY, port, |
| self.proto[self.IP][self.UDP])) |
| rules.append(self.create_rule(self.IPV6, self.DENY, port, |
| self.proto[self.IP][self.UDP])) |
| # Permit ip any any in the end |
| rules.append(self.create_rule(self.IPV4, self.PERMIT, |
| self.PORTS_ALL, 0)) |
| rules.append(self.create_rule(self.IPV6, self.PERMIT, |
| self.PORTS_ALL, 0)) |
| |
| # Apply rules |
| self.apply_rules(rules, "deny ip4/ip6 udp "+str(port)) |
| |
| # Traffic should not pass |
| self.run_verify_negat_test(self.IP, self.IPRANDOM, |
| self.proto[self.IP][self.UDP], port) |
| |
| self.logger.info("ACLP_TEST_FINISH_0020") |
| |
| def test_0021_udp_deny_port_verify_fragment_deny(self): |
| """ deny single UDPv4/v6, permit ip any, verify non-initial fragment |
| blocked |
| """ |
| self.logger.info("ACLP_TEST_START_0021") |
| |
| port = random.randint(0, 65535) |
| # Add an ACL |
| rules = [] |
| rules.append(self.create_rule(self.IPV4, self.DENY, port, |
| self.proto[self.IP][self.UDP])) |
| rules.append(self.create_rule(self.IPV6, self.DENY, port, |
| self.proto[self.IP][self.UDP])) |
| # deny ip any any in the end |
| rules.append(self.create_rule(self.IPV4, self.PERMIT, |
| self.PORTS_ALL, 0)) |
| rules.append(self.create_rule(self.IPV6, self.PERMIT, |
| self.PORTS_ALL, 0)) |
| |
| # Apply rules |
| self.apply_rules(rules, "deny ip4/ip6 udp "+str(port)) |
| |
| # Traffic should not pass |
| self.run_verify_negat_test(self.IP, self.IPRANDOM, |
| self.proto[self.IP][self.UDP], port, True) |
| |
| self.logger.info("ACLP_TEST_FINISH_0021") |
| |
| def test_0022_zero_length_udp_ipv4(self): |
| """ VPP-687 zero length udp ipv4 packet""" |
| self.logger.info("ACLP_TEST_START_0022") |
| |
| port = random.randint(0, 65535) |
| # Add an ACL |
| rules = [] |
| rules.append(self.create_rule(self.IPV4, self.PERMIT, port, |
| self.proto[self.IP][self.UDP])) |
| # deny ip any any in the end |
| rules.append( |
| self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0)) |
| |
| # Apply rules |
| self.apply_rules(rules, "permit empty udp ip4 " + str(port)) |
| |
| # Traffic should still pass |
| # Create incoming packet streams for packet-generator interfaces |
| pkts_cnt = 0 |
| pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes, |
| self.IP, self.IPV4, |
| self.proto[self.IP][self.UDP], port, |
| False, False) |
| if len(pkts) > 0: |
| self.pg0.add_stream(pkts) |
| pkts_cnt += len(pkts) |
| |
| # Enable packet capture and start packet sendingself.IPV |
| self.pg_enable_capture(self.pg_interfaces) |
| self.pg_start() |
| |
| self.pg1.get_capture(pkts_cnt) |
| |
| self.logger.info("ACLP_TEST_FINISH_0022") |
| |
| def test_0023_zero_length_udp_ipv6(self): |
| """ VPP-687 zero length udp ipv6 packet""" |
| self.logger.info("ACLP_TEST_START_0023") |
| |
| port = random.randint(0, 65535) |
| # Add an ACL |
| rules = [] |
| rules.append(self.create_rule(self.IPV6, self.PERMIT, port, |
| self.proto[self.IP][self.UDP])) |
| # deny ip any any in the end |
| rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_ALL, 0)) |
| |
| # Apply rules |
| self.apply_rules(rules, "permit empty udp ip6 "+str(port)) |
| |
| # Traffic should still pass |
| # Create incoming packet streams for packet-generator interfaces |
| pkts_cnt = 0 |
| pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes, |
| self.IP, self.IPV6, |
| self.proto[self.IP][self.UDP], port, |
| False, False) |
| if len(pkts) > 0: |
| self.pg0.add_stream(pkts) |
| pkts_cnt += len(pkts) |
| |
| # Enable packet capture and start packet sendingself.IPV |
| self.pg_enable_capture(self.pg_interfaces) |
| self.pg_start() |
| |
| # Verify outgoing packet streams per packet-generator interface |
| self.pg1.get_capture(pkts_cnt) |
| |
| self.logger.info("ACLP_TEST_FINISH_0023") |
| |
| def test_0108_tcp_permit_v4(self): |
| """ permit TCPv4 + non-match range |
| """ |
| self.logger.info("ACLP_TEST_START_0108") |
| |
| # Add an ACL |
| rules = [] |
| rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE_2, |
| self.proto[self.IP][self.TCP])) |
| rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE, |
| self.proto[self.IP][self.TCP])) |
| # deny ip any any in the end |
| rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0)) |
| |
| # Apply rules |
| self.apply_rules(rules, "permit ipv4 tcp") |
| |
| # Traffic should still pass |
| self.run_verify_test(self.IP, self.IPV4, self.proto[self.IP][self.TCP]) |
| |
| self.logger.info("ACLP_TEST_FINISH_0108") |
| |
| def test_0109_tcp_permit_v6(self): |
| """ permit TCPv6 + non-match range |
| """ |
| self.logger.info("ACLP_TEST_START_0109") |
| |
| # Add an ACL |
| rules = [] |
| rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_RANGE_2, |
| self.proto[self.IP][self.TCP])) |
| rules.append(self.create_rule(self.IPV6, self.PERMIT, self.PORTS_RANGE, |
| self.proto[self.IP][self.TCP])) |
| # deny ip any any in the end |
| rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_ALL, 0)) |
| |
| # Apply rules |
| self.apply_rules(rules, "permit ip6 tcp") |
| |
| # Traffic should still pass |
| self.run_verify_test(self.IP, self.IPV6, self.proto[self.IP][self.TCP]) |
| |
| self.logger.info("ACLP_TEST_FINISH_0109") |
| |
| def test_0110_udp_permit_v4(self): |
| """ permit UDPv4 + non-match range |
| """ |
| self.logger.info("ACLP_TEST_START_0110") |
| |
| # Add an ACL |
| rules = [] |
| rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE_2, |
| self.proto[self.IP][self.UDP])) |
| rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE, |
| self.proto[self.IP][self.UDP])) |
| # deny ip any any in the end |
| rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0)) |
| |
| # Apply rules |
| self.apply_rules(rules, "permit ipv4 udp") |
| |
| # Traffic should still pass |
| self.run_verify_test(self.IP, self.IPV4, self.proto[self.IP][self.UDP]) |
| |
| self.logger.info("ACLP_TEST_FINISH_0110") |
| |
| def test_0111_udp_permit_v6(self): |
| """ permit UDPv6 + non-match range |
| """ |
| self.logger.info("ACLP_TEST_START_0111") |
| |
| # Add an ACL |
| rules = [] |
| rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_RANGE_2, |
| self.proto[self.IP][self.UDP])) |
| rules.append(self.create_rule(self.IPV6, self.PERMIT, self.PORTS_RANGE, |
| self.proto[self.IP][self.UDP])) |
| # deny ip any any in the end |
| rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_ALL, 0)) |
| |
| # Apply rules |
| self.apply_rules(rules, "permit ip6 udp") |
| |
| # Traffic should still pass |
| self.run_verify_test(self.IP, self.IPV6, self.proto[self.IP][self.UDP]) |
| |
| self.logger.info("ACLP_TEST_FINISH_0111") |
| |
| def test_0112_tcp_deny(self): |
| """ deny TCPv4/v6 + non-match range |
| """ |
| self.logger.info("ACLP_TEST_START_0112") |
| |
| # Add an ACL |
| rules = [] |
| rules.append(self.create_rule(self.IPV4, self.PERMIT, |
| self.PORTS_RANGE_2, |
| self.proto[self.IP][self.TCP])) |
| rules.append(self.create_rule(self.IPV6, self.PERMIT, |
| self.PORTS_RANGE_2, |
| self.proto[self.IP][self.TCP])) |
| rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE, |
| self.proto[self.IP][self.TCP])) |
| rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_RANGE, |
| self.proto[self.IP][self.TCP])) |
| # permit ip any any in the end |
| rules.append(self.create_rule(self.IPV4, self.PERMIT, |
| self.PORTS_ALL, 0)) |
| rules.append(self.create_rule(self.IPV6, self.PERMIT, |
| self.PORTS_ALL, 0)) |
| |
| # Apply rules |
| self.apply_rules(rules, "deny ip4/ip6 tcp") |
| |
| # Traffic should not pass |
| self.run_verify_negat_test(self.IP, self.IPRANDOM, |
| self.proto[self.IP][self.TCP]) |
| |
| self.logger.info("ACLP_TEST_FINISH_0112") |
| |
| def test_0113_udp_deny(self): |
| """ deny UDPv4/v6 + non-match range |
| """ |
| self.logger.info("ACLP_TEST_START_0113") |
| |
| # Add an ACL |
| rules = [] |
| rules.append(self.create_rule(self.IPV4, self.PERMIT, |
| self.PORTS_RANGE_2, |
| self.proto[self.IP][self.UDP])) |
| rules.append(self.create_rule(self.IPV6, self.PERMIT, |
| self.PORTS_RANGE_2, |
| self.proto[self.IP][self.UDP])) |
| rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE, |
| self.proto[self.IP][self.UDP])) |
| rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_RANGE, |
| self.proto[self.IP][self.UDP])) |
| # permit ip any any in the end |
| rules.append(self.create_rule(self.IPV4, self.PERMIT, |
| self.PORTS_ALL, 0)) |
| rules.append(self.create_rule(self.IPV6, self.PERMIT, |
| self.PORTS_ALL, 0)) |
| |
| # Apply rules |
| self.apply_rules(rules, "deny ip4/ip6 udp") |
| |
| # Traffic should not pass |
| self.run_verify_negat_test(self.IP, self.IPRANDOM, |
| self.proto[self.IP][self.UDP]) |
| |
| self.logger.info("ACLP_TEST_FINISH_0113") |
| |
| def test_0300_tcp_permit_v4_etype_aaaa(self): |
| """ permit TCPv4, send 0xAAAA etype |
| """ |
| self.logger.info("ACLP_TEST_START_0300") |
| |
| # Add an ACL |
| rules = [] |
| rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE_2, |
| self.proto[self.IP][self.TCP])) |
| rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE, |
| self.proto[self.IP][self.TCP])) |
| # deny ip any any in the end |
| rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0)) |
| |
| # Apply rules |
| self.apply_rules(rules, "permit ipv4 tcp") |
| |
| # Traffic should still pass also for an odd ethertype |
| self.run_verify_test(self.IP, self.IPV4, self.proto[self.IP][self.TCP], |
| 0, False, True, 0xaaaa) |
| self.logger.info("ACLP_TEST_FINISH_0300") |
| |
| def test_0305_tcp_permit_v4_etype_blacklist_aaaa(self): |
| """ permit TCPv4, whitelist 0x0BBB ethertype, send 0xAAAA-blocked |
| """ |
| self.logger.info("ACLP_TEST_START_0305") |
| |
| # Add an ACL |
| rules = [] |
| rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE_2, |
| self.proto[self.IP][self.TCP])) |
| rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE, |
| self.proto[self.IP][self.TCP])) |
| # deny ip any any in the end |
| rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0)) |
| |
| # Apply rules |
| self.apply_rules(rules, "permit ipv4 tcp") |
| |
| # whitelist the 0xbbbb etype - so the 0xaaaa should be blocked |
| self.etype_whitelist([0xbbb], 1) |
| |
| # The oddball ethertype should be blocked |
| self.run_verify_negat_test(self.IP, self.IPV4, |
| self.proto[self.IP][self.TCP], |
| 0, False, 0xaaaa) |
| |
| # remove the whitelist |
| self.etype_whitelist([], 0) |
| |
| self.logger.info("ACLP_TEST_FINISH_0305") |
| |
| def test_0306_tcp_permit_v4_etype_blacklist_aaaa(self): |
| """ permit TCPv4, whitelist 0x0BBB ethertype, send 0x0BBB - pass |
| """ |
| self.logger.info("ACLP_TEST_START_0306") |
| |
| # Add an ACL |
| rules = [] |
| rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE_2, |
| self.proto[self.IP][self.TCP])) |
| rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE, |
| self.proto[self.IP][self.TCP])) |
| # deny ip any any in the end |
| rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0)) |
| |
| # Apply rules |
| self.apply_rules(rules, "permit ipv4 tcp") |
| |
| # whitelist the 0xbbbb etype - so the 0xaaaa should be blocked |
| self.etype_whitelist([0xbbb], 1) |
| |
| # The whitelisted traffic, should pass |
| self.run_verify_test(self.IP, self.IPV4, self.proto[self.IP][self.TCP], |
| 0, False, True, 0x0bbb) |
| |
| # remove the whitelist, the previously blocked 0xAAAA should pass now |
| self.etype_whitelist([], 0) |
| |
| self.logger.info("ACLP_TEST_FINISH_0306") |
| |
| def test_0307_tcp_permit_v4_etype_blacklist_aaaa(self): |
| """ permit TCPv4, whitelist 0x0BBB, remove, send 0xAAAA - pass |
| """ |
| self.logger.info("ACLP_TEST_START_0307") |
| |
| # Add an ACL |
| rules = [] |
| rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE_2, |
| self.proto[self.IP][self.TCP])) |
| rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE, |
| self.proto[self.IP][self.TCP])) |
| # deny ip any any in the end |
| rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0)) |
| |
| # Apply rules |
| self.apply_rules(rules, "permit ipv4 tcp") |
| |
| # whitelist the 0xbbbb etype - so the 0xaaaa should be blocked |
| self.etype_whitelist([0xbbb], 1) |
| # remove the whitelist, the previously blocked 0xAAAA should pass now |
| self.etype_whitelist([], 0) |
| |
| # The whitelisted traffic, should pass |
| self.run_verify_test(self.IP, self.IPV4, self.proto[self.IP][self.TCP], |
| 0, False, True, 0xaaaa) |
| |
| self.logger.info("ACLP_TEST_FINISH_0306") |
| |
| def test_0315_del_intf(self): |
| """ apply an acl and delete the interface |
| """ |
| self.logger.info("ACLP_TEST_START_0315") |
| |
| # Add an ACL |
| rules = [] |
| rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE_2, |
| self.proto[self.IP][self.TCP])) |
| rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE, |
| self.proto[self.IP][self.TCP])) |
| # deny ip any any in the end |
| rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0)) |
| |
| # create an interface |
| intf = [] |
| intf.append(VppLoInterface(self)) |
| |
| # Apply rules |
| self.apply_rules_to(rules, "permit ipv4 tcp", intf[0].sw_if_index) |
| |
| # Remove the interface |
| intf[0].remove_vpp_config() |
| |
| self.logger.info("ACLP_TEST_FINISH_0315") |
| |
| if __name__ == '__main__': |
| unittest.main(testRunner=VppTestRunner) |