ryu/build/lib.linux-armv7l-2.7/ryu/tests/unit/packet/test_bfd.py

   1 # Copyright (C) 2014 Xinguard, Inc.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 # http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  12 # implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15
  16 import unittest
  17 import logging
  18 import struct
  19 import inspect
  20 from nose.tools import ok_, eq_, nottest
  21
  22 from ryu.ofproto import ether
  23 from ryu.ofproto import inet
  24 from ryu.lib.packet import packet
  25 from ryu.lib.packet import ethernet
  26 from ryu.lib.packet import ipv4
  27 from ryu.lib.packet import udp
  28 from ryu.lib.packet import bfd
  29 from ryu.lib import addrconv
  30
  31 LOG = logging.getLogger(__name__)
  32
  33
  34 class TestBFD(unittest.TestCase):
  35     def setUp(self):
  36         # BFD packet without authentication.
  37         self.data = b'\xb0\xa8\x6e\x18\xb8\x08\x64\x87' \
  38                     + b'\x88\xe9\xcb\xc8\x08\x00\x45\xc0' \
  39                     + b'\x00\x34\x68\x49\x00\x00\xff\x11' \
  40                     + b'\xf4\x73\xac\x1c\x03\x01\xac\x1c' \
  41                     + b'\x03\x02\xc0\x00\x0e\xc8\x00\x20' \
  42                     + b'\xd9\x02\x21\xc0\x03\x18\x00\x00' \
  43                     + b'\x00\x06\x00\x00\x00\x07\x00\x00' \
  44                     + b'\xea\x60\x00\x00\xea\x60\x00\x00' \
  45                     + b'\x00\x00'
  46
  47         # BFD packet using simple password authentication.
  48         self.data_auth_simple = b'\x08\x00\x27\xd1\x95\x7c\x08\x00' \
  49                                 + b'\x27\xed\x54\x41\x08\x00\x45\xc0' \
  50                                 + b'\x00\x3d\x0c\x90\x00\x00\xff\x11' \
  51                                 + b'\xbb\x0b\xc0\xa8\x39\x02\xc0\xa8' \
  52                                 + b'\x39\x01\xc0\x00\x0e\xc8\x00\x29' \
  53                                 + b'\x46\x35\x20\x44\x03\x21\x00\x00' \
  54                                 + b'\x00\x01\x00\x00\x00\x00\x00\x0f' \
  55                                 + b'\x42\x40\x00\x0f\x42\x40\x00\x00' \
  56                                 + b'\x00\x00\x01\x09\x02\x73\x65\x63' \
  57                                 + b'\x72\x65\x74'
  58
  59         # BFD packet using md5 authentication.
  60         self.data_auth_md5 = b'\x08\x00\x27\xd1\x95\x7c\x08\x00' \
  61                              + b'\x27\xed\x54\x41\x08\x00\x45\xc0' \
  62                              + b'\x00\x4c\x0c\x44\x00\x00\xff\x11' \
  63                              + b'\xbb\x48\xc0\xa8\x39\x02\xc0\xa8' \
  64                              + b'\x39\x01\xc0\x00\x0e\xc8\x00\x38' \
  65                              + b'\x51\xbc\x20\x44\x03\x30\x00\x00' \
  66                              + b'\x00\x01\x00\x00\x00\x00\x00\x0f' \
  67                              + b'\x42\x40\x00\x0f\x42\x40\x00\x00' \
  68                              + b'\x00\x00\x02\x18\x02\x00\x00\x00' \
  69                              + b'\x41\xdb\x66\xa8\xf9\x25\x5a\x8b' \
  70                              + b'\xcb\x7e\x4b\xec\x25\xa6\x2c\x23' \
  71                              + b'\xda\x0f'
  72
  73         # BFD packet using SHA1 authentication.
  74         self.data_auth_sha1 = b'\x08\x00\x27\xd1\x95\x7c\x08\x00' \
  75                               + b'\x27\xed\x54\x41\x08\x00\x45\xc0' \
  76                               + b'\x00\x50\x0b\x90\x00\x00\xff\x11' \
  77                               + b'\xbb\xf8\xc0\xa8\x39\x02\xc0\xa8' \
  78                               + b'\x39\x01\xc0\x00\x0e\xc8\x00\x3c' \
  79                               + b'\xb9\x92\x20\x44\x03\x34\x00\x00' \
  80                               + b'\x00\x01\x00\x00\x00\x00\x00\x0f' \
  81                               + b'\x42\x40\x00\x0f\x42\x40\x00\x00' \
  82                               + b'\x00\x00\x04\x1c\x02\x00\x00\x00' \
  83                               + b'\x41\xb1\x46\x20\x10\x81\x03\xd7' \
  84                               + b'\xf4\xde\x87\x61\x4c\x24\x61\x1f' \
  85                               + b'\x3c\xc1\x6a\x00\x69\x23'
  86
  87         # BFD Key chain {auth_key_id: auth_key/password}
  88         self.auth_keys = {2: b"secret"}
  89
  90     def tearDown(self):
  91         pass
  92
  93     def test_parse(self):
  94         buf = self.data
  95         pkt = packet.Packet(buf)
  96         i = iter(pkt)
  97
  98         eq_(type(next(i)), ethernet.ethernet)
  99         eq_(type(next(i)), ipv4.ipv4)
 100         eq_(type(next(i)), udp.udp)
 101         eq_(type(bfd.bfd.parser(next(i))[0]), bfd.bfd)
 102
 103     def test_parse_with_auth_simple(self):
 104         buf = self.data_auth_simple
 105         pkt = packet.Packet(buf)
 106         i = iter(pkt)
 107
 108         eq_(type(next(i)), ethernet.ethernet)
 109         eq_(type(next(i)), ipv4.ipv4)
 110         eq_(type(next(i)), udp.udp)
 111
 112         bfd_obj = bfd.bfd.parser(next(i))[0]
 113         eq_(type(bfd_obj), bfd.bfd)
 114         eq_(type(bfd_obj.auth_cls), bfd.SimplePassword)
 115         ok_(bfd_obj.authenticate(self.auth_keys))
 116
 117     def test_parse_with_auth_md5(self):
 118         buf = self.data_auth_md5
 119         pkt = packet.Packet(buf)
 120         i = iter(pkt)
 121
 122         eq_(type(next(i)), ethernet.ethernet)
 123         eq_(type(next(i)), ipv4.ipv4)
 124         eq_(type(next(i)), udp.udp)
 125
 126         bfd_obj = bfd.bfd.parser(next(i))[0]
 127         eq_(type(bfd_obj), bfd.bfd)
 128         eq_(type(bfd_obj.auth_cls), bfd.KeyedMD5)
 129         ok_(bfd_obj.authenticate(self.auth_keys))
 130
 131     def test_parse_with_auth_sha1(self):
 132         buf = self.data_auth_sha1
 133         pkt = packet.Packet(buf)
 134         i = iter(pkt)
 135
 136         eq_(type(next(i)), ethernet.ethernet)
 137         eq_(type(next(i)), ipv4.ipv4)
 138         eq_(type(next(i)), udp.udp)
 139
 140         bfd_obj = bfd.bfd.parser(next(i))[0]
 141         eq_(type(bfd_obj), bfd.bfd)
 142         eq_(type(bfd_obj.auth_cls), bfd.KeyedSHA1)
 143         ok_(bfd_obj.authenticate(self.auth_keys))
 144
 145     def test_serialize(self):
 146         pkt = packet.Packet()
 147
 148         eth_pkt = ethernet.ethernet('b0:a8:6e:18:b8:08', '64:87:88:e9:cb:c8')
 149         pkt.add_protocol(eth_pkt)
 150
 151         ip_pkt = ipv4.ipv4(src='172.28.3.1', dst='172.28.3.2', tos=192,
 152                            identification=26697, proto=inet.IPPROTO_UDP)
 153         pkt.add_protocol(ip_pkt)
 154
 155         udp_pkt = udp.udp(49152, 3784)
 156         pkt.add_protocol(udp_pkt)
 157
 158         bfd_pkt = bfd.bfd(ver=1, diag=bfd.BFD_DIAG_CTRL_DETECT_TIME_EXPIRED,
 159                           state=bfd.BFD_STATE_UP, detect_mult=3, my_discr=6,
 160                           your_discr=7, desired_min_tx_interval=60000,
 161                           required_min_rx_interval=60000,
 162                           required_min_echo_rx_interval=0)
 163         pkt.add_protocol(bfd_pkt)
 164
 165         eq_(len(pkt.protocols), 4)
 166
 167         pkt.serialize()
 168         eq_(pkt.data, self.data)
 169
 170     def test_serialize_with_auth_simple(self):
 171         pkt = packet.Packet()
 172
 173         eth_pkt = ethernet.ethernet('08:00:27:d1:95:7c', '08:00:27:ed:54:41')
 174         pkt.add_protocol(eth_pkt)
 175
 176         ip_pkt = ipv4.ipv4(src='192.168.57.2', dst='192.168.57.1', tos=192,
 177                            identification=3216, proto=inet.IPPROTO_UDP)
 178         pkt.add_protocol(ip_pkt)
 179
 180         udp_pkt = udp.udp(49152, 3784)
 181         pkt.add_protocol(udp_pkt)
 182
 183         auth_cls = bfd.SimplePassword(auth_key_id=2,
 184                                       password=self.auth_keys[2])
 185
 186         bfd_pkt = bfd.bfd(ver=1, diag=bfd.BFD_DIAG_NO_DIAG,
 187                           flags=bfd.BFD_FLAG_AUTH_PRESENT,
 188                           state=bfd.BFD_STATE_DOWN, detect_mult=3, my_discr=1,
 189                           your_discr=0, desired_min_tx_interval=1000000,
 190                           required_min_rx_interval=1000000,
 191                           required_min_echo_rx_interval=0,
 192                           auth_cls=auth_cls)
 193
 194         pkt.add_protocol(bfd_pkt)
 195
 196         eq_(len(pkt.protocols), 4)
 197
 198         pkt.serialize()
 199         eq_(pkt.data, self.data_auth_simple)
 200
 201     def test_serialize_with_auth_md5(self):
 202         pkt = packet.Packet()
 203
 204         eth_pkt = ethernet.ethernet('08:00:27:d1:95:7c', '08:00:27:ed:54:41')
 205         pkt.add_protocol(eth_pkt)
 206
 207         ip_pkt = ipv4.ipv4(src='192.168.57.2', dst='192.168.57.1', tos=192,
 208                            identification=3140, proto=inet.IPPROTO_UDP)
 209         pkt.add_protocol(ip_pkt)
 210
 211         udp_pkt = udp.udp(49152, 3784)
 212         pkt.add_protocol(udp_pkt)
 213
 214         auth_cls = bfd.KeyedMD5(auth_key_id=2, seq=16859,
 215                                 auth_key=self.auth_keys[2])
 216
 217         bfd_pkt = bfd.bfd(ver=1, diag=bfd.BFD_DIAG_NO_DIAG,
 218                           flags=bfd.BFD_FLAG_AUTH_PRESENT,
 219                           state=bfd.BFD_STATE_DOWN, detect_mult=3, my_discr=1,
 220                           your_discr=0, desired_min_tx_interval=1000000,
 221                           required_min_rx_interval=1000000,
 222                           required_min_echo_rx_interval=0,
 223                           auth_cls=auth_cls)
 224
 225         pkt.add_protocol(bfd_pkt)
 226
 227         eq_(len(pkt.protocols), 4)
 228
 229         pkt.serialize()
 230         eq_(pkt.data, self.data_auth_md5)
 231
 232     def test_serialize_with_auth_sha1(self):
 233         pkt = packet.Packet()
 234
 235         eth_pkt = ethernet.ethernet('08:00:27:d1:95:7c', '08:00:27:ed:54:41')
 236         pkt.add_protocol(eth_pkt)
 237
 238         ip_pkt = ipv4.ipv4(src='192.168.57.2', dst='192.168.57.1', tos=192,
 239                            identification=2960, proto=inet.IPPROTO_UDP)
 240         pkt.add_protocol(ip_pkt)
 241
 242         udp_pkt = udp.udp(49152, 3784)
 243         pkt.add_protocol(udp_pkt)
 244
 245         auth_cls = bfd.KeyedSHA1(auth_key_id=2, seq=16817,
 246                                  auth_key=self.auth_keys[2])
 247
 248         bfd_pkt = bfd.bfd(ver=1, diag=bfd.BFD_DIAG_NO_DIAG,
 249                           flags=bfd.BFD_FLAG_AUTH_PRESENT,
 250                           state=bfd.BFD_STATE_DOWN, detect_mult=3, my_discr=1,
 251                           your_discr=0, desired_min_tx_interval=1000000,
 252                           required_min_rx_interval=1000000,
 253                           required_min_echo_rx_interval=0,
 254                           auth_cls=auth_cls)
 255
 256         pkt.add_protocol(bfd_pkt)
 257
 258         eq_(len(pkt.protocols), 4)
 259
 260         pkt.serialize()
 261         eq_(pkt.data, self.data_auth_sha1)
 262
 263     def test_json(self):
 264         bfd1 = bfd.bfd(ver=1, diag=bfd.BFD_DIAG_CTRL_DETECT_TIME_EXPIRED,
 265                        state=bfd.BFD_STATE_UP, detect_mult=3, my_discr=6,
 266                        your_discr=7, desired_min_tx_interval=60000,
 267                        required_min_rx_interval=60000,
 268                        required_min_echo_rx_interval=0)
 269
 270         jsondict = bfd1.to_jsondict()
 271         bfd2 = bfd.bfd.from_jsondict(jsondict['bfd'])
 272         eq_(str(bfd1), str(bfd2))
 273
 274     def test_json_with_auth_simple(self):
 275         auth_cls = bfd.SimplePassword(auth_key_id=2,
 276                                       password=self.auth_keys[2])
 277
 278         bfd1 = bfd.bfd(ver=1, diag=bfd.BFD_DIAG_NO_DIAG,
 279                        flags=bfd.BFD_FLAG_AUTH_PRESENT,
 280                        state=bfd.BFD_STATE_DOWN, detect_mult=3, my_discr=1,
 281                        your_discr=0, desired_min_tx_interval=1000000,
 282                        required_min_rx_interval=1000000,
 283                        required_min_echo_rx_interval=0,
 284                        auth_cls=auth_cls)
 285
 286         jsondict = bfd1.to_jsondict()
 287         bfd2 = bfd.bfd.from_jsondict(jsondict['bfd'])
 288         eq_(str(bfd1), str(bfd2))
 289
 290     def test_json_with_auth_md5(self):
 291         auth_cls = bfd.KeyedMD5(auth_key_id=2, seq=16859,
 292                                 auth_key=self.auth_keys[2])
 293
 294         bfd1 = bfd.bfd(ver=1, diag=bfd.BFD_DIAG_NO_DIAG,
 295                        flags=bfd.BFD_FLAG_AUTH_PRESENT,
 296                        state=bfd.BFD_STATE_DOWN, detect_mult=3, my_discr=1,
 297                        your_discr=0, desired_min_tx_interval=1000000,
 298                        required_min_rx_interval=1000000,
 299                        required_min_echo_rx_interval=0,
 300                        auth_cls=auth_cls)
 301
 302         jsondict = bfd1.to_jsondict()
 303         bfd2 = bfd.bfd.from_jsondict(jsondict['bfd'])
 304         eq_(str(bfd1), str(bfd2))
 305
 306     def test_json_with_auth_sha1(self):
 307         auth_cls = bfd.KeyedSHA1(auth_key_id=2, seq=16859,
 308                                  auth_key=self.auth_keys[2])
 309
 310         bfd1 = bfd.bfd(ver=1, diag=bfd.BFD_DIAG_NO_DIAG,
 311                        flags=bfd.BFD_FLAG_AUTH_PRESENT,
 312                        state=bfd.BFD_STATE_DOWN, detect_mult=3, my_discr=1,
 313                        your_discr=0, desired_min_tx_interval=1000000,
 314                        required_min_rx_interval=1000000,
 315                        required_min_echo_rx_interval=0,
 316                        auth_cls=auth_cls)
 317
 318         jsondict = bfd1.to_jsondict()
 319         bfd2 = bfd.bfd.from_jsondict(jsondict['bfd'])
 320         eq_(str(bfd1), str(bfd2))