test/test_classifier.py

   1 #!/usr/bin/env python
   2
   3 import unittest
   4 import socket
   5 import binascii
   6
   7 from framework import VppTestCase, VppTestRunner
   8
   9 from scapy.packet import Raw
  10 from scapy.layers.l2 import Ether
  11 from scapy.layers.inet import IP, UDP
  12 from util import ppp
  13
  14
  15 class TestClassifier(VppTestCase):
  16     """ Classifier Test Case """
  17
  18     def setUp(self):
  19         """
  20         Perform test setup before test case.
  21
  22         **Config:**
  23             - create 4 pg interfaces
  24                 - untagged pg0/pg1/pg2 interface
  25                     pg0 -------> pg1 (IP ACL)
  26                            \
  27                             ---> pg2 (MAC ACL))
  28                              \
  29                               -> pg3 (PBR)
  30             - setup interfaces:
  31                 - put it into UP state
  32                 - set IPv4 addresses
  33                 - resolve neighbor address using ARP
  34
  35         :ivar list interfaces: pg interfaces.
  36         :ivar list pg_if_packet_sizes: packet sizes in test.
  37         :ivar dict acl_tbl_idx: ACL table index.
  38         :ivar int pbr_vrfid: VRF id for PBR test.
  39         """
  40         super(TestClassifier, self).setUp()
  41
  42         # create 4 pg interfaces
  43         self.create_pg_interfaces(range(4))
  44
  45         # packet sizes to test
  46         self.pg_if_packet_sizes = [64, 9018]
  47
  48         self.interfaces = list(self.pg_interfaces)
  49
  50         # ACL & PBR vars
  51         self.acl_tbl_idx = {}
  52         self.pbr_vrfid = 200
  53
  54         # setup all interfaces
  55         for intf in self.interfaces:
  56             intf.admin_up()
  57             intf.config_ip4()
  58             intf.resolve_arp()
  59
  60     def tearDown(self):
  61         """Run standard test teardown and acl related log."""
  62         super(TestClassifier, self).tearDown()
  63         if not self.vpp_dead:
  64             self.logger.info(self.vapi.cli("show classify table verbose"))
  65             self.logger.info(self.vapi.cli("show ip fib"))
  66
  67     def config_pbr_fib_entry(self, intf):
  68         """Configure fib entry to route traffic toward PBR VRF table
  69
  70         :param VppInterface intf: destination interface to be routed for PBR.
  71
  72         """
  73         addr_len = 24
  74         self.vapi.ip_add_del_route(intf.local_ip4n,
  75                                    addr_len,
  76                                    intf.remote_ip4n,
  77                                    table_id=self.pbr_vrfid)
  78
  79     def create_stream(self, src_if, dst_if, packet_sizes):
  80         """Create input packet stream for defined interfaces.
  81
  82         :param VppInterface src_if: Source Interface for packet stream.
  83         :param VppInterface dst_if: Destination Interface for packet stream.
  84         :param list packet_sizes: packet size to test.
  85         """
  86         pkts = []
  87         for size in packet_sizes:
  88             info = self.create_packet_info(src_if, dst_if)
  89             payload = self.info_to_payload(info)
  90             p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
  91                  IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4) /
  92                  UDP(sport=1234, dport=5678) /
  93                  Raw(payload))
  94             info.data = p.copy()
  95             self.extend_packet(p, size)
  96             pkts.append(p)
  97         return pkts
  98
  99     def verify_capture(self, dst_if, capture):
 100         """Verify captured input packet stream for defined interface.
 101
 102         :param VppInterface dst_if: Interface to verify captured packet stream.
 103         :param list capture: Captured packet stream.
 104         """
 105         self.logger.info("Verifying capture on interface %s" % dst_if.name)
 106         last_info = dict()
 107         for i in self.interfaces:
 108             last_info[i.sw_if_index] = None
 109         dst_sw_if_index = dst_if.sw_if_index
 110         for packet in capture:
 111             try:
 112                 ip = packet[IP]
 113                 udp = packet[UDP]
 114                 payload_info = self.payload_to_info(str(packet[Raw]))
 115                 packet_index = payload_info.index
 116                 self.assertEqual(payload_info.dst, dst_sw_if_index)
 117                 self.logger.debug(
 118                     "Got packet on port %s: src=%u (id=%u)" %
 119                     (dst_if.name, payload_info.src, packet_index))
 120                 next_info = self.get_next_packet_info_for_interface2(
 121                     payload_info.src, dst_sw_if_index,
 122                     last_info[payload_info.src])
 123                 last_info[payload_info.src] = next_info
 124                 self.assertTrue(next_info is not None)
 125                 self.assertEqual(packet_index, next_info.index)
 126                 saved_packet = next_info.data
 127                 # Check standard fields
 128                 self.assertEqual(ip.src, saved_packet[IP].src)
 129                 self.assertEqual(ip.dst, saved_packet[IP].dst)
 130                 self.assertEqual(udp.sport, saved_packet[UDP].sport)
 131                 self.assertEqual(udp.dport, saved_packet[UDP].dport)
 132             except:
 133                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
 134                 raise
 135         for i in self.interfaces:
 136             remaining_packet = self.get_next_packet_info_for_interface2(
 137                 i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index])
 138             self.assertTrue(remaining_packet is None,
 139                             "Interface %s: Packet expected from interface %s "
 140                             "didn't arrive" % (dst_if.name, i.name))
 141
 142     @staticmethod
 143     def build_ip_mask(proto='', src_ip='', dst_ip='',
 144                       src_port='', dst_port=''):
 145         """Build IP ACL mask data with hexstring format
 146
 147         :param str proto: protocol number <0-ff>
 148         :param str src_ip: source ip address <0-ffffffff>
 149         :param str dst_ip: destination ip address <0-ffffffff>
 150         :param str src_port: source port number <0-ffff>
 151         :param str dst_port: destination port number <0-ffff>
 152         """
 153
 154         return ('{:0>20}{:0>12}{:0>8}{:0>12}{:0>4}'.format(
 155             proto, src_ip, dst_ip, src_port, dst_port)).rstrip('0')
 156
 157     @staticmethod
 158     def build_ip_match(proto='', src_ip='', dst_ip='',
 159                        src_port='', dst_port=''):
 160         """Build IP ACL match data with hexstring format
 161
 162         :param str proto: protocol number with valid option "<0-ff>"
 163         :param str src_ip: source ip address with format of "x.x.x.x"
 164         :param str dst_ip: destination ip address with format of "x.x.x.x"
 165         :param str src_port: source port number <0-ffff>
 166         :param str dst_port: destination port number <0-ffff>
 167         """
 168         if src_ip:
 169             src_ip = socket.inet_aton(src_ip).encode('hex')
 170         if dst_ip:
 171             dst_ip = socket.inet_aton(dst_ip).encode('hex')
 172
 173         return ('{:0>20}{:0>12}{:0>8}{:0>12}{:0>4}'.format(
 174             proto, src_ip, dst_ip, src_port, dst_port)).rstrip('0')
 175
 176     @staticmethod
 177     def build_mac_mask(dst_mac='', src_mac='', ether_type=''):
 178         """Build MAC ACL mask data with hexstring format
 179
 180         :param str dst_mac: source MAC address <0-ffffffffffff>
 181         :param str src_mac: destination MAC address <0-ffffffffffff>
 182         :param str ether_type: ethernet type <0-ffff>
 183         """
 184
 185         return ('{:0>12}{:0>12}{:0>4}'.format(dst_mac, src_mac,
 186                                               ether_type)).rstrip('0')
 187
 188     @staticmethod
 189     def build_mac_match(dst_mac='', src_mac='', ether_type=''):
 190         """Build MAC ACL match data with hexstring format
 191
 192         :param str dst_mac: source MAC address <x:x:x:x:x:x>
 193         :param str src_mac: destination MAC address <x:x:x:x:x:x>
 194         :param str ether_type: ethernet type <0-ffff>
 195         """
 196         if dst_mac:
 197             dst_mac = dst_mac.replace(':', '')
 198         if src_mac:
 199             src_mac = src_mac.replace(':', '')
 200
 201         return ('{:0>12}{:0>12}{:0>4}'.format(dst_mac, src_mac,
 202                                               ether_type)).rstrip('0')
 203
 204     def create_classify_table(self, key, mask, data_offset=0, is_add=1):
 205         """Create Classify Table
 206
 207         :param str key: key for classify table (ex, ACL name).
 208         :param str mask: mask value for interested traffic.
 209         :param int match_n_vectors:
 210         :param int is_add: option to configure classify table.
 211             - create(1) or delete(0)
 212         """
 213         r = self.vapi.classify_add_del_table(
 214             is_add,
 215             binascii.unhexlify(mask),
 216             match_n_vectors=(len(mask) - 1) // 32 + 1,
 217             miss_next_index=0,
 218             current_data_flag=1,
 219             current_data_offset=data_offset)
 220         self.assertIsNotNone(r, msg='No response msg for add_del_table')
 221         self.acl_tbl_idx[key] = r.new_table_index
 222
 223     def create_classify_session(self, intf, table_index, match,
 224                                 pbr_option=0, vrfid=0, is_add=1):
 225         """Create Classify Session
 226
 227         :param VppInterface intf: Interface to apply classify session.
 228         :param int table_index: table index to identify classify table.
 229         :param str match: matched value for interested traffic.
 230         :param int pbr_action: enable/disable PBR feature.
 231         :param int vrfid: VRF id.
 232         :param int is_add: option to configure classify session.
 233             - create(1) or delete(0)
 234         """
 235         r = self.vapi.classify_add_del_session(
 236             is_add,
 237             table_index,
 238             binascii.unhexlify(match),
 239             opaque_index=0,
 240             action=pbr_option,
 241             metadata=vrfid)
 242         self.assertIsNotNone(r, msg='No response msg for add_del_session')
 243
 244     def input_acl_set_interface(self, intf, table_index, is_add=1):
 245         """Configure Input ACL interface
 246
 247         :param VppInterface intf: Interface to apply Input ACL feature.
 248         :param int table_index: table index to identify classify table.
 249         :param int is_add: option to configure classify session.
 250             - enable(1) or disable(0)
 251         """
 252         r = self.vapi.input_acl_set_interface(
 253             is_add,
 254             intf.sw_if_index,
 255             ip4_table_index=table_index)
 256         self.assertIsNotNone(r, msg='No response msg for acl_set_interface')
 257
 258     def test_acl_ip(self):
 259         """ IP ACL test
 260
 261         Test scenario for basic IP ACL with source IP
 262             - Create IPv4 stream for pg0 -> pg1 interface.
 263             - Create ACL with source IP address.
 264             - Send and verify received packets on pg1 interface.
 265         """
 266
 267         # Basic ACL testing with source IP
 268         pkts = self.create_stream(self.pg0, self.pg1, self.pg_if_packet_sizes)
 269         self.pg0.add_stream(pkts)
 270
 271         self.create_classify_table('ip', self.build_ip_mask(src_ip='ffffffff'))
 272         self.create_classify_session(
 273             self.pg0, self.acl_tbl_idx.get('ip'),
 274             self.build_ip_match(src_ip=self.pg0.remote_ip4))
 275         self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('ip'))
 276
 277         self.pg_enable_capture(self.pg_interfaces)
 278         self.pg_start()
 279
 280         pkts = self.pg1.get_capture(len(pkts))
 281         self.verify_capture(self.pg1, pkts)
 282         self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('ip'), 0)
 283         self.pg0.assert_nothing_captured(remark="packets forwarded")
 284         self.pg2.assert_nothing_captured(remark="packets forwarded")
 285         self.pg3.assert_nothing_captured(remark="packets forwarded")
 286
 287     def test_acl_mac(self):
 288         """ MAC ACL test
 289
 290         Test scenario for basic MAC ACL with source MAC
 291             - Create IPv4 stream for pg0 -> pg2 interface.
 292             - Create ACL with source MAC address.
 293             - Send and verify received packets on pg2 interface.
 294         """
 295
 296         # Basic ACL testing with source MAC
 297         pkts = self.create_stream(self.pg0, self.pg2, self.pg_if_packet_sizes)
 298         self.pg0.add_stream(pkts)
 299
 300         self.create_classify_table('mac',
 301                                    self.build_mac_mask(src_mac='ffffffffffff'),
 302                                    data_offset=-14)
 303         self.create_classify_session(
 304             self.pg0, self.acl_tbl_idx.get('mac'),
 305             self.build_mac_match(src_mac=self.pg0.remote_mac))
 306         self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('mac'))
 307
 308         self.pg_enable_capture(self.pg_interfaces)
 309         self.pg_start()
 310
 311         pkts = self.pg2.get_capture(len(pkts))
 312         self.verify_capture(self.pg2, pkts)
 313         self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('mac'), 0)
 314         self.pg0.assert_nothing_captured(remark="packets forwarded")
 315         self.pg1.assert_nothing_captured(remark="packets forwarded")
 316         self.pg3.assert_nothing_captured(remark="packets forwarded")
 317
 318     def test_acl_pbr(self):
 319         """ IP PBR test
 320
 321         Test scenario for PBR with source IP
 322             - Create IPv4 stream for pg0 -> pg3 interface.
 323             - Configure PBR fib entry for packet forwarding.
 324             - Send and verify received packets on pg3 interface.
 325         """
 326
 327         # PBR testing with source IP
 328         pkts = self.create_stream(self.pg0, self.pg3, self.pg_if_packet_sizes)
 329         self.pg0.add_stream(pkts)
 330
 331         self.create_classify_table(
 332             'pbr', self.build_ip_mask(
 333                 src_ip='ffffffff'))
 334         pbr_option = 1
 335         self.create_classify_session(
 336             self.pg0, self.acl_tbl_idx.get('pbr'),
 337             self.build_ip_match(src_ip=self.pg0.remote_ip4),
 338             pbr_option, self.pbr_vrfid)
 339         self.config_pbr_fib_entry(self.pg3)
 340         self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('pbr'))
 341
 342         self.pg_enable_capture(self.pg_interfaces)
 343         self.pg_start()
 344
 345         pkts = self.pg3.get_capture(len(pkts))
 346         self.verify_capture(self.pg3, pkts)
 347         self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('pbr'), 0)
 348         self.pg0.assert_nothing_captured(remark="packets forwarded")
 349         self.pg1.assert_nothing_captured(remark="packets forwarded")
 350         self.pg2.assert_nothing_captured(remark="packets forwarded")
 351
 352
 353 if __name__ == '__main__':
 354     unittest.main(testRunner=VppTestRunner)