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("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(
 154             proto, src_ip, 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:
 168             src_ip = socket.inet_aton(src_ip).encode('hex')
 169         if dst_ip:
 170             dst_ip = socket.inet_aton(dst_ip).encode('hex')
 171
 172         return ('{:0>20}{:0>12}{:0>8}{:0>12}{:0>4}'.format(
 173             proto, src_ip, dst_ip, src_port, dst_port)).rstrip('0')
 174
 175     @staticmethod
 176     def build_mac_mask(dst_mac='', src_mac='', ether_type=''):
 177         """Build MAC ACL mask data with hexstring format
 178
 179         :param str dst_mac: source MAC address <0-ffffffffffff>
 180         :param str src_mac: destination MAC address <0-ffffffffffff>
 181         :param str ether_type: ethernet type <0-ffff>
 182         """
 183
 184         return ('{:0>12}{:0>12}{:0>4}'.format(dst_mac, src_mac,
 185                                               ether_type)).rstrip('0')
 186
 187     @staticmethod
 188     def build_mac_match(dst_mac='', src_mac='', ether_type=''):
 189         """Build MAC ACL match data with hexstring format
 190
 191         :param str dst_mac: source MAC address <x:x:x:x:x:x>
 192         :param str src_mac: destination MAC address <x:x:x:x:x:x>
 193         :param str ether_type: ethernet type <0-ffff>
 194         """
 195         if dst_mac:
 196             dst_mac = dst_mac.replace(':', '')
 197         if src_mac:
 198             src_mac = src_mac.replace(':', '')
 199
 200         return ('{:0>12}{:0>12}{:0>4}'.format(dst_mac, src_mac,
 201                                               ether_type)).rstrip('0')
 202
 203     def create_classify_table(self, key, mask, data_offset=0, is_add=1):
 204         """Create Classify Table
 205
 206         :param str key: key for classify table (ex, ACL name).
 207         :param str mask: mask value for interested traffic.
 208         :param int match_n_vectors:
 209         :param int is_add: option to configure classify table.
 210             - create(1) or delete(0)
 211         """
 212         r = self.vapi.classify_add_del_table(
 213             is_add,
 214             binascii.unhexlify(mask),
 215             match_n_vectors=(len(mask) - 1) // 32 + 1,
 216             miss_next_index=0,
 217             current_data_flag=1,
 218             current_data_offset=data_offset)
 219         self.assertIsNotNone(r, msg='No response msg for add_del_table')
 220         self.acl_tbl_idx[key] = r.new_table_index
 221
 222     def create_classify_session(self, intf, table_index, match,
 223                                 pbr_option=0, vrfid=0, is_add=1):
 224         """Create Classify Session
 225
 226         :param VppInterface intf: Interface to apply classify session.
 227         :param int table_index: table index to identify classify table.
 228         :param str match: matched value for interested traffic.
 229         :param int pbr_action: enable/disable PBR feature.
 230         :param int vrfid: VRF id.
 231         :param int is_add: option to configure classify session.
 232             - create(1) or delete(0)
 233         """
 234         r = self.vapi.classify_add_del_session(
 235             is_add,
 236             table_index,
 237             binascii.unhexlify(match),
 238             opaque_index=0,
 239             action=pbr_option,
 240             metadata=vrfid)
 241         self.assertIsNotNone(r, msg='No response msg for add_del_session')
 242
 243     def input_acl_set_interface(self, intf, table_index, is_add=1):
 244         """Configure Input ACL interface
 245
 246         :param VppInterface intf: Interface to apply Input ACL feature.
 247         :param int table_index: table index to identify classify table.
 248         :param int is_add: option to configure classify session.
 249             - enable(1) or disable(0)
 250         """
 251         r = self.vapi.input_acl_set_interface(
 252             is_add,
 253             intf.sw_if_index,
 254             ip4_table_index=table_index)
 255         self.assertIsNotNone(r, msg='No response msg for acl_set_interface')
 256
 257     def test_acl_ip(self):
 258         """ IP ACL test
 259
 260         Test scenario for basic IP ACL with source IP
 261             - Create IPv4 stream for pg0 -> pg1 interface.
 262             - Create ACL with source IP address.
 263             - Send and verify received packets on pg1 interface.
 264         """
 265
 266         # Basic ACL testing with source IP
 267         pkts = self.create_stream(self.pg0, self.pg1, self.pg_if_packet_sizes)
 268         self.pg0.add_stream(pkts)
 269
 270         self.create_classify_table('ip', self.build_ip_mask(src_ip='ffffffff'))
 271         self.create_classify_session(
 272             self.pg0, self.acl_tbl_idx.get('ip'),
 273             self.build_ip_match(src_ip=self.pg0.remote_ip4))
 274         self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('ip'))
 275
 276         self.pg_enable_capture(self.pg_interfaces)
 277         self.pg_start()
 278
 279         pkts = self.pg1.get_capture(len(pkts))
 280         self.verify_capture(self.pg1, pkts)
 281         self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('ip'), 0)
 282         self.pg0.assert_nothing_captured(remark="packets forwarded")
 283         self.pg2.assert_nothing_captured(remark="packets forwarded")
 284         self.pg3.assert_nothing_captured(remark="packets forwarded")
 285
 286     def test_acl_mac(self):
 287         """ MAC ACL test
 288
 289         Test scenario for basic MAC ACL with source MAC
 290             - Create IPv4 stream for pg0 -> pg2 interface.
 291             - Create ACL with source MAC address.
 292             - Send and verify received packets on pg2 interface.
 293         """
 294
 295         # Basic ACL testing with source MAC
 296         pkts = self.create_stream(self.pg0, self.pg2, self.pg_if_packet_sizes)
 297         self.pg0.add_stream(pkts)
 298
 299         self.create_classify_table(
 300             'mac', self.build_mac_mask(src_mac='ffffffffffff'), data_offset=-14)
 301         self.create_classify_session(
 302             self.pg0, self.acl_tbl_idx.get('mac'),
 303             self.build_mac_match(src_mac=self.pg0.remote_mac))
 304         self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('mac'))
 305
 306         self.pg_enable_capture(self.pg_interfaces)
 307         self.pg_start()
 308
 309         pkts = self.pg2.get_capture(len(pkts))
 310         self.verify_capture(self.pg2, pkts)
 311         self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('mac'), 0)
 312         self.pg0.assert_nothing_captured(remark="packets forwarded")
 313         self.pg1.assert_nothing_captured(remark="packets forwarded")
 314         self.pg3.assert_nothing_captured(remark="packets forwarded")
 315
 316     def test_acl_pbr(self):
 317         """ IP PBR test
 318
 319         Test scenario for PBR with source IP
 320             - Create IPv4 stream for pg0 -> pg3 interface.
 321             - Configure PBR fib entry for packet forwarding.
 322             - Send and verify received packets on pg3 interface.
 323         """
 324
 325         # PBR testing with source IP
 326         pkts = self.create_stream(self.pg0, self.pg3, self.pg_if_packet_sizes)
 327         self.pg0.add_stream(pkts)
 328
 329         self.create_classify_table('pbr', self.build_ip_mask(src_ip='ffffffff'))
 330         pbr_option = 1
 331         self.create_classify_session(
 332             self.pg0, self.acl_tbl_idx.get('pbr'),
 333             self.build_ip_match(src_ip=self.pg0.remote_ip4),
 334             pbr_option, self.pbr_vrfid)
 335         self.config_pbr_fib_entry(self.pg3)
 336         self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('pbr'))
 337
 338         self.pg_enable_capture(self.pg_interfaces)
 339         self.pg_start()
 340
 341         pkts = self.pg3.get_capture(len(pkts))
 342         self.verify_capture(self.pg3, pkts)
 343         self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('pbr'), 0)
 344         self.pg0.assert_nothing_captured(remark="packets forwarded")
 345         self.pg1.assert_nothing_captured(remark="packets forwarded")
 346         self.pg2.assert_nothing_captured(remark="packets forwarded")
 347
 348
 349 if __name__ == '__main__':
 350     unittest.main(testRunner=VppTestRunner)