diff options
Diffstat (limited to 'test')
| -rwxr-xr-x | test/framework.py | 24 | ||||
| -rw-r--r-- | test/template_classifier.py | 66 | ||||
| -rw-r--r-- | test/test_classifier.py | 282 |
3 files changed, 352 insertions, 20 deletions
diff --git a/test/framework.py b/test/framework.py index b7004613edd..731c5e18043 100755 --- a/test/framework.py +++ b/test/framework.py @@ -23,6 +23,7 @@ from traceback import format_exception from logging import FileHandler, DEBUG, Formatter from enum import Enum from abc import ABC, abstractmethod +from struct import pack, unpack import scapy.compat from scapy.packet import Raw @@ -1042,8 +1043,10 @@ class VppTestCase(CPUInterface, unittest.TestCase): :returns: string containing serialized data from packet info """ - return "%d %d %d %d %d" % (info.index, info.src, info.dst, - info.ip, info.proto) + + # retrieve payload, currently 18 bytes (4 x ints + 1 short) + return pack('iiiih', info.index, info.src, + info.dst, info.ip, info.proto) @staticmethod def payload_to_info(payload, payload_field='load'): @@ -1058,13 +1061,18 @@ class VppTestCase(CPUInterface, unittest.TestCase): :returns: _PacketInfo object containing de-serialized data from payload """ - numbers = getattr(payload, payload_field).split() + + # retrieve payload, currently 18 bytes (4 x ints + 1 short) + payload_b = getattr(payload, payload_field)[:18] + info = _PacketInfo() - info.index = int(numbers[0]) - info.src = int(numbers[1]) - info.dst = int(numbers[2]) - info.ip = int(numbers[3]) - info.proto = int(numbers[4]) + info.index, info.src, info.dst, info.ip, info.proto \ + = unpack('iiiih', payload_b) + + # some SRv6 TCs depend on get an exception if bad values are detected + if info.index > 0x4000: + raise ValueError('Index value is invalid') + return info def get_next_packet_info(self, info): diff --git a/test/template_classifier.py b/test/template_classifier.py index b8f79b4f99f..b9a8ca9d7f3 100644 --- a/test/template_classifier.py +++ b/test/template_classifier.py @@ -5,6 +5,7 @@ import socket from socket import AF_INET, AF_INET6 import unittest import sys +from dataclasses import dataclass from framework import VppTestCase @@ -15,6 +16,19 @@ from scapy.layers.inet6 import IPv6 from util import ppp +@dataclass +class VarMask: + offset: int + spec: str + + +@dataclass +class VarMatch: + offset: int + value: int + length: int + + class TestClassifier(VppTestCase): @staticmethod @@ -97,12 +111,9 @@ class TestClassifier(VppTestCase): self.logger.info(self.vapi.cli("show classify table verbose")) self.logger.info(self.vapi.cli("show ip fib")) + self.logger.info(self.vapi.cli("show error")) - acl_active_table = 'ip_out' - if self.af == AF_INET6: - acl_active_table = 'ip6_out' - - if self.acl_active_table == acl_active_table: + if self.acl_active_table.endswith('out'): self.output_acl_set_interface( self.pg0, self.acl_tbl_idx.get(self.acl_active_table), 0) self.acl_active_table = '' @@ -134,7 +145,7 @@ class TestClassifier(VppTestCase): src_mac = src_mac.replace(':', '') return ('{!s:0>12}{!s:0>12}{!s:0>4}'.format( - dst_mac, src_mac, ether_type)).rstrip('0') + dst_mac, src_mac, ether_type)).rstrip() @staticmethod def build_mac_mask(dst_mac='', src_mac='', ether_type=''): @@ -146,7 +157,7 @@ class TestClassifier(VppTestCase): """ return ('{!s:0>12}{!s:0>12}{!s:0>4}'.format( - dst_mac, src_mac, ether_type)).rstrip('0') + dst_mac, src_mac, ether_type)).rstrip() @staticmethod def build_ip_mask(proto='', src_ip='', dst_ip='', @@ -179,6 +190,18 @@ class TestClassifier(VppTestCase): nh, src_ip, dst_ip, src_port, dst_port)).rstrip('0') @staticmethod + def build_payload_mask(masks): + payload_mask = '' + + for mask in masks: + # offset is specified in bytes, convert to hex format. + length = (mask.offset * 2) + len(mask.spec) + format_spec = '{!s:0>' + str(length) + '}' + payload_mask += format_spec.format(mask.spec) + + return payload_mask.rstrip('0') + + @staticmethod def build_ip_match(proto=0, src_ip='', dst_ip='', src_port=0, dst_port=0): """Build IP ACL match data with hexstring format. @@ -228,8 +251,23 @@ class TestClassifier(VppTestCase): hex(nh)[2:], src_ip, dst_ip, hex(src_port)[2:], hex(dst_port)[2:])).rstrip('0') + @staticmethod + def build_payload_match(matches): + payload_match = '' + + for match in matches: + sval = str(hex(match.value)[2:]) + # offset is specified in bytes, convert to hex format. + length = (match.offset + match.length) * 2 + + format_spec = '{!s:0>' + str(length) + '}' + payload_match += format_spec.format(sval) + + return payload_match.rstrip('0') + def create_stream(self, src_if, dst_if, packet_sizes, - proto_l=UDP(sport=1234, dport=5678)): + proto_l=UDP(sport=1234, dport=5678), + payload_ex=None): """Create input packet stream for defined interfaces. :param VppInterface src_if: Source Interface for packet stream. @@ -242,6 +280,11 @@ class TestClassifier(VppTestCase): for size in packet_sizes: info = self.create_packet_info(src_if, dst_if) payload = self.info_to_payload(info) + + # append any additional payload after info + if payload_ex is not None: + payload += payload_ex + if self.af == AF_INET: p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) / IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4) / @@ -306,12 +349,14 @@ class TestClassifier(VppTestCase): "Interface %s: Packet expected from interface %s " "didn't arrive" % (dst_if.name, i.name)) - def create_classify_table(self, key, mask, data_offset=0): + def create_classify_table(self, key, mask, data_offset=0, + next_table_index=None): """Create Classify Table :param str key: key for classify table (ex, ACL name). :param str mask: mask value for interested traffic. :param int data_offset: + :param str next_table_index """ mask_match, mask_match_len = self._resolve_mask_match(mask) r = self.vapi.classify_add_del_table( @@ -321,7 +366,8 @@ class TestClassifier(VppTestCase): match_n_vectors=(len(mask) - 1) // 32 + 1, miss_next_index=0, current_data_flag=1, - current_data_offset=data_offset) + current_data_offset=data_offset, + next_table_index=next_table_index) self.assertIsNotNone(r, 'No response msg for add_del_table') self.acl_tbl_idx[key] = r.new_table_index diff --git a/test/test_classifier.py b/test/test_classifier.py index 11c0985f4d4..1c7fb4c5a5c 100644 --- a/test/test_classifier.py +++ b/test/test_classifier.py @@ -5,12 +5,12 @@ import socket import unittest from framework import VppTestCase, VppTestRunner +from scapy.packet import Raw, Packet -from scapy.packet import Raw from scapy.layers.l2 import Ether from scapy.layers.inet import IP, UDP, TCP from util import ppp -from template_classifier import TestClassifier +from template_classifier import TestClassifier, VarMask, VarMatch from vpp_ip_route import VppIpRoute, VppRoutePath from vpp_ip import INVALID_INDEX @@ -504,6 +504,284 @@ class TestClassifierMAC(TestClassifier): self.pg3.assert_nothing_captured(remark="packets forwarded") +class TestClassifierComplex(TestClassifier): + """ Large & Nested Classifiers Test Cases """ + + @classmethod + def setUpClass(cls): + super(TestClassifierComplex, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(TestClassifierComplex, cls).tearDownClass() + + def test_iacl_large(self): + """ Large input ACL test + + Test scenario for Large ACL matching on ethernet+ip+udp headers + - Create IPv4 stream for pg0 -> pg1 interface. + - Create large acl matching on ethernet+ip+udp header fields + - Send and verify received packets on pg1 interface. + """ + + # 40b offset = 80bytes - (sizeof(UDP/IP/ETH) + 4b) + # + 4b as build_ip_ma*() func, do not match against UDP Len & Chksum + msk = VarMask(offset=40, spec='ffff') + mth = VarMatch(offset=40, value=0x1234, length=2) + + payload_msk = self.build_payload_mask([msk]) + payload_match = self.build_payload_match([mth]) + + sport = 13720 + dport = 9080 + + # 36b offset = 80bytes - (sizeof(UDP/IP/ETH)) + packet_ex = bytes.fromhex(('0' * 36) + '1234') + pkts = self.create_stream(self.pg0, self.pg1, self.pg_if_packet_sizes, + UDP(sport=sport, dport=dport), + packet_ex) + self.pg0.add_stream(pkts) + + key = 'large_in' + self.create_classify_table( + key, + self.build_mac_mask(src_mac='ffffffffffff', + dst_mac='ffffffffffff', + ether_type='ffff') + + self.build_ip_mask(proto='ff', + src_ip='ffffffff', + dst_ip='ffffffff', + src_port='ffff', + dst_port='ffff') + + payload_msk, + data_offset=-14) + + self.create_classify_session( + self.acl_tbl_idx.get(key), + self.build_mac_match(src_mac=self.pg0.remote_mac, + dst_mac=self.pg0.local_mac, + # ipv4 next header + ether_type='0800') + + self.build_ip_match(proto=socket.IPPROTO_UDP, + src_ip=self.pg0.remote_ip4, + dst_ip=self.pg1.remote_ip4, + src_port=sport, + dst_port=dport) + + payload_match + ) + + self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key)) + self.acl_active_table = key + + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + + pkts = self.pg1.get_capture(len(pkts)) + self.verify_capture(self.pg1, pkts) + self.pg0.assert_nothing_captured(remark="packets forwarded") + self.pg2.assert_nothing_captured(remark="packets forwarded") + self.pg3.assert_nothing_captured(remark="packets forwarded") + + def test_oacl_large(self): + """ Large output ACL test + Test scenario for Large ACL matching on ethernet+ip+udp headers + - Create IPv4 stream for pg1 -> pg0 interface. + - Create large acl matching on ethernet+ip+udp header fields + - Send and verify received packets on pg0 interface. + """ + + # 40b offset = 80bytes - (sizeof(UDP/IP/ETH) + 4b) + # + 4b as build_ip_ma*() func, do not match against UDP Len & Chksum + msk = VarMask(offset=40, spec='ffff') + mth = VarMatch(offset=40, value=0x1234, length=2) + + payload_msk = self.build_payload_mask([msk]) + payload_match = self.build_payload_match([mth]) + + sport = 13720 + dport = 9080 + + # 36b offset = 80bytes - (sizeof(UDP/IP/ETH)) + packet_ex = bytes.fromhex(('0' * 36) + '1234') + pkts = self.create_stream(self.pg1, self.pg0, self.pg_if_packet_sizes, + UDP(sport=sport, dport=dport), + packet_ex) + self.pg1.add_stream(pkts) + + key = 'large_out' + self.create_classify_table( + key, + self.build_mac_mask(src_mac='ffffffffffff', + dst_mac='ffffffffffff', + ether_type='ffff') + + self.build_ip_mask(proto='ff', + src_ip='ffffffff', + dst_ip='ffffffff', + src_port='ffff', + dst_port='ffff') + + payload_msk, + data_offset=-14) + + self.create_classify_session( + self.acl_tbl_idx.get(key), + self.build_mac_match(src_mac=self.pg0.local_mac, + dst_mac=self.pg0.remote_mac, + # ipv4 next header + ether_type='0800') + + self.build_ip_match(proto=socket.IPPROTO_UDP, + src_ip=self.pg1.remote_ip4, + dst_ip=self.pg0.remote_ip4, + src_port=sport, + dst_port=dport) + + payload_match) + + self.output_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key)) + self.acl_active_table = key + + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + + pkts = self.pg0.get_capture(len(pkts)) + self.verify_capture(self.pg0, pkts) + self.pg1.assert_nothing_captured(remark="packets forwarded") + self.pg2.assert_nothing_captured(remark="packets forwarded") + self.pg3.assert_nothing_captured(remark="packets forwarded") + + def test_iacl_nested(self): + """ Nested input ACL test + + Test scenario for Large ACL matching on ethernet+ip+udp headers + - Create IPv4 stream for pg0 -> pg1 interface. + - Create 1st classifier table, without any entries + - Create nested acl matching on ethernet+ip+udp header fields + - Send and verify received packets on pg1 interface. + """ + + sport = 13720 + dport = 9080 + pkts = self.create_stream(self.pg0, self.pg1, self.pg_if_packet_sizes, + UDP(sport=sport, dport=dport)) + + self.pg0.add_stream(pkts) + + subtable_key = 'subtable_in' + self.create_classify_table( + subtable_key, + self.build_mac_mask(src_mac='ffffffffffff', + dst_mac='ffffffffffff', + ether_type='ffff') + + self.build_ip_mask(proto='ff', + src_ip='ffffffff', + dst_ip='ffffffff', + src_port='ffff', + dst_port='ffff'), + data_offset=-14) + + key = 'nested_in' + self.create_classify_table( + key, + self.build_mac_mask(src_mac='ffffffffffff', + dst_mac='ffffffffffff', + ether_type='ffff') + + self.build_ip_mask(proto='ff', + src_ip='ffffffff', + dst_ip='ffffffff', + src_port='ffff', + dst_port='ffff'), + next_table_index=self.acl_tbl_idx.get(subtable_key)) + + self.create_classify_session( + self.acl_tbl_idx.get(subtable_key), + self.build_mac_match(src_mac=self.pg0.remote_mac, + dst_mac=self.pg0.local_mac, + # ipv4 next header + ether_type='0800') + + self.build_ip_match(proto=socket.IPPROTO_UDP, + src_ip=self.pg0.remote_ip4, + dst_ip=self.pg1.remote_ip4, + src_port=sport, + dst_port=dport)) + + self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key)) + self.acl_active_table = key + + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + + pkts = self.pg1.get_capture(len(pkts)) + self.verify_capture(self.pg1, pkts) + self.pg0.assert_nothing_captured(remark="packets forwarded") + self.pg2.assert_nothing_captured(remark="packets forwarded") + self.pg3.assert_nothing_captured(remark="packets forwarded") + + def test_oacl_nested(self): + """ Nested output ACL test + + Test scenario for Large ACL matching on ethernet+ip+udp headers + - Create IPv4 stream for pg1 -> pg0 interface. + - Create 1st classifier table, without any entries + - Create nested acl matching on ethernet+ip+udp header fields + - Send and verify received packets on pg0 interface. + """ + + sport = 13720 + dport = 9080 + pkts = self.create_stream(self.pg1, self.pg0, self.pg_if_packet_sizes, + UDP(sport=sport, dport=dport)) + self.pg1.add_stream(pkts) + + subtable_key = 'subtable_out' + self.create_classify_table( + subtable_key, + self.build_mac_mask(src_mac='ffffffffffff', + dst_mac='ffffffffffff', + ether_type='ffff') + + self.build_ip_mask(proto='ff', + src_ip='ffffffff', + dst_ip='ffffffff', + src_port='ffff', + dst_port='ffff'), + data_offset=-14) + + key = 'nested_out' + self.create_classify_table( + key, + self.build_mac_mask(src_mac='ffffffffffff', + dst_mac='ffffffffffff', + ether_type='ffff') + + self.build_ip_mask(proto='ff', + src_ip='ffffffff', + dst_ip='ffffffff', + src_port='ffff', + dst_port='ffff'), + next_table_index=self.acl_tbl_idx.get(subtable_key), + data_offset=-14) + + self.create_classify_session( + self.acl_tbl_idx.get(subtable_key), + self.build_mac_match(src_mac=self.pg0.local_mac, + dst_mac=self.pg0.remote_mac, + # ipv4 next header + ether_type='0800') + + self.build_ip_match(proto=socket.IPPROTO_UDP, + src_ip=self.pg1.remote_ip4, + dst_ip=self.pg0.remote_ip4, + src_port=sport, + dst_port=dport)) + + self.output_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key)) + self.acl_active_table = key + + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + + pkts = self.pg0.get_capture(len(pkts)) + self.verify_capture(self.pg0, pkts) + self.pg1.assert_nothing_captured(remark="packets forwarded") + self.pg2.assert_nothing_captured(remark="packets forwarded") + self.pg3.assert_nothing_captured(remark="packets forwarded") + + class TestClassifierPBR(TestClassifier): """ Classifier PBR Test Case """ |