test/test_punt.py

   1 #!/usr/bin/env python
   2 import binascii
   3 import random
   4 import socket
   5 import unittest
   6 import os
   7 import scapy.layers.inet6 as inet6
   8
   9 from util import ppp, ppc
  10 from re import compile
  11 from scapy.packet import Raw
  12 from scapy.layers.l2 import Ether
  13 from scapy.layers.inet import IP, UDP, ICMP
  14 from scapy.layers.inet6 import IPv6, ICMPv6DestUnreach
  15 from framework import VppTestCase, VppTestRunner
  16
  17
  18 class TestPuntSocket(VppTestCase):
  19     """ Punt Socket """
  20
  21     tempdir = ""
  22     sock = None
  23     err_ptr = compile(r"^([\d]+)\s+([-\w]+)\s+([ -\.\w)(]+)$")
  24
  25     @classmethod
  26     def setUpConstants(cls):
  27         tempdir = cls.tempdir
  28         cls.extra_vpp_punt_config = [
  29             "punt", "{", "socket", cls.tempdir+"/socket_punt", "}"]
  30         super(TestPuntSocket, cls).setUpConstants()
  31
  32     def process_cli(self, exp, ptr):
  33         for line in self.vapi.cli(exp).split('\n')[1:]:
  34             m = ptr.match(line.strip())
  35             if m:
  36                 yield m.groups()
  37
  38     def show_errors(self):
  39         for pack in self.process_cli("show errors", self.err_ptr):
  40             try:
  41                 count, node, reason = pack
  42             except ValueError:
  43                 pass
  44             else:
  45                 yield count, node, reason
  46
  47     def get_punt_count(self, counter):
  48         errors = list(self.show_errors())
  49         for count, node, reason in errors:
  50             if (node == counter and
  51                     reason == u'Socket TX'):
  52                 return int(count)
  53         return 0
  54
  55     def socket_client_create(self, sock_name):
  56         self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
  57         try:
  58             os.unlink(sock_name)
  59         except:
  60             self.logger.debug("Unlink socket faild")
  61         self.sock.bind(sock_name)
  62
  63     def socket_client_close(self):
  64         self.sock.close()
  65
  66
  67 class TestIP4PuntSocket(TestPuntSocket):
  68     """ Punt Socket for IPv4 """
  69
  70     def setUp(self):
  71         super(TestIP4PuntSocket, self).setUp()
  72
  73         self.create_pg_interfaces(range(2))
  74
  75         for i in self.pg_interfaces:
  76             i.admin_up()
  77             i.config_ip4()
  78             i.resolve_arp()
  79
  80     def tearDown(self):
  81         super(TestIP4PuntSocket, self).tearDown()
  82         for i in self.pg_interfaces:
  83             i.unconfig_ip4()
  84             i.admin_down()
  85
  86     def test_punt_socket_dump(self):
  87         """ Punt socket registration"""
  88
  89         punts = self.vapi.punt_socket_dump(0)
  90         self.assertEqual(len(punts), 0)
  91
  92         #
  93         # configure a punt socket
  94         #
  95         self.vapi.punt_socket_register(1111, self.tempdir+"/socket_punt_1111")
  96         self.vapi.punt_socket_register(2222, self.tempdir+"/socket_punt_2222")
  97         punts = self.vapi.punt_socket_dump(0)
  98         self.assertEqual(len(punts), 2)
  99         self.assertEqual(punts[0].punt.l4_port, 1111)
 100         # self.assertEqual(punts[0].pathname, "/tmp/punt_socket_udp_1234")
 101         self.assertEqual(punts[1].punt.l4_port, 2222)
 102         # self.assertEqual(punts[1].pathname, "/tmp/punt_socket_udp_5678")
 103
 104         #
 105         # deregister a punt socket
 106         #
 107         self.vapi.punt_socket_deregister(1111)
 108         punts = self.vapi.punt_socket_dump(0)
 109         self.assertEqual(len(punts), 1)
 110
 111         #
 112         # configure a punt socket again
 113         #
 114         self.vapi.punt_socket_register(1111, self.tempdir+"/socket_punt_1111")
 115         self.vapi.punt_socket_register(3333, self.tempdir+"/socket_punt_3333")
 116         punts = self.vapi.punt_socket_dump(0)
 117         self.assertEqual(len(punts), 3)
 118
 119         #
 120         # deregister all punt socket
 121         #
 122         self.vapi.punt_socket_deregister(1111)
 123         self.vapi.punt_socket_deregister(2222)
 124         self.vapi.punt_socket_deregister(3333)
 125         punts = self.vapi.punt_socket_dump(0)
 126         self.assertEqual(len(punts), 0)
 127
 128     def test_punt_socket_traffic(self):
 129         """ Punt socket traffic"""
 130
 131         nr_packets = 8
 132         p = (Ether(src=self.pg0.remote_mac,
 133                    dst=self.pg0.local_mac) /
 134              IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) /
 135              UDP(sport=9876, dport=1234) /
 136              Raw('\xa5' * 100))
 137
 138         pkts = p * nr_packets
 139
 140         punts = self.vapi.punt_socket_dump(0)
 141         self.assertEqual(len(punts), 0)
 142
 143         #
 144         # expect ICMP - port unreachable for all packets
 145         #
 146         self.vapi.cli("clear trace")
 147         self.pg0.add_stream(pkts)
 148         self.pg_enable_capture(self.pg_interfaces)
 149         self.pg_start()
 150         rx = self.pg0.get_capture(nr_packets)
 151         for p in rx:
 152             self.assertEqual(int(p[IP].proto), 1)   # ICMP
 153             self.assertEqual(int(p[ICMP].code), 3)  # unreachable
 154
 155         #
 156         # configure a punt socket
 157         #
 158         self.socket_client_create(self.tempdir+"/socket_punt_1234")
 159         self.vapi.punt_socket_register(1234, self.tempdir+"/socket_punt_1234")
 160         punts = self.vapi.punt_socket_dump(0)
 161         self.assertEqual(len(punts), 1)
 162
 163         #
 164         # expect punt socket and no packets on pg0
 165         #
 166         self.vapi.cli("clear errors")
 167         self.pg0.add_stream(pkts)
 168         self.pg_enable_capture(self.pg_interfaces)
 169         self.pg_start()
 170         self.pg0.get_capture(0)
 171         self.socket_client_close()
 172
 173         #
 174         # remove punt socket. expect ICMP - port unreachable for all packets
 175         #
 176         self.vapi.punt_socket_deregister(1234)
 177         punts = self.vapi.punt_socket_dump(0)
 178         self.assertEqual(len(punts), 0)
 179         self.pg0.add_stream(pkts)
 180         self.pg_enable_capture(self.pg_interfaces)
 181         self.pg_start()
 182         # FIXME - when punt socket deregister is implemented
 183         # self.pg0.get_capture(nr_packets)
 184
 185
 186 class TestIP6PuntSocket(TestPuntSocket):
 187     """ Punt Socket for IPv6"""
 188
 189     def setUp(self):
 190         super(TestIP6PuntSocket, self).setUp()
 191
 192         self.create_pg_interfaces(range(2))
 193
 194         for i in self.pg_interfaces:
 195             i.admin_up()
 196             i.config_ip6()
 197             i.resolve_ndp()
 198
 199     def tearDown(self):
 200         super(TestIP6PuntSocket, self).tearDown()
 201         for i in self.pg_interfaces:
 202             i.unconfig_ip6()
 203             i.admin_down()
 204
 205     def test_punt_socket_dump(self):
 206         """ Punt socket registration """
 207
 208         punts = self.vapi.punt_socket_dump(0)
 209         self.assertEqual(len(punts), 0)
 210
 211         #
 212         # configure a punt socket
 213         #
 214         self.vapi.punt_socket_register(1111, self.tempdir+"/socket_punt_1111",
 215                                        is_ip4=0)
 216         self.vapi.punt_socket_register(2222, self.tempdir+"/socket_punt_2222",
 217                                        is_ip4=0)
 218         punts = self.vapi.punt_socket_dump(1)
 219         self.assertEqual(len(punts), 2)
 220         self.assertEqual(punts[0].punt.l4_port, 1111)
 221         # self.assertEqual(punts[0].pathname, "/tmp/punt_socket_udp_1234")
 222         self.assertEqual(punts[1].punt.l4_port, 2222)
 223         # self.assertEqual(punts[1].pathname, "/tmp/punt_socket_udp_5678")
 224
 225         #
 226         # deregister a punt socket
 227         #
 228         self.vapi.punt_socket_deregister(1111, is_ip4=0)
 229         punts = self.vapi.punt_socket_dump(1)
 230         self.assertEqual(len(punts), 1)
 231
 232         #
 233         # configure a punt socket again
 234         #
 235         self.vapi.punt_socket_register(1111, self.tempdir+"/socket_punt_1111",
 236                                        is_ip4=0)
 237         punts = self.vapi.punt_socket_dump(1)
 238         self.assertEqual(len(punts), 2)
 239
 240         #
 241         # deregister all punt socket
 242         #
 243         self.vapi.punt_socket_deregister(1111, is_ip4=0)
 244         self.vapi.punt_socket_deregister(2222, is_ip4=0)
 245         self.vapi.punt_socket_deregister(3333, is_ip4=0)
 246         punts = self.vapi.punt_socket_dump(1)
 247         self.assertEqual(len(punts), 0)
 248
 249     def test_punt_socket_traffic(self):
 250         """ Punt socket traffic"""
 251
 252         nr_packets = 2
 253         p = (Ether(src=self.pg0.remote_mac,
 254                    dst=self.pg0.local_mac) /
 255              IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6) /
 256              inet6.UDP(sport=9876, dport=1234) /
 257              Raw('\xa5' * 100))
 258
 259         pkts = p * nr_packets
 260
 261         punts = self.vapi.punt_socket_dump(1)
 262         self.assertEqual(len(punts), 0)
 263
 264         #
 265         # expect ICMPv6 - destination unreachable for all packets
 266         #
 267         self.vapi.cli("clear trace")
 268         self.pg0.add_stream(pkts)
 269         self.pg_enable_capture(self.pg_interfaces)
 270         self.pg_start()
 271         rx = self.pg0.get_capture(nr_packets)
 272         for p in rx:
 273             self.assertEqual(int(p[IPv6].nh), 58)                # ICMPv6
 274             self.assertEqual(int(p[ICMPv6DestUnreach].code), 4)  # unreachable
 275
 276         #
 277         # configure a punt socket
 278         #
 279         self.socket_client_create(self.tempdir+"/socket_punt_1234")
 280         self.vapi.punt_socket_register(1234, self.tempdir+"/socket_punt_1234",
 281                                        is_ip4=0)
 282         punts = self.vapi.punt_socket_dump(1)
 283         self.assertEqual(len(punts), 1)
 284
 285         #
 286         # expect punt socket and no packets on pg0
 287         #
 288         self.vapi.cli("clear errors")
 289         self.pg0.add_stream(pkts)
 290         self.pg_enable_capture(self.pg_interfaces)
 291         self.pg_start()
 292         self.pg0.get_capture(0)
 293         self.socket_client_close()
 294
 295         #
 296         # remove punt socket. expect ICMP - dest. unreachable for all packets
 297         #
 298         self.vapi.punt_socket_deregister(1234, is_ip4=0)
 299         punts = self.vapi.punt_socket_dump(1)
 300         self.assertEqual(len(punts), 0)
 301         self.pg0.add_stream(pkts)
 302         self.pg_enable_capture(self.pg_interfaces)
 303         self.pg_start()
 304         # FIXME - when punt socket deregister is implemented
 305 #        self.pg0.get_capture(nr_packets)
 306
 307
 308 if __name__ == '__main__':
 309     unittest.main(testRunner=VppTestRunner)