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