test/test_bfd.py

   1 #!/usr/bin/env python
   2 """ BFD tests """
   3
   4 from __future__ import division
   5
   6 import binascii
   7 import hashlib
   8 import time
   9 import unittest
  10 from random import randint, shuffle, getrandbits
  11 from socket import AF_INET, AF_INET6, inet_ntop
  12 from struct import pack, unpack
  13
  14 from six import moves
  15 from scapy.layers.inet import UDP, IP
  16 from scapy.layers.inet6 import IPv6
  17 from scapy.layers.l2 import Ether
  18 from scapy.packet import Raw
  19
  20 from bfd import VppBFDAuthKey, BFD, BFDAuthType, VppBFDUDPSession, \
  21     BFDDiagCode, BFDState, BFD_vpp_echo
  22 from framework import VppTestCase, VppTestRunner, running_extended_tests
  23 from util import ppp
  24 from vpp_ip import DpoProto
  25 from vpp_ip_route import VppIpRoute, VppRoutePath
  26 from vpp_lo_interface import VppLoInterface
  27 from vpp_papi_provider import UnexpectedApiReturnValueError
  28 from vpp_pg_interface import CaptureTimeoutError, is_ipv6_misc
  29
  30 USEC_IN_SEC = 1000000
  31
  32
  33 class AuthKeyFactory(object):
  34     """Factory class for creating auth keys with unique conf key ID"""
  35
  36     def __init__(self):
  37         self._conf_key_ids = {}
  38
  39     def create_random_key(self, test, auth_type=BFDAuthType.keyed_sha1):
  40         """ create a random key with unique conf key id """
  41         conf_key_id = randint(0, 0xFFFFFFFF)
  42         while conf_key_id in self._conf_key_ids:
  43             conf_key_id = randint(0, 0xFFFFFFFF)
  44         self._conf_key_ids[conf_key_id] = 1
  45         key = str(bytearray([randint(0, 255) for _ in range(randint(1, 20))]))
  46         return VppBFDAuthKey(test=test, auth_type=auth_type,
  47                              conf_key_id=conf_key_id, key=key)
  48
  49
  50 @unittest.skipUnless(running_extended_tests, "part of extended tests")
  51 class BFDAPITestCase(VppTestCase):
  52     """Bidirectional Forwarding Detection (BFD) - API"""
  53
  54     pg0 = None
  55     pg1 = None
  56
  57     @classmethod
  58     def setUpClass(cls):
  59         super(BFDAPITestCase, cls).setUpClass()
  60         cls.vapi.cli("set log class bfd level debug")
  61         try:
  62             cls.create_pg_interfaces(range(2))
  63             for i in cls.pg_interfaces:
  64                 i.config_ip4()
  65                 i.config_ip6()
  66                 i.resolve_arp()
  67
  68         except Exception:
  69             super(BFDAPITestCase, cls).tearDownClass()
  70             raise
  71
  72     def setUp(self):
  73         super(BFDAPITestCase, self).setUp()
  74         self.factory = AuthKeyFactory()
  75
  76     def test_add_bfd(self):
  77         """ create a BFD session """
  78         session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
  79         session.add_vpp_config()
  80         self.logger.debug("Session state is %s", session.state)
  81         session.remove_vpp_config()
  82         session.add_vpp_config()
  83         self.logger.debug("Session state is %s", session.state)
  84         session.remove_vpp_config()
  85
  86     def test_double_add(self):
  87         """ create the same BFD session twice (negative case) """
  88         session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
  89         session.add_vpp_config()
  90
  91         with self.vapi.assert_negative_api_retval():
  92             session.add_vpp_config()
  93
  94         session.remove_vpp_config()
  95
  96     def test_add_bfd6(self):
  97         """ create IPv6 BFD session """
  98         session = VppBFDUDPSession(
  99             self, self.pg0, self.pg0.remote_ip6, af=AF_INET6)
 100         session.add_vpp_config()
 101         self.logger.debug("Session state is %s", session.state)
 102         session.remove_vpp_config()
 103         session.add_vpp_config()
 104         self.logger.debug("Session state is %s", session.state)
 105         session.remove_vpp_config()
 106
 107     def test_mod_bfd(self):
 108         """ modify BFD session parameters """
 109         session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4,
 110                                    desired_min_tx=50000,
 111                                    required_min_rx=10000,
 112                                    detect_mult=1)
 113         session.add_vpp_config()
 114         s = session.get_bfd_udp_session_dump_entry()
 115         self.assert_equal(session.desired_min_tx,
 116                           s.desired_min_tx,
 117                           "desired min transmit interval")
 118         self.assert_equal(session.required_min_rx,
 119                           s.required_min_rx,
 120                           "required min receive interval")
 121         self.assert_equal(session.detect_mult, s.detect_mult, "detect mult")
 122         session.modify_parameters(desired_min_tx=session.desired_min_tx * 2,
 123                                   required_min_rx=session.required_min_rx * 2,
 124                                   detect_mult=session.detect_mult * 2)
 125         s = session.get_bfd_udp_session_dump_entry()
 126         self.assert_equal(session.desired_min_tx,
 127                           s.desired_min_tx,
 128                           "desired min transmit interval")
 129         self.assert_equal(session.required_min_rx,
 130                           s.required_min_rx,
 131                           "required min receive interval")
 132         self.assert_equal(session.detect_mult, s.detect_mult, "detect mult")
 133
 134     def test_add_sha1_keys(self):
 135         """ add SHA1 keys """
 136         key_count = 10
 137         keys = [self.factory.create_random_key(
 138             self) for i in range(0, key_count)]
 139         for key in keys:
 140             self.assertFalse(key.query_vpp_config())
 141         for key in keys:
 142             key.add_vpp_config()
 143         for key in keys:
 144             self.assertTrue(key.query_vpp_config())
 145         # remove randomly
 146         indexes = range(key_count)
 147         shuffle(indexes)
 148         removed = []
 149         for i in indexes:
 150             key = keys[i]
 151             key.remove_vpp_config()
 152             removed.append(i)
 153             for j in range(key_count):
 154                 key = keys[j]
 155                 if j in removed:
 156                     self.assertFalse(key.query_vpp_config())
 157                 else:
 158                     self.assertTrue(key.query_vpp_config())
 159         # should be removed now
 160         for key in keys:
 161             self.assertFalse(key.query_vpp_config())
 162         # add back and remove again
 163         for key in keys:
 164             key.add_vpp_config()
 165         for key in keys:
 166             self.assertTrue(key.query_vpp_config())
 167         for key in keys:
 168             key.remove_vpp_config()
 169         for key in keys:
 170             self.assertFalse(key.query_vpp_config())
 171
 172     def test_add_bfd_sha1(self):
 173         """ create a BFD session (SHA1) """
 174         key = self.factory.create_random_key(self)
 175         key.add_vpp_config()
 176         session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4,
 177                                    sha1_key=key)
 178         session.add_vpp_config()
 179         self.logger.debug("Session state is %s", session.state)
 180         session.remove_vpp_config()
 181         session.add_vpp_config()
 182         self.logger.debug("Session state is %s", session.state)
 183         session.remove_vpp_config()
 184
 185     def test_double_add_sha1(self):
 186         """ create the same BFD session twice (negative case) (SHA1) """
 187         key = self.factory.create_random_key(self)
 188         key.add_vpp_config()
 189         session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4,
 190                                    sha1_key=key)
 191         session.add_vpp_config()
 192         with self.assertRaises(Exception):
 193             session.add_vpp_config()
 194
 195     def test_add_auth_nonexistent_key(self):
 196         """ create BFD session using non-existent SHA1 (negative case) """
 197         session = VppBFDUDPSession(
 198             self, self.pg0, self.pg0.remote_ip4,
 199             sha1_key=self.factory.create_random_key(self))
 200         with self.assertRaises(Exception):
 201             session.add_vpp_config()
 202
 203     def test_shared_sha1_key(self):
 204         """ share single SHA1 key between multiple BFD sessions """
 205         key = self.factory.create_random_key(self)
 206         key.add_vpp_config()
 207         sessions = [
 208             VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4,
 209                              sha1_key=key),
 210             VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip6,
 211                              sha1_key=key, af=AF_INET6),
 212             VppBFDUDPSession(self, self.pg1, self.pg1.remote_ip4,
 213                              sha1_key=key),
 214             VppBFDUDPSession(self, self.pg1, self.pg1.remote_ip6,
 215                              sha1_key=key, af=AF_INET6)]
 216         for s in sessions:
 217             s.add_vpp_config()
 218         removed = 0
 219         for s in sessions:
 220             e = key.get_bfd_auth_keys_dump_entry()
 221             self.assert_equal(e.use_count, len(sessions) - removed,
 222                               "Use count for shared key")
 223             s.remove_vpp_config()
 224             removed += 1
 225         e = key.get_bfd_auth_keys_dump_entry()
 226         self.assert_equal(e.use_count, len(sessions) - removed,
 227                           "Use count for shared key")
 228
 229     def test_activate_auth(self):
 230         """ activate SHA1 authentication """
 231         key = self.factory.create_random_key(self)
 232         key.add_vpp_config()
 233         session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
 234         session.add_vpp_config()
 235         session.activate_auth(key)
 236
 237     def test_deactivate_auth(self):
 238         """ deactivate SHA1 authentication """
 239         key = self.factory.create_random_key(self)
 240         key.add_vpp_config()
 241         session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
 242         session.add_vpp_config()
 243         session.activate_auth(key)
 244         session.deactivate_auth()
 245
 246     def test_change_key(self):
 247         """ change SHA1 key """
 248         key1 = self.factory.create_random_key(self)
 249         key2 = self.factory.create_random_key(self)
 250         while key2.conf_key_id == key1.conf_key_id:
 251             key2 = self.factory.create_random_key(self)
 252         key1.add_vpp_config()
 253         key2.add_vpp_config()
 254         session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4,
 255                                    sha1_key=key1)
 256         session.add_vpp_config()
 257         session.activate_auth(key2)
 258
 259     def test_set_del_udp_echo_source(self):
 260         """ set/del udp echo source """
 261         self.create_loopback_interfaces(1)
 262         self.loopback0 = self.lo_interfaces[0]
 263         self.loopback0.admin_up()
 264         echo_source = self.vapi.bfd_udp_get_echo_source()
 265         self.assertFalse(echo_source.is_set)
 266         self.assertFalse(echo_source.have_usable_ip4)
 267         self.assertFalse(echo_source.have_usable_ip6)
 268
 269         self.vapi.bfd_udp_set_echo_source(self.loopback0.sw_if_index)
 270         echo_source = self.vapi.bfd_udp_get_echo_source()
 271         self.assertTrue(echo_source.is_set)
 272         self.assertEqual(echo_source.sw_if_index, self.loopback0.sw_if_index)
 273         self.assertFalse(echo_source.have_usable_ip4)
 274         self.assertFalse(echo_source.have_usable_ip6)
 275
 276         self.loopback0.config_ip4()
 277         unpacked = unpack("!L", self.loopback0.local_ip4n)
 278         echo_ip4 = pack("!L", unpacked[0] ^ 1)
 279         echo_source = self.vapi.bfd_udp_get_echo_source()
 280         self.assertTrue(echo_source.is_set)
 281         self.assertEqual(echo_source.sw_if_index, self.loopback0.sw_if_index)
 282         self.assertTrue(echo_source.have_usable_ip4)
 283         self.assertEqual(echo_source.ip4_addr, echo_ip4)
 284         self.assertFalse(echo_source.have_usable_ip6)
 285
 286         self.loopback0.config_ip6()
 287         unpacked = unpack("!LLLL", self.loopback0.local_ip6n)
 288         echo_ip6 = pack("!LLLL", unpacked[0], unpacked[1], unpacked[2],
 289                         unpacked[3] ^ 1)
 290         echo_source = self.vapi.bfd_udp_get_echo_source()
 291         self.assertTrue(echo_source.is_set)
 292         self.assertEqual(echo_source.sw_if_index, self.loopback0.sw_if_index)
 293         self.assertTrue(echo_source.have_usable_ip4)
 294         self.assertEqual(echo_source.ip4_addr, echo_ip4)
 295         self.assertTrue(echo_source.have_usable_ip6)
 296         self.assertEqual(echo_source.ip6_addr, echo_ip6)
 297
 298         self.vapi.bfd_udp_del_echo_source()
 299         echo_source = self.vapi.bfd_udp_get_echo_source()
 300         self.assertFalse(echo_source.is_set)
 301         self.assertFalse(echo_source.have_usable_ip4)
 302         self.assertFalse(echo_source.have_usable_ip6)
 303
 304
 305 @unittest.skipUnless(running_extended_tests, "part of extended tests")
 306 class BFDTestSession(object):
 307     """ BFD session as seen from test framework side """
 308
 309     def __init__(self, test, interface, af, detect_mult=3, sha1_key=None,
 310                  bfd_key_id=None, our_seq_number=None):
 311         self.test = test
 312         self.af = af
 313         self.sha1_key = sha1_key
 314         self.bfd_key_id = bfd_key_id
 315         self.interface = interface
 316         self.udp_sport = randint(49152, 65535)
 317         if our_seq_number is None:
 318             self.our_seq_number = randint(0, 40000000)
 319         else:
 320             self.our_seq_number = our_seq_number
 321         self.vpp_seq_number = None
 322         self.my_discriminator = 0
 323         self.desired_min_tx = 300000
 324         self.required_min_rx = 300000
 325         self.required_min_echo_rx = None
 326         self.detect_mult = detect_mult
 327         self.diag = BFDDiagCode.no_diagnostic
 328         self.your_discriminator = None
 329         self.state = BFDState.down
 330         self.auth_type = BFDAuthType.no_auth
 331
 332     def inc_seq_num(self):
 333         """ increment sequence number, wrapping if needed """
 334         if self.our_seq_number == 0xFFFFFFFF:
 335             self.our_seq_number = 0
 336         else:
 337             self.our_seq_number += 1
 338
 339     def update(self, my_discriminator=None, your_discriminator=None,
 340                desired_min_tx=None, required_min_rx=None,
 341                required_min_echo_rx=None, detect_mult=None,
 342                diag=None, state=None, auth_type=None):
 343         """ update BFD parameters associated with session """
 344         if my_discriminator is not None:
 345             self.my_discriminator = my_discriminator
 346         if your_discriminator is not None:
 347             self.your_discriminator = your_discriminator
 348         if required_min_rx is not None:
 349             self.required_min_rx = required_min_rx
 350         if required_min_echo_rx is not None:
 351             self.required_min_echo_rx = required_min_echo_rx
 352         if desired_min_tx is not None:
 353             self.desired_min_tx = desired_min_tx
 354         if detect_mult is not None:
 355             self.detect_mult = detect_mult
 356         if diag is not None:
 357             self.diag = diag
 358         if state is not None:
 359             self.state = state
 360         if auth_type is not None:
 361             self.auth_type = auth_type
 362
 363     def fill_packet_fields(self, packet):
 364         """ set packet fields with known values in packet """
 365         bfd = packet[BFD]
 366         if self.my_discriminator:
 367             self.test.logger.debug("BFD: setting packet.my_discriminator=%s",
 368                                    self.my_discriminator)
 369             bfd.my_discriminator = self.my_discriminator
 370         if self.your_discriminator:
 371             self.test.logger.debug("BFD: setting packet.your_discriminator=%s",
 372                                    self.your_discriminator)
 373             bfd.your_discriminator = self.your_discriminator
 374         if self.required_min_rx:
 375             self.test.logger.debug(
 376                 "BFD: setting packet.required_min_rx_interval=%s",
 377                 self.required_min_rx)
 378             bfd.required_min_rx_interval = self.required_min_rx
 379         if self.required_min_echo_rx:
 380             self.test.logger.debug(
 381                 "BFD: setting packet.required_min_echo_rx=%s",
 382                 self.required_min_echo_rx)
 383             bfd.required_min_echo_rx_interval = self.required_min_echo_rx
 384         if self.desired_min_tx:
 385             self.test.logger.debug(
 386                 "BFD: setting packet.desired_min_tx_interval=%s",
 387                 self.desired_min_tx)
 388             bfd.desired_min_tx_interval = self.desired_min_tx
 389         if self.detect_mult:
 390             self.test.logger.debug(
 391                 "BFD: setting packet.detect_mult=%s", self.detect_mult)
 392             bfd.detect_mult = self.detect_mult
 393         if self.diag:
 394             self.test.logger.debug("BFD: setting packet.diag=%s", self.diag)
 395             bfd.diag = self.diag
 396         if self.state:
 397             self.test.logger.debug("BFD: setting packet.state=%s", self.state)
 398             bfd.state = self.state
 399         if self.auth_type:
 400             # this is used by a negative test-case
 401             self.test.logger.debug("BFD: setting packet.auth_type=%s",
 402                                    self.auth_type)
 403             bfd.auth_type = self.auth_type
 404
 405     def create_packet(self):
 406         """ create a BFD packet, reflecting the current state of session """
 407         if self.sha1_key:
 408             bfd = BFD(flags="A")
 409             bfd.auth_type = self.sha1_key.auth_type
 410             bfd.auth_len = BFD.sha1_auth_len
 411             bfd.auth_key_id = self.bfd_key_id
 412             bfd.auth_seq_num = self.our_seq_number
 413             bfd.length = BFD.sha1_auth_len + BFD.bfd_pkt_len
 414         else:
 415             bfd = BFD()
 416         if self.af == AF_INET6:
 417             packet = (Ether(src=self.interface.remote_mac,
 418                             dst=self.interface.local_mac) /
 419                       IPv6(src=self.interface.remote_ip6,
 420                            dst=self.interface.local_ip6,
 421                            hlim=255) /
 422                       UDP(sport=self.udp_sport, dport=BFD.udp_dport) /
 423                       bfd)
 424         else:
 425             packet = (Ether(src=self.interface.remote_mac,
 426                             dst=self.interface.local_mac) /
 427                       IP(src=self.interface.remote_ip4,
 428                          dst=self.interface.local_ip4,
 429                          ttl=255) /
 430                       UDP(sport=self.udp_sport, dport=BFD.udp_dport) /
 431                       bfd)
 432         self.test.logger.debug("BFD: Creating packet")
 433         self.fill_packet_fields(packet)
 434         if self.sha1_key:
 435             hash_material = str(packet[BFD])[:32] + self.sha1_key.key + \
 436                 "\0" * (20 - len(self.sha1_key.key))
 437             self.test.logger.debug("BFD: Calculated SHA1 hash: %s" %
 438                                    hashlib.sha1(hash_material).hexdigest())
 439             packet[BFD].auth_key_hash = hashlib.sha1(hash_material).digest()
 440         return packet
 441
 442     def send_packet(self, packet=None, interface=None):
 443         """ send packet on interface, creating the packet if needed """
 444         if packet is None:
 445             packet = self.create_packet()
 446         if interface is None:
 447             interface = self.test.pg0
 448         self.test.logger.debug(ppp("Sending packet:", packet))
 449         interface.add_stream(packet)
 450         self.test.pg_start()
 451
 452     def verify_sha1_auth(self, packet):
 453         """ Verify correctness of authentication in BFD layer. """
 454         bfd = packet[BFD]
 455         self.test.assert_equal(bfd.auth_len, 28, "Auth section length")
 456         self.test.assert_equal(bfd.auth_type, self.sha1_key.auth_type,
 457                                BFDAuthType)
 458         self.test.assert_equal(bfd.auth_key_id, self.bfd_key_id, "Key ID")
 459         self.test.assert_equal(bfd.auth_reserved, 0, "Reserved")
 460         if self.vpp_seq_number is None:
 461             self.vpp_seq_number = bfd.auth_seq_num
 462             self.test.logger.debug("Received initial sequence number: %s" %
 463                                    self.vpp_seq_number)
 464         else:
 465             recvd_seq_num = bfd.auth_seq_num
 466             self.test.logger.debug("Received followup sequence number: %s" %
 467                                    recvd_seq_num)
 468             if self.vpp_seq_number < 0xffffffff:
 469                 if self.sha1_key.auth_type == \
 470                         BFDAuthType.meticulous_keyed_sha1:
 471                     self.test.assert_equal(recvd_seq_num,
 472                                            self.vpp_seq_number + 1,
 473                                            "BFD sequence number")
 474                 else:
 475                     self.test.assert_in_range(recvd_seq_num,
 476                                               self.vpp_seq_number,
 477                                               self.vpp_seq_number + 1,
 478                                               "BFD sequence number")
 479             else:
 480                 if self.sha1_key.auth_type == \
 481                         BFDAuthType.meticulous_keyed_sha1:
 482                     self.test.assert_equal(recvd_seq_num, 0,
 483                                            "BFD sequence number")
 484                 else:
 485                     self.test.assertIn(recvd_seq_num, (self.vpp_seq_number, 0),
 486                                        "BFD sequence number not one of "
 487                                        "(%s, 0)" % self.vpp_seq_number)
 488             self.vpp_seq_number = recvd_seq_num
 489         # last 20 bytes represent the hash - so replace them with the key,
 490         # pad the result with zeros and hash the result
 491         hash_material = bfd.original[:-20] + self.sha1_key.key + \
 492             "\0" * (20 - len(self.sha1_key.key))
 493         expected_hash = hashlib.sha1(hash_material).hexdigest()
 494         self.test.assert_equal(binascii.hexlify(bfd.auth_key_hash),
 495                                expected_hash, "Auth key hash")
 496
 497     def verify_bfd(self, packet):
 498         """ Verify correctness of BFD layer. """
 499         bfd = packet[BFD]
 500         self.test.assert_equal(bfd.version, 1, "BFD version")
 501         self.test.assert_equal(bfd.your_discriminator,
 502                                self.my_discriminator,
 503                                "BFD - your discriminator")
 504         if self.sha1_key:
 505             self.verify_sha1_auth(packet)
 506
 507
 508 def bfd_session_up(test):
 509     """ Bring BFD session up """
 510     test.logger.info("BFD: Waiting for slow hello")
 511     p = wait_for_bfd_packet(test, 2)
 512     old_offset = None
 513     if hasattr(test, 'vpp_clock_offset'):
 514         old_offset = test.vpp_clock_offset
 515     test.vpp_clock_offset = time.time() - p.time
 516     test.logger.debug("BFD: Calculated vpp clock offset: %s",
 517                       test.vpp_clock_offset)
 518     if old_offset:
 519         test.assertAlmostEqual(
 520             old_offset, test.vpp_clock_offset, delta=0.5,
 521             msg="vpp clock offset not stable (new: %s, old: %s)" %
 522             (test.vpp_clock_offset, old_offset))
 523     test.logger.info("BFD: Sending Init")
 524     test.test_session.update(my_discriminator=randint(0, 40000000),
 525                              your_discriminator=p[BFD].my_discriminator,
 526                              state=BFDState.init)
 527     if test.test_session.sha1_key and test.test_session.sha1_key.auth_type == \
 528             BFDAuthType.meticulous_keyed_sha1:
 529         test.test_session.inc_seq_num()
 530     test.test_session.send_packet()
 531     test.logger.info("BFD: Waiting for event")
 532     e = test.vapi.wait_for_event(1, "bfd_udp_session_details")
 533     verify_event(test, e, expected_state=BFDState.up)
 534     test.logger.info("BFD: Session is Up")
 535     test.test_session.update(state=BFDState.up)
 536     if test.test_session.sha1_key and test.test_session.sha1_key.auth_type == \
 537             BFDAuthType.meticulous_keyed_sha1:
 538         test.test_session.inc_seq_num()
 539     test.test_session.send_packet()
 540     test.assert_equal(test.vpp_session.state, BFDState.up, BFDState)
 541
 542
 543 def bfd_session_down(test):
 544     """ Bring BFD session down """
 545     test.assert_equal(test.vpp_session.state, BFDState.up, BFDState)
 546     test.test_session.update(state=BFDState.down)
 547     if test.test_session.sha1_key and test.test_session.sha1_key.auth_type == \
 548             BFDAuthType.meticulous_keyed_sha1:
 549         test.test_session.inc_seq_num()
 550     test.test_session.send_packet()
 551     test.logger.info("BFD: Waiting for event")
 552     e = test.vapi.wait_for_event(1, "bfd_udp_session_details")
 553     verify_event(test, e, expected_state=BFDState.down)
 554     test.logger.info("BFD: Session is Down")
 555     test.assert_equal(test.vpp_session.state, BFDState.down, BFDState)
 556
 557
 558 def verify_bfd_session_config(test, session, state=None):
 559     dump = session.get_bfd_udp_session_dump_entry()
 560     test.assertIsNotNone(dump)
 561     # since dump is not none, we have verified that sw_if_index and addresses
 562     # are valid (in get_bfd_udp_session_dump_entry)
 563     if state:
 564         test.assert_equal(dump.state, state, "session state")
 565     test.assert_equal(dump.required_min_rx, session.required_min_rx,
 566                       "required min rx interval")
 567     test.assert_equal(dump.desired_min_tx, session.desired_min_tx,
 568                       "desired min tx interval")
 569     test.assert_equal(dump.detect_mult, session.detect_mult,
 570                       "detect multiplier")
 571     if session.sha1_key is None:
 572         test.assert_equal(dump.is_authenticated, 0, "is_authenticated flag")
 573     else:
 574         test.assert_equal(dump.is_authenticated, 1, "is_authenticated flag")
 575         test.assert_equal(dump.bfd_key_id, session.bfd_key_id,
 576                           "bfd key id")
 577         test.assert_equal(dump.conf_key_id,
 578                           session.sha1_key.conf_key_id,
 579                           "config key id")
 580
 581
 582 def verify_ip(test, packet):
 583     """ Verify correctness of IP layer. """
 584     if test.vpp_session.af == AF_INET6:
 585         ip = packet[IPv6]
 586         local_ip = test.pg0.local_ip6
 587         remote_ip = test.pg0.remote_ip6
 588         test.assert_equal(ip.hlim, 255, "IPv6 hop limit")
 589     else:
 590         ip = packet[IP]
 591         local_ip = test.pg0.local_ip4
 592         remote_ip = test.pg0.remote_ip4
 593         test.assert_equal(ip.ttl, 255, "IPv4 TTL")
 594     test.assert_equal(ip.src, local_ip, "IP source address")
 595     test.assert_equal(ip.dst, remote_ip, "IP destination address")
 596
 597
 598 def verify_udp(test, packet):
 599     """ Verify correctness of UDP layer. """
 600     udp = packet[UDP]
 601     test.assert_equal(udp.dport, BFD.udp_dport, "UDP destination port")
 602     test.assert_in_range(udp.sport, BFD.udp_sport_min, BFD.udp_sport_max,
 603                          "UDP source port")
 604
 605
 606 def verify_event(test, event, expected_state):
 607     """ Verify correctness of event values. """
 608     e = event
 609     test.logger.debug("BFD: Event: %s" % moves.reprlib.repr(e))
 610     test.assert_equal(e.sw_if_index,
 611                       test.vpp_session.interface.sw_if_index,
 612                       "BFD interface index")
 613     is_ipv6 = 0
 614     if test.vpp_session.af == AF_INET6:
 615         is_ipv6 = 1
 616     test.assert_equal(e.is_ipv6, is_ipv6, "is_ipv6")
 617     if test.vpp_session.af == AF_INET:
 618         test.assert_equal(e.local_addr[:4], test.vpp_session.local_addr_n,
 619                           "Local IPv4 address")
 620         test.assert_equal(e.peer_addr[:4], test.vpp_session.peer_addr_n,
 621                           "Peer IPv4 address")
 622     else:
 623         test.assert_equal(e.local_addr, test.vpp_session.local_addr_n,
 624                           "Local IPv6 address")
 625         test.assert_equal(e.peer_addr, test.vpp_session.peer_addr_n,
 626                           "Peer IPv6 address")
 627     test.assert_equal(e.state, expected_state, BFDState)
 628
 629
 630 def wait_for_bfd_packet(test, timeout=1, pcap_time_min=None):
 631     """ wait for BFD packet and verify its correctness
 632
 633     :param timeout: how long to wait
 634     :param pcap_time_min: ignore packets with pcap timestamp lower than this
 635
 636     :returns: tuple (packet, time spent waiting for packet)
 637     """
 638     test.logger.info("BFD: Waiting for BFD packet")
 639     deadline = time.time() + timeout
 640     counter = 0
 641     while True:
 642         counter += 1
 643         # sanity check
 644         test.assert_in_range(counter, 0, 100, "number of packets ignored")
 645         time_left = deadline - time.time()
 646         if time_left < 0:
 647             raise CaptureTimeoutError("Packet did not arrive within timeout")
 648         p = test.pg0.wait_for_packet(timeout=time_left)
 649         test.logger.debug(ppp("BFD: Got packet:", p))
 650         if pcap_time_min is not None and p.time < pcap_time_min:
 651             test.logger.debug(ppp("BFD: ignoring packet (pcap time %s < "
 652                                   "pcap time min %s):" %
 653                                   (p.time, pcap_time_min), p))
 654         else:
 655             break
 656     bfd = p[BFD]
 657     if bfd is None:
 658         raise Exception(ppp("Unexpected or invalid BFD packet:", p))
 659     if bfd.payload:
 660         raise Exception(ppp("Unexpected payload in BFD packet:", bfd))
 661     verify_ip(test, p)
 662     verify_udp(test, p)
 663     test.test_session.verify_bfd(p)
 664     return p
 665
 666
 667 @unittest.skipUnless(running_extended_tests, "part of extended tests")
 668 class BFD4TestCase(VppTestCase):
 669     """Bidirectional Forwarding Detection (BFD)"""
 670
 671     pg0 = None
 672     vpp_clock_offset = None
 673     vpp_session = None
 674     test_session = None
 675
 676     @classmethod
 677     def setUpClass(cls):
 678         super(BFD4TestCase, cls).setUpClass()
 679         cls.vapi.cli("set log class bfd level debug")
 680         try:
 681             cls.create_pg_interfaces([0])
 682             cls.create_loopback_interfaces(1)
 683             cls.loopback0 = cls.lo_interfaces[0]
 684             cls.loopback0.config_ip4()
 685             cls.loopback0.admin_up()
 686             cls.pg0.config_ip4()
 687             cls.pg0.configure_ipv4_neighbors()
 688             cls.pg0.admin_up()
 689             cls.pg0.resolve_arp()
 690
 691         except Exception:
 692             super(BFD4TestCase, cls).tearDownClass()
 693             raise
 694
 695     def setUp(self):
 696         super(BFD4TestCase, self).setUp()
 697         self.factory = AuthKeyFactory()
 698         self.vapi.want_bfd_events()
 699         self.pg0.enable_capture()
 700         try:
 701             self.vpp_session = VppBFDUDPSession(self, self.pg0,
 702                                                 self.pg0.remote_ip4)
 703             self.vpp_session.add_vpp_config()
 704             self.vpp_session.admin_up()
 705             self.test_session = BFDTestSession(self, self.pg0, AF_INET)
 706         except:
 707             self.vapi.want_bfd_events(enable_disable=0)
 708             raise
 709
 710     def tearDown(self):
 711         if not self.vpp_dead:
 712             self.vapi.want_bfd_events(enable_disable=0)
 713         self.vapi.collect_events()  # clear the event queue
 714         super(BFD4TestCase, self).tearDown()
 715
 716     def test_session_up(self):
 717         """ bring BFD session up """
 718         bfd_session_up(self)
 719
 720     def test_session_up_by_ip(self):
 721         """ bring BFD session up - first frame looked up by address pair """
 722         self.logger.info("BFD: Sending Slow control frame")
 723         self.test_session.update(my_discriminator=randint(0, 40000000))
 724         self.test_session.send_packet()
 725         self.pg0.enable_capture()
 726         p = self.pg0.wait_for_packet(1)
 727         self.assert_equal(p[BFD].your_discriminator,
 728                           self.test_session.my_discriminator,
 729                           "BFD - your discriminator")
 730         self.assert_equal(p[BFD].state, BFDState.init, BFDState)
 731         self.test_session.update(your_discriminator=p[BFD].my_discriminator,
 732                                  state=BFDState.up)
 733         self.logger.info("BFD: Waiting for event")
 734         e = self.vapi.wait_for_event(1, "bfd_udp_session_details")
 735         verify_event(self, e, expected_state=BFDState.init)
 736         self.logger.info("BFD: Sending Up")
 737         self.test_session.send_packet()
 738         self.logger.info("BFD: Waiting for event")
 739         e = self.vapi.wait_for_event(1, "bfd_udp_session_details")
 740         verify_event(self, e, expected_state=BFDState.up)
 741         self.logger.info("BFD: Session is Up")
 742         self.test_session.update(state=BFDState.up)
 743         self.test_session.send_packet()
 744         self.assert_equal(self.vpp_session.state, BFDState.up, BFDState)
 745
 746     def test_session_down(self):
 747         """ bring BFD session down """
 748         bfd_session_up(self)
 749         bfd_session_down(self)
 750
 751     @unittest.skipUnless(running_extended_tests, "part of extended tests")
 752     def test_hold_up(self):
 753         """ hold BFD session up """
 754         bfd_session_up(self)
 755         for dummy in range(self.test_session.detect_mult * 2):
 756             wait_for_bfd_packet(self)
 757             self.test_session.send_packet()
 758         self.assert_equal(len(self.vapi.collect_events()), 0,
 759                           "number of bfd events")
 760
 761     @unittest.skipUnless(running_extended_tests, "part of extended tests")
 762     def test_slow_timer(self):
 763         """ verify slow periodic control frames while session down """
 764         packet_count = 3
 765         self.logger.info("BFD: Waiting for %d BFD packets", packet_count)
 766         prev_packet = wait_for_bfd_packet(self, 2)
 767         for dummy in range(packet_count):
 768             next_packet = wait_for_bfd_packet(self, 2)
 769             time_diff = next_packet.time - prev_packet.time
 770             # spec says the range should be <0.75, 1>, allow extra 0.05 margin
 771             # to work around timing issues
 772             self.assert_in_range(
 773                 time_diff, 0.70, 1.05, "time between slow packets")
 774             prev_packet = next_packet
 775
 776     @unittest.skipUnless(running_extended_tests, "part of extended tests")
 777     def test_zero_remote_min_rx(self):
 778         """ no packets when zero remote required min rx interval """
 779         bfd_session_up(self)
 780         self.test_session.update(required_min_rx=0)
 781         self.test_session.send_packet()
 782         for dummy in range(self.test_session.detect_mult):
 783             self.sleep(self.vpp_session.required_min_rx / USEC_IN_SEC,
 784                        "sleep before transmitting bfd packet")
 785             self.test_session.send_packet()
 786             try:
 787                 p = wait_for_bfd_packet(self, timeout=0)
 788                 self.logger.error(ppp("Received unexpected packet:", p))
 789             except CaptureTimeoutError:
 790                 pass
 791         self.assert_equal(
 792             len(self.vapi.collect_events()), 0, "number of bfd events")
 793         self.test_session.update(required_min_rx=300000)
 794         for dummy in range(3):
 795             self.test_session.send_packet()
 796             wait_for_bfd_packet(
 797                 self, timeout=self.test_session.required_min_rx / USEC_IN_SEC)
 798         self.assert_equal(
 799             len(self.vapi.collect_events()), 0, "number of bfd events")
 800
 801     @unittest.skipUnless(running_extended_tests, "part of extended tests")
 802     def test_conn_down(self):
 803         """ verify session goes down after inactivity """
 804         bfd_session_up(self)
 805         detection_time = self.test_session.detect_mult *\
 806             self.vpp_session.required_min_rx / USEC_IN_SEC
 807         self.sleep(detection_time, "waiting for BFD session time-out")
 808         e = self.vapi.wait_for_event(1, "bfd_udp_session_details")
 809         verify_event(self, e, expected_state=BFDState.down)
 810
 811     @unittest.skipUnless(running_extended_tests, "part of extended tests")
 812     def test_large_required_min_rx(self):
 813         """ large remote required min rx interval """
 814         bfd_session_up(self)
 815         p = wait_for_bfd_packet(self)
 816         interval = 3000000
 817         self.test_session.update(required_min_rx=interval)
 818         self.test_session.send_packet()
 819         time_mark = time.time()
 820         count = 0
 821         # busy wait here, trying to collect a packet or event, vpp is not
 822         # allowed to send packets and the session will timeout first - so the
 823         # Up->Down event must arrive before any packets do
 824         while time.time() < time_mark + interval / USEC_IN_SEC:
 825             try:
 826                 p = wait_for_bfd_packet(self, timeout=0)
 827                 # if vpp managed to send a packet before we did the session
 828                 # session update, then that's fine, ignore it
 829                 if p.time < time_mark - self.vpp_clock_offset:
 830                     continue
 831                 self.logger.error(ppp("Received unexpected packet:", p))
 832                 count += 1
 833             except CaptureTimeoutError:
 834                 pass
 835             events = self.vapi.collect_events()
 836             if len(events) > 0:
 837                 verify_event(self, events[0], BFDState.down)
 838                 break
 839         self.assert_equal(count, 0, "number of packets received")
 840
 841     @unittest.skipUnless(running_extended_tests, "part of extended tests")
 842     def test_immediate_remote_min_rx_reduction(self):
 843         """ immediately honor remote required min rx reduction """
 844         self.vpp_session.remove_vpp_config()
 845         self.vpp_session = VppBFDUDPSession(
 846             self, self.pg0, self.pg0.remote_ip4, desired_min_tx=10000)
 847         self.pg0.enable_capture()
 848         self.vpp_session.add_vpp_config()
 849         self.test_session.update(desired_min_tx=1000000,
 850                                  required_min_rx=1000000)
 851         bfd_session_up(self)
 852         reference_packet = wait_for_bfd_packet(self)
 853         time_mark = time.time()
 854         interval = 300000
 855         self.test_session.update(required_min_rx=interval)
 856         self.test_session.send_packet()
 857         extra_time = time.time() - time_mark
 858         p = wait_for_bfd_packet(self)
 859         # first packet is allowed to be late by time we spent doing the update
 860         # calculated in extra_time
 861         self.assert_in_range(p.time - reference_packet.time,
 862                              .95 * 0.75 * interval / USEC_IN_SEC,
 863                              1.05 * interval / USEC_IN_SEC + extra_time,
 864                              "time between BFD packets")
 865         reference_packet = p
 866         for dummy in range(3):
 867             p = wait_for_bfd_packet(self)
 868             diff = p.time - reference_packet.time
 869             self.assert_in_range(diff, .95 * .75 * interval / USEC_IN_SEC,
 870                                  1.05 * interval / USEC_IN_SEC,
 871                                  "time between BFD packets")
 872             reference_packet = p
 873
 874     @unittest.skipUnless(running_extended_tests, "part of extended tests")
 875     def test_modify_req_min_rx_double(self):
 876         """ modify session - double required min rx """
 877         bfd_session_up(self)
 878         p = wait_for_bfd_packet(self)
 879         self.test_session.update(desired_min_tx=10000,
 880                                  required_min_rx=10000)
 881         self.test_session.send_packet()
 882         # double required min rx
 883         self.vpp_session.modify_parameters(
 884             required_min_rx=2 * self.vpp_session.required_min_rx)
 885         p = wait_for_bfd_packet(
 886             self, pcap_time_min=time.time() - self.vpp_clock_offset)
 887         # poll bit needs to be set
 888         self.assertIn("P", p.sprintf("%BFD.flags%"),
 889                       "Poll bit not set in BFD packet")
 890         # finish poll sequence with final packet
 891         final = self.test_session.create_packet()
 892         final[BFD].flags = "F"
 893         timeout = self.test_session.detect_mult * \
 894             max(self.test_session.desired_min_tx,
 895                 self.vpp_session.required_min_rx) / USEC_IN_SEC
 896         self.test_session.send_packet(final)
 897         time_mark = time.time()
 898         e = self.vapi.wait_for_event(2 * timeout, "bfd_udp_session_details")
 899         verify_event(self, e, expected_state=BFDState.down)
 900         time_to_event = time.time() - time_mark
 901         self.assert_in_range(time_to_event, .9 * timeout,
 902                              1.1 * timeout, "session timeout")
 903
 904     @unittest.skipUnless(running_extended_tests, "part of extended tests")
 905     def test_modify_req_min_rx_halve(self):
 906         """ modify session - halve required min rx """
 907         self.vpp_session.modify_parameters(
 908             required_min_rx=2 * self.vpp_session.required_min_rx)
 909         bfd_session_up(self)
 910         p = wait_for_bfd_packet(self)
 911         self.test_session.update(desired_min_tx=10000,
 912                                  required_min_rx=10000)
 913         self.test_session.send_packet()
 914         p = wait_for_bfd_packet(
 915             self, pcap_time_min=time.time() - self.vpp_clock_offset)
 916         # halve required min rx
 917         old_required_min_rx = self.vpp_session.required_min_rx
 918         self.vpp_session.modify_parameters(
 919             required_min_rx=0.5 * self.vpp_session.required_min_rx)
 920         # now we wait 0.8*3*old-req-min-rx and the session should still be up
 921         self.sleep(0.8 * self.vpp_session.detect_mult *
 922                    old_required_min_rx / USEC_IN_SEC,
 923                    "wait before finishing poll sequence")
 924         self.assert_equal(len(self.vapi.collect_events()), 0,
 925                           "number of bfd events")
 926         p = wait_for_bfd_packet(self)
 927         # poll bit needs to be set
 928         self.assertIn("P", p.sprintf("%BFD.flags%"),
 929                       "Poll bit not set in BFD packet")
 930         # finish poll sequence with final packet
 931         final = self.test_session.create_packet()
 932         final[BFD].flags = "F"
 933         self.test_session.send_packet(final)
 934         # now the session should time out under new conditions
 935         detection_time = self.test_session.detect_mult *\
 936             self.vpp_session.required_min_rx / USEC_IN_SEC
 937         before = time.time()
 938         e = self.vapi.wait_for_event(
 939             2 * detection_time, "bfd_udp_session_details")
 940         after = time.time()
 941         self.assert_in_range(after - before,
 942                              0.9 * detection_time,
 943                              1.1 * detection_time,
 944                              "time before bfd session goes down")
 945         verify_event(self, e, expected_state=BFDState.down)
 946
 947     @unittest.skipUnless(running_extended_tests, "part of extended tests")
 948     def test_modify_detect_mult(self):
 949         """ modify detect multiplier """
 950         bfd_session_up(self)
 951         p = wait_for_bfd_packet(self)
 952         self.vpp_session.modify_parameters(detect_mult=1)
 953         p = wait_for_bfd_packet(
 954             self, pcap_time_min=time.time() - self.vpp_clock_offset)
 955         self.assert_equal(self.vpp_session.detect_mult,
 956                           p[BFD].detect_mult,
 957                           "detect mult")
 958         # poll bit must not be set
 959         self.assertNotIn("P", p.sprintf("%BFD.flags%"),
 960                          "Poll bit not set in BFD packet")
 961         self.vpp_session.modify_parameters(detect_mult=10)
 962         p = wait_for_bfd_packet(
 963             self, pcap_time_min=time.time() - self.vpp_clock_offset)
 964         self.assert_equal(self.vpp_session.detect_mult,
 965                           p[BFD].detect_mult,
 966                           "detect mult")
 967         # poll bit must not be set
 968         self.assertNotIn("P", p.sprintf("%BFD.flags%"),
 969                          "Poll bit not set in BFD packet")
 970
 971     @unittest.skipUnless(running_extended_tests, "part of extended tests")
 972     def test_queued_poll(self):
 973         """ test poll sequence queueing """
 974         bfd_session_up(self)
 975         p = wait_for_bfd_packet(self)
 976         self.vpp_session.modify_parameters(
 977             required_min_rx=2 * self.vpp_session.required_min_rx)
 978         p = wait_for_bfd_packet(self)
 979         poll_sequence_start = time.time()
 980         poll_sequence_length_min = 0.5
 981         send_final_after = time.time() + poll_sequence_length_min
 982         # poll bit needs to be set
 983         self.assertIn("P", p.sprintf("%BFD.flags%"),
 984                       "Poll bit not set in BFD packet")
 985         self.assert_equal(p[BFD].required_min_rx_interval,
 986                           self.vpp_session.required_min_rx,
 987                           "BFD required min rx interval")
 988         self.vpp_session.modify_parameters(
 989             required_min_rx=2 * self.vpp_session.required_min_rx)
 990         # 2nd poll sequence should be queued now
 991         # don't send the reply back yet, wait for some time to emulate
 992         # longer round-trip time
 993         packet_count = 0
 994         while time.time() < send_final_after:
 995             self.test_session.send_packet()
 996             p = wait_for_bfd_packet(self)
 997             self.assert_equal(len(self.vapi.collect_events()), 0,
 998                               "number of bfd events")
 999             self.assert_equal(p[BFD].required_min_rx_interval,
1000                               self.vpp_session.required_min_rx,
1001                               "BFD required min rx interval")
1002             packet_count += 1
1003             # poll bit must be set
1004             self.assertIn("P", p.sprintf("%BFD.flags%"),
1005                           "Poll bit not set in BFD packet")
1006         final = self.test_session.create_packet()
1007         final[BFD].flags = "F"
1008         self.test_session.send_packet(final)
1009         # finish 1st with final
1010         poll_sequence_length = time.time() - poll_sequence_start
1011         # vpp must wait for some time before starting new poll sequence
1012         poll_no_2_started = False
1013         for dummy in range(2 * packet_count):
1014             p = wait_for_bfd_packet(self)
1015             self.assert_equal(len(self.vapi.collect_events()), 0,
1016                               "number of bfd events")
1017             if "P" in p.sprintf("%BFD.flags%"):
1018                 poll_no_2_started = True
1019                 if time.time() < poll_sequence_start + poll_sequence_length:
1020                     raise Exception("VPP started 2nd poll sequence too soon")
1021                 final = self.test_session.create_packet()
1022                 final[BFD].flags = "F"
1023                 self.test_session.send_packet(final)
1024                 break
1025             else:
1026                 self.test_session.send_packet()
1027         self.assertTrue(poll_no_2_started, "2nd poll sequence not performed")
1028         # finish 2nd with final
1029         final = self.test_session.create_packet()
1030         final[BFD].flags = "F"
1031         self.test_session.send_packet(final)
1032         p = wait_for_bfd_packet(self)
1033         # poll bit must not be set
1034         self.assertNotIn("P", p.sprintf("%BFD.flags%"),
1035                          "Poll bit set in BFD packet")
1036
1037     @unittest.skipUnless(running_extended_tests, "part of extended tests")
1038     def test_poll_response(self):
1039         """ test correct response to control frame with poll bit set """
1040         bfd_session_up(self)
1041         poll = self.test_session.create_packet()
1042         poll[BFD].flags = "P"
1043         self.test_session.send_packet(poll)
1044         final = wait_for_bfd_packet(
1045             self, pcap_time_min=time.time() - self.vpp_clock_offset)
1046         self.assertIn("F", final.sprintf("%BFD.flags%"))
1047
1048     @unittest.skipUnless(running_extended_tests, "part of extended tests")
1049     def test_no_periodic_if_remote_demand(self):
1050         """ no periodic frames outside poll sequence if remote demand set """
1051         bfd_session_up(self)
1052         demand = self.test_session.create_packet()
1053         demand[BFD].flags = "D"
1054         self.test_session.send_packet(demand)
1055         transmit_time = 0.9 \
1056             * max(self.vpp_session.required_min_rx,
1057                   self.test_session.desired_min_tx) \
1058             / USEC_IN_SEC
1059         count = 0
1060         for dummy in range(self.test_session.detect_mult * 2):
1061             time.sleep(transmit_time)
1062             self.test_session.send_packet(demand)
1063             try:
1064                 p = wait_for_bfd_packet(self, timeout=0)
1065                 self.logger.error(ppp("Received unexpected packet:", p))
1066                 count += 1
1067             except CaptureTimeoutError:
1068                 pass
1069         events = self.vapi.collect_events()
1070         for e in events:
1071             self.logger.error("Received unexpected event: %s", e)
1072         self.assert_equal(count, 0, "number of packets received")
1073         self.assert_equal(len(events), 0, "number of events received")
1074
1075     def test_echo_looped_back(self):
1076         """ echo packets looped back """
1077         # don't need a session in this case..
1078         self.vpp_session.remove_vpp_config()
1079         self.pg0.enable_capture()
1080         echo_packet_count = 10
1081         # random source port low enough to increment a few times..
1082         udp_sport_tx = randint(1, 50000)
1083         udp_sport_rx = udp_sport_tx
1084         echo_packet = (Ether(src=self.pg0.remote_mac,
1085                              dst=self.pg0.local_mac) /
1086                        IP(src=self.pg0.remote_ip4,
1087                           dst=self.pg0.remote_ip4) /
1088                        UDP(dport=BFD.udp_dport_echo) /
1089                        Raw("this should be looped back"))
1090         for dummy in range(echo_packet_count):
1091             self.sleep(.01, "delay between echo packets")
1092             echo_packet[UDP].sport = udp_sport_tx
1093             udp_sport_tx += 1
1094             self.logger.debug(ppp("Sending packet:", echo_packet))
1095             self.pg0.add_stream(echo_packet)
1096             self.pg_start()
1097         for dummy in range(echo_packet_count):
1098             p = self.pg0.wait_for_packet(1)
1099             self.logger.debug(ppp("Got packet:", p))
1100             ether = p[Ether]
1101             self.assert_equal(self.pg0.remote_mac,
1102                               ether.dst, "Destination MAC")
1103             self.assert_equal(self.pg0.local_mac, ether.src, "Source MAC")
1104             ip = p[IP]
1105             self.assert_equal(self.pg0.remote_ip4, ip.dst, "Destination IP")
1106             self.assert_equal(self.pg0.remote_ip4, ip.src, "Destination IP")
1107             udp = p[UDP]
1108             self.assert_equal(udp.dport, BFD.udp_dport_echo,
1109                               "UDP destination port")
1110             self.assert_equal(udp.sport, udp_sport_rx, "UDP source port")
1111             udp_sport_rx += 1
1112             # need to compare the hex payload here, otherwise BFD_vpp_echo
1113             # gets in way
1114             self.assertEqual(str(p[UDP].payload),
1115                              str(echo_packet[UDP].payload),
1116                              "Received packet is not the echo packet sent")
1117         self.assert_equal(udp_sport_tx, udp_sport_rx, "UDP source port (== "
1118                           "ECHO packet identifier for test purposes)")
1119
1120     def test_echo(self):
1121         """ echo function """
1122         bfd_session_up(self)
1123         self.test_session.update(required_min_echo_rx=150000)
1124         self.test_session.send_packet()
1125         detection_time = self.test_session.detect_mult *\
1126             self.vpp_session.required_min_rx / USEC_IN_SEC
1127         # echo shouldn't work without echo source set
1128         for dummy in range(10):
1129             sleep = self.vpp_session.required_min_rx / USEC_IN_SEC
1130             self.sleep(sleep, "delay before sending bfd packet")
1131             self.test_session.send_packet()
1132         p = wait_for_bfd_packet(
1133             self, pcap_time_min=time.time() - self.vpp_clock_offset)
1134         self.assert_equal(p[BFD].required_min_rx_interval,
1135                           self.vpp_session.required_min_rx,
1136                           "BFD required min rx interval")
1137         self.test_session.send_packet()
1138         self.vapi.bfd_udp_set_echo_source(self.loopback0.sw_if_index)
1139         echo_seen = False
1140         # should be turned on - loopback echo packets
1141         for dummy in range(3):
1142             loop_until = time.time() + 0.75 * detection_time
1143             while time.time() < loop_until:
1144                 p = self.pg0.wait_for_packet(1)
1145                 self.logger.debug(ppp("Got packet:", p))
1146                 if p[UDP].dport == BFD.udp_dport_echo:
1147                     self.assert_equal(
1148                         p[IP].dst, self.pg0.local_ip4, "BFD ECHO dst IP")
1149                     self.assertNotEqual(p[IP].src, self.loopback0.local_ip4,
1150                                         "BFD ECHO src IP equal to loopback IP")
1151                     self.logger.debug(ppp("Looping back packet:", p))
1152                     self.assert_equal(p[Ether].dst, self.pg0.remote_mac,
1153                                       "ECHO packet destination MAC address")
1154                     p[Ether].dst = self.pg0.local_mac
1155                     self.pg0.add_stream(p)
1156                     self.pg_start()
1157                     echo_seen = True
1158                 elif p.haslayer(BFD):
1159                     if echo_seen:
1160                         self.assertGreaterEqual(
1161                             p[BFD].required_min_rx_interval,
1162                             1000000)
1163                     if "P" in p.sprintf("%BFD.flags%"):
1164                         final = self.test_session.create_packet()
1165                         final[BFD].flags = "F"
1166                         self.test_session.send_packet(final)
1167                 else:
1168                     raise Exception(ppp("Received unknown packet:", p))
1169
1170                 self.assert_equal(len(self.vapi.collect_events()), 0,
1171                                   "number of bfd events")
1172             self.test_session.send_packet()
1173         self.assertTrue(echo_seen, "No echo packets received")
1174
1175     @unittest.skipUnless(running_extended_tests, "part of extended tests")
1176     def test_echo_fail(self):
1177         """ session goes down if echo function fails """
1178         bfd_session_up(self)
1179         self.test_session.update(required_min_echo_rx=150000)
1180         self.test_session.send_packet()
1181         detection_time = self.test_session.detect_mult *\
1182             self.vpp_session.required_min_rx / USEC_IN_SEC
1183         self.vapi.bfd_udp_set_echo_source(self.loopback0.sw_if_index)
1184         # echo function should be used now, but we will drop the echo packets
1185         verified_diag = False
1186         for dummy in range(3):
1187             loop_until = time.time() + 0.75 * detection_time
1188             while time.time() < loop_until:
1189                 p = self.pg0.wait_for_packet(1)
1190                 self.logger.debug(ppp("Got packet:", p))
1191                 if p[UDP].dport == BFD.udp_dport_echo:
1192                     # dropped
1193                     pass
1194                 elif p.haslayer(BFD):
1195                     if "P" in p.sprintf("%BFD.flags%"):
1196                         self.assertGreaterEqual(
1197                             p[BFD].required_min_rx_interval,
1198                             1000000)
1199                         final = self.test_session.create_packet()
1200                         final[BFD].flags = "F"
1201                         self.test_session.send_packet(final)
1202                     if p[BFD].state == BFDState.down:
1203                         self.assert_equal(p[BFD].diag,
1204                                           BFDDiagCode.echo_function_failed,
1205                                           BFDDiagCode)
1206                         verified_diag = True
1207                 else:
1208                     raise Exception(ppp("Received unknown packet:", p))
1209             self.test_session.send_packet()
1210         events = self.vapi.collect_events()
1211         self.assert_equal(len(events), 1, "number of bfd events")
1212         self.assert_equal(events[0].state, BFDState.down, BFDState)
1213         self.assertTrue(verified_diag, "Incorrect diagnostics code received")
1214
1215     @unittest.skipUnless(running_extended_tests, "part of extended tests")
1216     def test_echo_stop(self):
1217         """ echo function stops if peer sets required min echo rx zero """
1218         bfd_session_up(self)
1219         self.test_session.update(required_min_echo_rx=150000)
1220         self.test_session.send_packet()
1221         self.vapi.bfd_udp_set_echo_source(self.loopback0.sw_if_index)
1222         # wait for first echo packet
1223         while True:
1224             p = self.pg0.wait_for_packet(1)
1225             self.logger.debug(ppp("Got packet:", p))
1226             if p[UDP].dport == BFD.udp_dport_echo:
1227                 self.logger.debug(ppp("Looping back packet:", p))
1228                 p[Ether].dst = self.pg0.local_mac
1229                 self.pg0.add_stream(p)
1230                 self.pg_start()
1231                 break
1232             elif p.haslayer(BFD):
1233                 # ignore BFD
1234                 pass
1235             else:
1236                 raise Exception(ppp("Received unknown packet:", p))
1237         self.test_session.update(required_min_echo_rx=0)
1238         self.test_session.send_packet()
1239         # echo packets shouldn't arrive anymore
1240         for dummy in range(5):
1241             wait_for_bfd_packet(
1242                 self, pcap_time_min=time.time() - self.vpp_clock_offset)
1243             self.test_session.send_packet()
1244             events = self.vapi.collect_events()
1245             self.assert_equal(len(events), 0, "number of bfd events")
1246
1247     @unittest.skipUnless(running_extended_tests, "part of extended tests")
1248     def test_echo_source_removed(self):
1249         """ echo function stops if echo source is removed """
1250         bfd_session_up(self)
1251         self.test_session.update(required_min_echo_rx=150000)
1252         self.test_session.send_packet()
1253         self.vapi.bfd_udp_set_echo_source(self.loopback0.sw_if_index)
1254         # wait for first echo packet
1255         while True:
1256             p = self.pg0.wait_for_packet(1)
1257             self.logger.debug(ppp("Got packet:", p))
1258             if p[UDP].dport == BFD.udp_dport_echo:
1259                 self.logger.debug(ppp("Looping back packet:", p))
1260                 p[Ether].dst = self.pg0.local_mac
1261                 self.pg0.add_stream(p)
1262                 self.pg_start()
1263                 break
1264             elif p.haslayer(BFD):
1265                 # ignore BFD
1266                 pass
1267             else:
1268                 raise Exception(ppp("Received unknown packet:", p))
1269         self.vapi.bfd_udp_del_echo_source()
1270         self.test_session.send_packet()
1271         # echo packets shouldn't arrive anymore
1272         for dummy in range(5):
1273             wait_for_bfd_packet(
1274                 self, pcap_time_min=time.time() - self.vpp_clock_offset)
1275             self.test_session.send_packet()
1276             events = self.vapi.collect_events()
1277             self.assert_equal(len(events), 0, "number of bfd events")
1278
1279     @unittest.skipUnless(running_extended_tests, "part of extended tests")
1280     def test_stale_echo(self):
1281         """ stale echo packets don't keep a session up """
1282         bfd_session_up(self)
1283         self.test_session.update(required_min_echo_rx=150000)
1284         self.vapi.bfd_udp_set_echo_source(self.loopback0.sw_if_index)
1285         self.test_session.send_packet()
1286         # should be turned on - loopback echo packets
1287         echo_packet = None
1288         timeout_at = None
1289         timeout_ok = False
1290         for dummy in range(10 * self.vpp_session.detect_mult):
1291             p = self.pg0.wait_for_packet(1)
1292             if p[UDP].dport == BFD.udp_dport_echo:
1293                 if echo_packet is None:
1294                     self.logger.debug(ppp("Got first echo packet:", p))
1295                     echo_packet = p
1296                     timeout_at = time.time() + self.vpp_session.detect_mult * \
1297                         self.test_session.required_min_echo_rx / USEC_IN_SEC
1298                 else:
1299                     self.logger.debug(ppp("Got followup echo packet:", p))
1300                 self.logger.debug(ppp("Looping back first echo packet:", p))
1301                 echo_packet[Ether].dst = self.pg0.local_mac
1302                 self.pg0.add_stream(echo_packet)
1303                 self.pg_start()
1304             elif p.haslayer(BFD):
1305                 self.logger.debug(ppp("Got packet:", p))
1306                 if "P" in p.sprintf("%BFD.flags%"):
1307                     final = self.test_session.create_packet()
1308                     final[BFD].flags = "F"
1309                     self.test_session.send_packet(final)
1310                 if p[BFD].state == BFDState.down:
1311                     self.assertIsNotNone(
1312                         timeout_at,
1313                         "Session went down before first echo packet received")
1314                     now = time.time()
1315                     self.assertGreaterEqual(
1316                         now, timeout_at,
1317                         "Session timeout at %s, but is expected at %s" %
1318                         (now, timeout_at))
1319                     self.assert_equal(p[BFD].diag,
1320                                       BFDDiagCode.echo_function_failed,
1321                                       BFDDiagCode)
1322                     events = self.vapi.collect_events()
1323                     self.assert_equal(len(events), 1, "number of bfd events")
1324                     self.assert_equal(events[0].state, BFDState.down, BFDState)
1325                     timeout_ok = True
1326                     break
1327             else:
1328                 raise Exception(ppp("Received unknown packet:", p))
1329             self.test_session.send_packet()
1330         self.assertTrue(timeout_ok, "Expected timeout event didn't occur")
1331
1332     @unittest.skipUnless(running_extended_tests, "part of extended tests")
1333     def test_invalid_echo_checksum(self):
1334         """ echo packets with invalid checksum don't keep a session up """
1335         bfd_session_up(self)
1336         self.test_session.update(required_min_echo_rx=150000)
1337         self.vapi.bfd_udp_set_echo_source(self.loopback0.sw_if_index)
1338         self.test_session.send_packet()
1339         # should be turned on - loopback echo packets
1340         timeout_at = None
1341         timeout_ok = False
1342         for dummy in range(10 * self.vpp_session.detect_mult):
1343             p = self.pg0.wait_for_packet(1)
1344             if p[UDP].dport == BFD.udp_dport_echo:
1345                 self.logger.debug(ppp("Got echo packet:", p))
1346                 if timeout_at is None:
1347                     timeout_at = time.time() + self.vpp_session.detect_mult * \
1348                         self.test_session.required_min_echo_rx / USEC_IN_SEC
1349                 p[BFD_vpp_echo].checksum = getrandbits(64)
1350                 p[Ether].dst = self.pg0.local_mac
1351                 self.logger.debug(ppp("Looping back modified echo packet:", p))
1352                 self.pg0.add_stream(p)
1353                 self.pg_start()
1354             elif p.haslayer(BFD):
1355                 self.logger.debug(ppp("Got packet:", p))
1356                 if "P" in p.sprintf("%BFD.flags%"):
1357                     final = self.test_session.create_packet()
1358                     final[BFD].flags = "F"
1359                     self.test_session.send_packet(final)
1360                 if p[BFD].state == BFDState.down:
1361                     self.assertIsNotNone(
1362                         timeout_at,
1363                         "Session went down before first echo packet received")
1364                     now = time.time()
1365                     self.assertGreaterEqual(
1366                         now, timeout_at,
1367                         "Session timeout at %s, but is expected at %s" %
1368                         (now, timeout_at))
1369                     self.assert_equal(p[BFD].diag,
1370                                       BFDDiagCode.echo_function_failed,
1371                                       BFDDiagCode)
1372                     events = self.vapi.collect_events()
1373                     self.assert_equal(len(events), 1, "number of bfd events")
1374                     self.assert_equal(events[0].state, BFDState.down, BFDState)
1375                     timeout_ok = True
1376                     break
1377             else:
1378                 raise Exception(ppp("Received unknown packet:", p))
1379             self.test_session.send_packet()
1380         self.assertTrue(timeout_ok, "Expected timeout event didn't occur")
1381
1382     @unittest.skipUnless(running_extended_tests, "part of extended tests")
1383     def test_admin_up_down(self):
1384         """ put session admin-up and admin-down """
1385         bfd_session_up(self)
1386         self.vpp_session.admin_down()
1387         self.pg0.enable_capture()
1388         e = self.vapi.wait_for_event(1, "bfd_udp_session_details")
1389         verify_event(self, e, expected_state=BFDState.admin_down)
1390         for dummy in range(2):
1391             p = wait_for_bfd_packet(self)
1392             self.assert_equal(p[BFD].state, BFDState.admin_down, BFDState)
1393         # try to bring session up - shouldn't be possible
1394         self.test_session.update(state=BFDState.init)
1395         self.test_session.send_packet()
1396         for dummy in range(2):
1397             p = wait_for_bfd_packet(self)
1398             self.assert_equal(p[BFD].state, BFDState.admin_down, BFDState)
1399         self.vpp_session.admin_up()
1400         self.test_session.update(state=BFDState.down)
1401         e = self.vapi.wait_for_event(1, "bfd_udp_session_details")
1402         verify_event(self, e, expected_state=BFDState.down)
1403         p = wait_for_bfd_packet(
1404             self, pcap_time_min=time.time() - self.vpp_clock_offset)
1405         self.assert_equal(p[BFD].state, BFDState.down, BFDState)
1406         self.test_session.send_packet()
1407         p = wait_for_bfd_packet(
1408             self, pcap_time_min=time.time() - self.vpp_clock_offset)
1409         self.assert_equal(p[BFD].state, BFDState.init, BFDState)
1410         e = self.vapi.wait_for_event(1, "bfd_udp_session_details")
1411         verify_event(self, e, expected_state=BFDState.init)
1412         self.test_session.update(state=BFDState.up)
1413         self.test_session.send_packet()
1414         p = wait_for_bfd_packet(
1415             self, pcap_time_min=time.time() - self.vpp_clock_offset)
1416         self.assert_equal(p[BFD].state, BFDState.up, BFDState)
1417         e = self.vapi.wait_for_event(1, "bfd_udp_session_details")
1418         verify_event(self, e, expected_state=BFDState.up)
1419
1420     @unittest.skipUnless(running_extended_tests, "part of extended tests")
1421     def test_config_change_remote_demand(self):
1422         """ configuration change while peer in demand mode """
1423         bfd_session_up(self)
1424         demand = self.test_session.create_packet()
1425         demand[BFD].flags = "D"
1426         self.test_session.send_packet(demand)
1427         self.vpp_session.modify_parameters(
1428             required_min_rx=2 * self.vpp_session.required_min_rx)
1429         p = wait_for_bfd_packet(
1430             self, pcap_time_min=time.time() - self.vpp_clock_offset)
1431         # poll bit must be set
1432         self.assertIn("P", p.sprintf("%BFD.flags%"), "Poll bit not set")
1433         # terminate poll sequence
1434         final = self.test_session.create_packet()
1435         final[BFD].flags = "D+F"
1436         self.test_session.send_packet(final)
1437         # vpp should be quiet now again
1438         transmit_time = 0.9 \
1439             * max(self.vpp_session.required_min_rx,
1440                   self.test_session.desired_min_tx) \
1441             / USEC_IN_SEC
1442         count = 0
1443         for dummy in range(self.test_session.detect_mult * 2):
1444             time.sleep(transmit_time)
1445             self.test_session.send_packet(demand)
1446             try:
1447                 p = wait_for_bfd_packet(self, timeout=0)
1448                 self.logger.error(ppp("Received unexpected packet:", p))
1449                 count += 1
1450             except CaptureTimeoutError:
1451                 pass
1452         events = self.vapi.collect_events()
1453         for e in events:
1454             self.logger.error("Received unexpected event: %s", e)
1455         self.assert_equal(count, 0, "number of packets received")
1456         self.assert_equal(len(events), 0, "number of events received")
1457
1458     def test_intf_deleted(self):
1459         """ interface with bfd session deleted """
1460         intf = VppLoInterface(self)
1461         intf.config_ip4()
1462         intf.admin_up()
1463         sw_if_index = intf.sw_if_index
1464         vpp_session = VppBFDUDPSession(self, intf, intf.remote_ip4)
1465         vpp_session.add_vpp_config()
1466         vpp_session.admin_up()
1467         intf.remove_vpp_config()
1468         e = self.vapi.wait_for_event(1, "bfd_udp_session_details")
1469         self.assert_equal(e.sw_if_index, sw_if_index, "sw_if_index")
1470         self.assertFalse(vpp_session.query_vpp_config())
1471
1472
1473 @unittest.skipUnless(running_extended_tests, "part of extended tests")
1474 class BFD6TestCase(VppTestCase):
1475     """Bidirectional Forwarding Detection (BFD) (IPv6) """
1476
1477     pg0 = None
1478     vpp_clock_offset = None
1479     vpp_session = None
1480     test_session = None
1481
1482     @classmethod
1483     def setUpClass(cls):
1484         super(BFD6TestCase, cls).setUpClass()
1485         cls.vapi.cli("set log class bfd level debug")
1486         try:
1487             cls.create_pg_interfaces([0])
1488             cls.pg0.config_ip6()
1489             cls.pg0.configure_ipv6_neighbors()
1490             cls.pg0.admin_up()
1491             cls.pg0.resolve_ndp()
1492             cls.create_loopback_interfaces(1)
1493             cls.loopback0 = cls.lo_interfaces[0]
1494             cls.loopback0.config_ip6()
1495             cls.loopback0.admin_up()
1496
1497         except Exception:
1498             super(BFD6TestCase, cls).tearDownClass()
1499             raise
1500
1501     def setUp(self):
1502         super(BFD6TestCase, self).setUp()
1503         self.factory = AuthKeyFactory()
1504         self.vapi.want_bfd_events()
1505         self.pg0.enable_capture()
1506         try:
1507             self.vpp_session = VppBFDUDPSession(self, self.pg0,
1508                                                 self.pg0.remote_ip6,
1509                                                 af=AF_INET6)
1510             self.vpp_session.add_vpp_config()
1511             self.vpp_session.admin_up()
1512             self.test_session = BFDTestSession(self, self.pg0, AF_INET6)
1513             self.logger.debug(self.vapi.cli("show adj nbr"))
1514         except:
1515             self.vapi.want_bfd_events(enable_disable=0)
1516             raise
1517
1518     def tearDown(self):
1519         if not self.vpp_dead:
1520             self.vapi.want_bfd_events(enable_disable=0)
1521         self.vapi.collect_events()  # clear the event queue
1522         super(BFD6TestCase, self).tearDown()
1523
1524     def test_session_up(self):
1525         """ bring BFD session up """
1526         bfd_session_up(self)
1527
1528     def test_session_up_by_ip(self):
1529         """ bring BFD session up - first frame looked up by address pair """
1530         self.logger.info("BFD: Sending Slow control frame")
1531         self.test_session.update(my_discriminator=randint(0, 40000000))
1532         self.test_session.send_packet()
1533         self.pg0.enable_capture()
1534         p = self.pg0.wait_for_packet(1)
1535         self.assert_equal(p[BFD].your_discriminator,
1536                           self.test_session.my_discriminator,
1537                           "BFD - your discriminator")
1538         self.assert_equal(p[BFD].state, BFDState.init, BFDState)
1539         self.test_session.update(your_discriminator=p[BFD].my_discriminator,
1540                                  state=BFDState.up)
1541         self.logger.info("BFD: Waiting for event")
1542         e = self.vapi.wait_for_event(1, "bfd_udp_session_details")
1543         verify_event(self, e, expected_state=BFDState.init)
1544         self.logger.info("BFD: Sending Up")
1545         self.test_session.send_packet()
1546         self.logger.info("BFD: Waiting for event")
1547         e = self.vapi.wait_for_event(1, "bfd_udp_session_details")
1548         verify_event(self, e, expected_state=BFDState.up)
1549         self.logger.info("BFD: Session is Up")
1550         self.test_session.update(state=BFDState.up)
1551         self.test_session.send_packet()
1552         self.assert_equal(self.vpp_session.state, BFDState.up, BFDState)
1553
1554     @unittest.skipUnless(running_extended_tests, "part of extended tests")
1555     def test_hold_up(self):
1556         """ hold BFD session up """
1557         bfd_session_up(self)
1558         for dummy in range(self.test_session.detect_mult * 2):
1559             wait_for_bfd_packet(self)
1560             self.test_session.send_packet()
1561         self.assert_equal(len(self.vapi.collect_events()), 0,
1562                           "number of bfd events")
1563         self.assert_equal(self.vpp_session.state, BFDState.up, BFDState)
1564
1565     def test_echo_looped_back(self):
1566         """ echo packets looped back """
1567         # don't need a session in this case..
1568         self.vpp_session.remove_vpp_config()
1569         self.pg0.enable_capture()
1570         echo_packet_count = 10
1571         # random source port low enough to increment a few times..
1572         udp_sport_tx = randint(1, 50000)
1573         udp_sport_rx = udp_sport_tx
1574         echo_packet = (Ether(src=self.pg0.remote_mac,
1575                              dst=self.pg0.local_mac) /
1576                        IPv6(src=self.pg0.remote_ip6,
1577                             dst=self.pg0.remote_ip6) /
1578                        UDP(dport=BFD.udp_dport_echo) /
1579                        Raw("this should be looped back"))
1580         for dummy in range(echo_packet_count):
1581             self.sleep(.01, "delay between echo packets")
1582             echo_packet[UDP].sport = udp_sport_tx
1583             udp_sport_tx += 1
1584             self.logger.debug(ppp("Sending packet:", echo_packet))
1585             self.pg0.add_stream(echo_packet)
1586             self.pg_start()
1587         for dummy in range(echo_packet_count):
1588             p = self.pg0.wait_for_packet(1)
1589             self.logger.debug(ppp("Got packet:", p))
1590             ether = p[Ether]
1591             self.assert_equal(self.pg0.remote_mac,
1592                               ether.dst, "Destination MAC")
1593             self.assert_equal(self.pg0.local_mac, ether.src, "Source MAC")
1594             ip = p[IPv6]
1595             self.assert_equal(self.pg0.remote_ip6, ip.dst, "Destination IP")
1596             self.assert_equal(self.pg0.remote_ip6, ip.src, "Destination IP")
1597             udp = p[UDP]
1598             self.assert_equal(udp.dport, BFD.udp_dport_echo,
1599                               "UDP destination port")
1600             self.assert_equal(udp.sport, udp_sport_rx, "UDP source port")
1601             udp_sport_rx += 1
1602             # need to compare the hex payload here, otherwise BFD_vpp_echo
1603             # gets in way
1604             self.assertEqual(str(p[UDP].payload),
1605                              str(echo_packet[UDP].payload),
1606                              "Received packet is not the echo packet sent")
1607         self.assert_equal(udp_sport_tx, udp_sport_rx, "UDP source port (== "
1608                           "ECHO packet identifier for test purposes)")
1609         self.assert_equal(udp_sport_tx, udp_sport_rx, "UDP source port (== "
1610                           "ECHO packet identifier for test purposes)")
1611
1612     def test_echo(self):
1613         """ echo function """
1614         bfd_session_up(self)
1615         self.test_session.update(required_min_echo_rx=150000)
1616         self.test_session.send_packet()
1617         detection_time = self.test_session.detect_mult *\
1618             self.vpp_session.required_min_rx / USEC_IN_SEC
1619         # echo shouldn't work without echo source set
1620         for dummy in range(10):
1621             sleep = self.vpp_session.required_min_rx / USEC_IN_SEC
1622             self.sleep(sleep, "delay before sending bfd packet")
1623             self.test_session.send_packet()
1624         p = wait_for_bfd_packet(
1625             self, pcap_time_min=time.time() - self.vpp_clock_offset)
1626         self.assert_equal(p[BFD].required_min_rx_interval,
1627                           self.vpp_session.required_min_rx,
1628                           "BFD required min rx interval")
1629         self.test_session.send_packet()
1630         self.vapi.bfd_udp_set_echo_source(self.loopback0.sw_if_index)
1631         echo_seen = False
1632         # should be turned on - loopback echo packets
1633         for dummy in range(3):
1634             loop_until = time.time() + 0.75 * detection_time
1635             while time.time() < loop_until:
1636                 p = self.pg0.wait_for_packet(1)
1637                 self.logger.debug(ppp("Got packet:", p))
1638                 if p[UDP].dport == BFD.udp_dport_echo:
1639                     self.assert_equal(
1640                         p[IPv6].dst, self.pg0.local_ip6, "BFD ECHO dst IP")
1641                     self.assertNotEqual(p[IPv6].src, self.loopback0.local_ip6,
1642                                         "BFD ECHO src IP equal to loopback IP")
1643                     self.logger.debug(ppp("Looping back packet:", p))
1644                     self.assert_equal(p[Ether].dst, self.pg0.remote_mac,
1645                                       "ECHO packet destination MAC address")
1646                     p[Ether].dst = self.pg0.local_mac
1647                     self.pg0.add_stream(p)
1648                     self.pg_start()
1649                     echo_seen = True
1650                 elif p.haslayer(BFD):
1651                     if echo_seen:
1652                         self.assertGreaterEqual(
1653                             p[BFD].required_min_rx_interval,
1654                             1000000)
1655                     if "P" in p.sprintf("%BFD.flags%"):
1656                         final = self.test_session.create_packet()
1657                         final[BFD].flags = "F"
1658                         self.test_session.send_packet(final)
1659                 else:
1660                     raise Exception(ppp("Received unknown packet:", p))
1661
1662                 self.assert_equal(len(self.vapi.collect_events()), 0,
1663                                   "number of bfd events")
1664             self.test_session.send_packet()
1665         self.assertTrue(echo_seen, "No echo packets received")
1666
1667     def test_intf_deleted(self):
1668         """ interface with bfd session deleted """
1669         intf = VppLoInterface(self)
1670         intf.config_ip6()
1671         intf.admin_up()
1672         sw_if_index = intf.sw_if_index
1673         vpp_session = VppBFDUDPSession(
1674             self, intf, intf.remote_ip6, af=AF_INET6)
1675         vpp_session.add_vpp_config()
1676         vpp_session.admin_up()
1677         intf.remove_vpp_config()
1678         e = self.vapi.wait_for_event(1, "bfd_udp_session_details")
1679         self.assert_equal(e.sw_if_index, sw_if_index, "sw_if_index")
1680         self.assertFalse(vpp_session.query_vpp_config())
1681
1682
1683 @unittest.skipUnless(running_extended_tests, "part of extended tests")
1684 class BFDFIBTestCase(VppTestCase):
1685     """ BFD-FIB interactions (IPv6) """
1686
1687     vpp_session = None
1688     test_session = None
1689
1690     def setUp(self):
1691         super(BFDFIBTestCase, self).setUp()
1692         self.create_pg_interfaces(range(1))
1693
1694         self.vapi.want_bfd_events()
1695         self.pg0.enable_capture()
1696
1697         for i in self.pg_interfaces:
1698             i.admin_up()
1699             i.config_ip6()
1700             i.configure_ipv6_neighbors()
1701
1702     def tearDown(self):
1703         if not self.vpp_dead:
1704             self.vapi.want_bfd_events(enable_disable=0)
1705
1706         super(BFDFIBTestCase, self).tearDown()
1707
1708     @staticmethod
1709     def pkt_is_not_data_traffic(p):
1710         """ not data traffic implies BFD or the usual IPv6 ND/RA"""
1711         if p.haslayer(BFD) or is_ipv6_misc(p):
1712             return True
1713         return False
1714
1715     def test_session_with_fib(self):
1716         """ BFD-FIB interactions """
1717
1718         # packets to match against both of the routes
1719         p = [(Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1720               IPv6(src="3001::1", dst="2001::1") /
1721               UDP(sport=1234, dport=1234) /
1722               Raw('\xa5' * 100)),
1723              (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1724               IPv6(src="3001::1", dst="2002::1") /
1725               UDP(sport=1234, dport=1234) /
1726               Raw('\xa5' * 100))]
1727
1728         # A recursive and a non-recursive route via a next-hop that
1729         # will have a BFD session
1730         ip_2001_s_64 = VppIpRoute(self, "2001::", 64,
1731                                   [VppRoutePath(self.pg0.remote_ip6,
1732                                                 self.pg0.sw_if_index,
1733                                                 proto=DpoProto.DPO_PROTO_IP6)],
1734                                   is_ip6=1)
1735         ip_2002_s_64 = VppIpRoute(self, "2002::", 64,
1736                                   [VppRoutePath(self.pg0.remote_ip6,
1737                                                 0xffffffff,
1738                                                 proto=DpoProto.DPO_PROTO_IP6)],
1739                                   is_ip6=1)
1740         ip_2001_s_64.add_vpp_config()
1741         ip_2002_s_64.add_vpp_config()
1742
1743         # bring the session up now the routes are present
1744         self.vpp_session = VppBFDUDPSession(self,
1745                                             self.pg0,
1746                                             self.pg0.remote_ip6,
1747                                             af=AF_INET6)
1748         self.vpp_session.add_vpp_config()
1749         self.vpp_session.admin_up()
1750         self.test_session = BFDTestSession(self, self.pg0, AF_INET6)
1751
1752         # session is up - traffic passes
1753         bfd_session_up(self)
1754
1755         self.pg0.add_stream(p)
1756         self.pg_start()
1757         for packet in p:
1758             captured = self.pg0.wait_for_packet(
1759                 1,
1760                 filter_out_fn=self.pkt_is_not_data_traffic)
1761             self.assertEqual(captured[IPv6].dst,
1762                              packet[IPv6].dst)
1763
1764         # session is up - traffic is dropped
1765         bfd_session_down(self)
1766
1767         self.pg0.add_stream(p)
1768         self.pg_start()
1769         with self.assertRaises(CaptureTimeoutError):
1770             self.pg0.wait_for_packet(1, self.pkt_is_not_data_traffic)
1771
1772         # session is up - traffic passes
1773         bfd_session_up(self)
1774
1775         self.pg0.add_stream(p)
1776         self.pg_start()
1777         for packet in p:
1778             captured = self.pg0.wait_for_packet(
1779                 1,
1780                 filter_out_fn=self.pkt_is_not_data_traffic)
1781             self.assertEqual(captured[IPv6].dst,
1782                              packet[IPv6].dst)
1783
1784
1785 @unittest.skipUnless(running_extended_tests, "part of extended tests")
1786 class BFDSHA1TestCase(VppTestCase):
1787     """Bidirectional Forwarding Detection (BFD) (SHA1 auth) """
1788
1789     pg0 = None
1790     vpp_clock_offset = None
1791     vpp_session = None
1792     test_session = None
1793
1794     @classmethod
1795     def setUpClass(cls):
1796         super(BFDSHA1TestCase, cls).setUpClass()
1797         cls.vapi.cli("set log class bfd level debug")
1798         try:
1799             cls.create_pg_interfaces([0])
1800             cls.pg0.config_ip4()
1801             cls.pg0.admin_up()
1802             cls.pg0.resolve_arp()
1803
1804         except Exception:
1805             super(BFDSHA1TestCase, cls).tearDownClass()
1806             raise
1807
1808     def setUp(self):
1809         super(BFDSHA1TestCase, self).setUp()
1810         self.factory = AuthKeyFactory()
1811         self.vapi.want_bfd_events()
1812         self.pg0.enable_capture()
1813
1814     def tearDown(self):
1815         if not self.vpp_dead:
1816             self.vapi.want_bfd_events(enable_disable=0)
1817         self.vapi.collect_events()  # clear the event queue
1818         super(BFDSHA1TestCase, self).tearDown()
1819
1820     def test_session_up(self):
1821         """ bring BFD session up """
1822         key = self.factory.create_random_key(self)
1823         key.add_vpp_config()
1824         self.vpp_session = VppBFDUDPSession(self, self.pg0,
1825                                             self.pg0.remote_ip4,
1826                                             sha1_key=key)
1827         self.vpp_session.add_vpp_config()
1828         self.vpp_session.admin_up()
1829         self.test_session = BFDTestSession(
1830             self, self.pg0, AF_INET, sha1_key=key,
1831             bfd_key_id=self.vpp_session.bfd_key_id)
1832         bfd_session_up(self)
1833
1834     @unittest.skipUnless(running_extended_tests, "part of extended tests")
1835     def test_hold_up(self):
1836         """ hold BFD session up """
1837         key = self.factory.create_random_key(self)
1838         key.add_vpp_config()
1839         self.vpp_session = VppBFDUDPSession(self, self.pg0,
1840                                             self.pg0.remote_ip4,
1841                                             sha1_key=key)
1842         self.vpp_session.add_vpp_config()
1843         self.vpp_session.admin_up()
1844         self.test_session = BFDTestSession(
1845             self, self.pg0, AF_INET, sha1_key=key,
1846             bfd_key_id=self.vpp_session.bfd_key_id)
1847         bfd_session_up(self)
1848         for dummy in range(self.test_session.detect_mult * 2):
1849             wait_for_bfd_packet(self)
1850             self.test_session.send_packet()
1851         self.assert_equal(self.vpp_session.state, BFDState.up, BFDState)
1852
1853     @unittest.skipUnless(running_extended_tests, "part of extended tests")
1854     def test_hold_up_meticulous(self):
1855         """ hold BFD session up - meticulous auth """
1856         key = self.factory.create_random_key(
1857             self, BFDAuthType.meticulous_keyed_sha1)
1858         key.add_vpp_config()
1859         self.vpp_session = VppBFDUDPSession(self, self.pg0,
1860                                             self.pg0.remote_ip4, sha1_key=key)
1861         self.vpp_session.add_vpp_config()
1862         self.vpp_session.admin_up()
1863         # specify sequence number so that it wraps
1864         self.test_session = BFDTestSession(
1865             self, self.pg0, AF_INET, sha1_key=key,
1866             bfd_key_id=self.vpp_session.bfd_key_id,
1867             our_seq_number=0xFFFFFFFF - 4)
1868         bfd_session_up(self)
1869         for dummy in range(30):
1870             wait_for_bfd_packet(self)
1871             self.test_session.inc_seq_num()
1872             self.test_session.send_packet()
1873         self.assert_equal(self.vpp_session.state, BFDState.up, BFDState)
1874
1875     @unittest.skipUnless(running_extended_tests, "part of extended tests")
1876     def test_send_bad_seq_number(self):
1877         """ session is not kept alive by msgs with bad sequence numbers"""
1878         key = self.factory.create_random_key(
1879             self, BFDAuthType.meticulous_keyed_sha1)
1880         key.add_vpp_config()
1881         self.vpp_session = VppBFDUDPSession(self, self.pg0,
1882                                             self.pg0.remote_ip4, sha1_key=key)
1883         self.vpp_session.add_vpp_config()
1884         self.test_session = BFDTestSession(
1885             self, self.pg0, AF_INET, sha1_key=key,
1886             bfd_key_id=self.vpp_session.bfd_key_id)
1887         bfd_session_up(self)
1888         detection_time = self.test_session.detect_mult *\
1889             self.vpp_session.required_min_rx / USEC_IN_SEC
1890         send_until = time.time() + 2 * detection_time
1891         while time.time() < send_until:
1892             self.test_session.send_packet()
1893             self.sleep(0.7 * self.vpp_session.required_min_rx / USEC_IN_SEC,
1894                        "time between bfd packets")
1895         e = self.vapi.collect_events()
1896         # session should be down now, because the sequence numbers weren't
1897         # updated
1898         self.assert_equal(len(e), 1, "number of bfd events")
1899         verify_event(self, e[0], expected_state=BFDState.down)
1900
1901     def execute_rogue_session_scenario(self, vpp_bfd_udp_session,
1902                                        legitimate_test_session,
1903                                        rogue_test_session,
1904                                        rogue_bfd_values=None):
1905         """ execute a rogue session interaction scenario
1906
1907         1. create vpp session, add config
1908         2. bring the legitimate session up
1909         3. copy the bfd values from legitimate session to rogue session
1910         4. apply rogue_bfd_values to rogue session
1911         5. set rogue session state to down
1912         6. send message to take the session down from the rogue session
1913         7. assert that the legitimate session is unaffected
1914         """
1915
1916         self.vpp_session = vpp_bfd_udp_session
1917         self.vpp_session.add_vpp_config()
1918         self.test_session = legitimate_test_session
1919         # bring vpp session up
1920         bfd_session_up(self)
1921         # send packet from rogue session
1922         rogue_test_session.update(
1923             my_discriminator=self.test_session.my_discriminator,
1924             your_discriminator=self.test_session.your_discriminator,
1925             desired_min_tx=self.test_session.desired_min_tx,
1926             required_min_rx=self.test_session.required_min_rx,
1927             detect_mult=self.test_session.detect_mult,
1928             diag=self.test_session.diag,
1929             state=self.test_session.state,
1930             auth_type=self.test_session.auth_type)
1931         if rogue_bfd_values:
1932             rogue_test_session.update(**rogue_bfd_values)
1933         rogue_test_session.update(state=BFDState.down)
1934         rogue_test_session.send_packet()
1935         wait_for_bfd_packet(self)
1936         self.assert_equal(self.vpp_session.state, BFDState.up, BFDState)
1937
1938     @unittest.skipUnless(running_extended_tests, "part of extended tests")
1939     def test_mismatch_auth(self):
1940         """ session is not brought down by unauthenticated msg """
1941         key = self.factory.create_random_key(self)
1942         key.add_vpp_config()
1943         vpp_session = VppBFDUDPSession(
1944             self, self.pg0, self.pg0.remote_ip4, sha1_key=key)
1945         legitimate_test_session = BFDTestSession(
1946             self, self.pg0, AF_INET, sha1_key=key,
1947             bfd_key_id=vpp_session.bfd_key_id)
1948         rogue_test_session = BFDTestSession(self, self.pg0, AF_INET)
1949         self.execute_rogue_session_scenario(vpp_session,
1950                                             legitimate_test_session,
1951                                             rogue_test_session)
1952
1953     @unittest.skipUnless(running_extended_tests, "part of extended tests")
1954     def test_mismatch_bfd_key_id(self):
1955         """ session is not brought down by msg with non-existent key-id """
1956         key = self.factory.create_random_key(self)
1957         key.add_vpp_config()
1958         vpp_session = VppBFDUDPSession(
1959             self, self.pg0, self.pg0.remote_ip4, sha1_key=key)
1960         # pick a different random bfd key id
1961         x = randint(0, 255)
1962         while x == vpp_session.bfd_key_id:
1963             x = randint(0, 255)
1964         legitimate_test_session = BFDTestSession(
1965             self, self.pg0, AF_INET, sha1_key=key,
1966             bfd_key_id=vpp_session.bfd_key_id)
1967         rogue_test_session = BFDTestSession(
1968             self, self.pg0, AF_INET, sha1_key=key, bfd_key_id=x)
1969         self.execute_rogue_session_scenario(vpp_session,
1970                                             legitimate_test_session,
1971                                             rogue_test_session)
1972
1973     @unittest.skipUnless(running_extended_tests, "part of extended tests")
1974     def test_mismatched_auth_type(self):
1975         """ session is not brought down by msg with wrong auth type """
1976         key = self.factory.create_random_key(self)
1977         key.add_vpp_config()
1978         vpp_session = VppBFDUDPSession(
1979             self, self.pg0, self.pg0.remote_ip4, sha1_key=key)
1980         legitimate_test_session = BFDTestSession(
1981             self, self.pg0, AF_INET, sha1_key=key,
1982             bfd_key_id=vpp_session.bfd_key_id)
1983         rogue_test_session = BFDTestSession(
1984             self, self.pg0, AF_INET, sha1_key=key,
1985             bfd_key_id=vpp_session.bfd_key_id)
1986         self.execute_rogue_session_scenario(
1987             vpp_session, legitimate_test_session, rogue_test_session,
1988             {'auth_type': BFDAuthType.keyed_md5})
1989
1990     @unittest.skipUnless(running_extended_tests, "part of extended tests")
1991     def test_restart(self):
1992         """ simulate remote peer restart and resynchronization """
1993         key = self.factory.create_random_key(
1994             self, BFDAuthType.meticulous_keyed_sha1)
1995         key.add_vpp_config()
1996         self.vpp_session = VppBFDUDPSession(self, self.pg0,
1997                                             self.pg0.remote_ip4, sha1_key=key)
1998         self.vpp_session.add_vpp_config()
1999         self.test_session = BFDTestSession(
2000             self, self.pg0, AF_INET, sha1_key=key,
2001             bfd_key_id=self.vpp_session.bfd_key_id, our_seq_number=0)
2002         bfd_session_up(self)
2003         # don't send any packets for 2*detection_time
2004         detection_time = self.test_session.detect_mult *\
2005             self.vpp_session.required_min_rx / USEC_IN_SEC
2006         self.sleep(2 * detection_time, "simulating peer restart")
2007         events = self.vapi.collect_events()
2008         self.assert_equal(len(events), 1, "number of bfd events")
2009         verify_event(self, events[0], expected_state=BFDState.down)
2010         self.test_session.update(state=BFDState.down)
2011         # reset sequence number
2012         self.test_session.our_seq_number = 0
2013         self.test_session.vpp_seq_number = None
2014         # now throw away any pending packets
2015         self.pg0.enable_capture()
2016         bfd_session_up(self)
2017
2018
2019 @unittest.skipUnless(running_extended_tests, "part of extended tests")
2020 class BFDAuthOnOffTestCase(VppTestCase):
2021     """Bidirectional Forwarding Detection (BFD) (changing auth) """
2022
2023     pg0 = None
2024     vpp_session = None
2025     test_session = None
2026
2027     @classmethod
2028     def setUpClass(cls):
2029         super(BFDAuthOnOffTestCase, cls).setUpClass()
2030         cls.vapi.cli("set log class bfd level debug")
2031         try:
2032             cls.create_pg_interfaces([0])
2033             cls.pg0.config_ip4()
2034             cls.pg0.admin_up()
2035             cls.pg0.resolve_arp()
2036
2037         except Exception:
2038             super(BFDAuthOnOffTestCase, cls).tearDownClass()
2039             raise
2040
2041     def setUp(self):
2042         super(BFDAuthOnOffTestCase, self).setUp()
2043         self.factory = AuthKeyFactory()
2044         self.vapi.want_bfd_events()
2045         self.pg0.enable_capture()
2046
2047     def tearDown(self):
2048         if not self.vpp_dead:
2049             self.vapi.want_bfd_events(enable_disable=0)
2050         self.vapi.collect_events()  # clear the event queue
2051         super(BFDAuthOnOffTestCase, self).tearDown()
2052
2053     def test_auth_on_immediate(self):
2054         """ turn auth on without disturbing session state (immediate) """
2055         key = self.factory.create_random_key(self)
2056         key.add_vpp_config()
2057         self.vpp_session = VppBFDUDPSession(self, self.pg0,
2058                                             self.pg0.remote_ip4)
2059         self.vpp_session.add_vpp_config()
2060         self.test_session = BFDTestSession(self, self.pg0, AF_INET)
2061         bfd_session_up(self)
2062         for dummy in range(self.test_session.detect_mult * 2):
2063             p = wait_for_bfd_packet(self)
2064             self.assert_equal(p[BFD].state, BFDState.up, BFDState)
2065             self.test_session.send_packet()
2066         self.vpp_session.activate_auth(key)
2067         self.test_session.bfd_key_id = self.vpp_session.bfd_key_id
2068         self.test_session.sha1_key = key
2069         for dummy in range(self.test_session.detect_mult * 2):
2070             p = wait_for_bfd_packet(self)
2071             self.assert_equal(p[BFD].state, BFDState.up, BFDState)
2072             self.test_session.send_packet()
2073         self.assert_equal(self.vpp_session.state, BFDState.up, BFDState)
2074         self.assert_equal(len(self.vapi.collect_events()), 0,
2075                           "number of bfd events")
2076
2077     def test_auth_off_immediate(self):
2078         """ turn auth off without disturbing session state (immediate) """
2079         key = self.factory.create_random_key(self)
2080         key.add_vpp_config()
2081         self.vpp_session = VppBFDUDPSession(self, self.pg0,
2082                                             self.pg0.remote_ip4, sha1_key=key)
2083         self.vpp_session.add_vpp_config()
2084         self.test_session = BFDTestSession(
2085             self, self.pg0, AF_INET, sha1_key=key,
2086             bfd_key_id=self.vpp_session.bfd_key_id)
2087         bfd_session_up(self)
2088         # self.vapi.want_bfd_events(enable_disable=0)
2089         for dummy in range(self.test_session.detect_mult * 2):
2090             p = wait_for_bfd_packet(self)
2091             self.assert_equal(p[BFD].state, BFDState.up, BFDState)
2092             self.test_session.inc_seq_num()
2093             self.test_session.send_packet()
2094         self.vpp_session.deactivate_auth()
2095         self.test_session.bfd_key_id = None
2096         self.test_session.sha1_key = None
2097         for dummy in range(self.test_session.detect_mult * 2):
2098             p = wait_for_bfd_packet(self)
2099             self.assert_equal(p[BFD].state, BFDState.up, BFDState)
2100             self.test_session.inc_seq_num()
2101             self.test_session.send_packet()
2102         self.assert_equal(self.vpp_session.state, BFDState.up, BFDState)
2103         self.assert_equal(len(self.vapi.collect_events()), 0,
2104                           "number of bfd events")
2105
2106     def test_auth_change_key_immediate(self):
2107         """ change auth key without disturbing session state (immediate) """
2108         key1 = self.factory.create_random_key(self)
2109         key1.add_vpp_config()
2110         key2 = self.factory.create_random_key(self)
2111         key2.add_vpp_config()
2112         self.vpp_session = VppBFDUDPSession(self, self.pg0,
2113                                             self.pg0.remote_ip4, sha1_key=key1)
2114         self.vpp_session.add_vpp_config()
2115         self.test_session = BFDTestSession(
2116             self, self.pg0, AF_INET, sha1_key=key1,
2117             bfd_key_id=self.vpp_session.bfd_key_id)
2118         bfd_session_up(self)
2119         for dummy in range(self.test_session.detect_mult * 2):
2120             p = wait_for_bfd_packet(self)
2121             self.assert_equal(p[BFD].state, BFDState.up, BFDState)
2122             self.test_session.send_packet()
2123         self.vpp_session.activate_auth(key2)
2124         self.test_session.bfd_key_id = self.vpp_session.bfd_key_id
2125         self.test_session.sha1_key = key2
2126         for dummy in range(self.test_session.detect_mult * 2):
2127             p = wait_for_bfd_packet(self)
2128             self.assert_equal(p[BFD].state, BFDState.up, BFDState)
2129             self.test_session.send_packet()
2130         self.assert_equal(self.vpp_session.state, BFDState.up, BFDState)
2131         self.assert_equal(len(self.vapi.collect_events()), 0,
2132                           "number of bfd events")
2133
2134     def test_auth_on_delayed(self):
2135         """ turn auth on without disturbing session state (delayed) """
2136         key = self.factory.create_random_key(self)
2137         key.add_vpp_config()
2138         self.vpp_session = VppBFDUDPSession(self, self.pg0,
2139                                             self.pg0.remote_ip4)
2140         self.vpp_session.add_vpp_config()
2141         self.test_session = BFDTestSession(self, self.pg0, AF_INET)
2142         bfd_session_up(self)
2143         for dummy in range(self.test_session.detect_mult * 2):
2144             wait_for_bfd_packet(self)
2145             self.test_session.send_packet()
2146         self.vpp_session.activate_auth(key, delayed=True)
2147         for dummy in range(self.test_session.detect_mult * 2):
2148             p = wait_for_bfd_packet(self)
2149             self.assert_equal(p[BFD].state, BFDState.up, BFDState)
2150             self.test_session.send_packet()
2151         self.test_session.bfd_key_id = self.vpp_session.bfd_key_id
2152         self.test_session.sha1_key = key
2153         self.test_session.send_packet()
2154         for dummy in range(self.test_session.detect_mult * 2):
2155             p = wait_for_bfd_packet(self)
2156             self.assert_equal(p[BFD].state, BFDState.up, BFDState)
2157             self.test_session.send_packet()
2158         self.assert_equal(self.vpp_session.state, BFDState.up, BFDState)
2159         self.assert_equal(len(self.vapi.collect_events()), 0,
2160                           "number of bfd events")
2161
2162     def test_auth_off_delayed(self):
2163         """ turn auth off without disturbing session state (delayed) """
2164         key = self.factory.create_random_key(self)
2165         key.add_vpp_config()
2166         self.vpp_session = VppBFDUDPSession(self, self.pg0,
2167                                             self.pg0.remote_ip4, sha1_key=key)
2168         self.vpp_session.add_vpp_config()
2169         self.test_session = BFDTestSession(
2170             self, self.pg0, AF_INET, sha1_key=key,
2171             bfd_key_id=self.vpp_session.bfd_key_id)
2172         bfd_session_up(self)
2173         for dummy in range(self.test_session.detect_mult * 2):
2174             p = wait_for_bfd_packet(self)
2175             self.assert_equal(p[BFD].state, BFDState.up, BFDState)
2176             self.test_session.send_packet()
2177         self.vpp_session.deactivate_auth(delayed=True)
2178         for dummy in range(self.test_session.detect_mult * 2):
2179             p = wait_for_bfd_packet(self)
2180             self.assert_equal(p[BFD].state, BFDState.up, BFDState)
2181             self.test_session.send_packet()
2182         self.test_session.bfd_key_id = None
2183         self.test_session.sha1_key = None
2184         self.test_session.send_packet()
2185         for dummy in range(self.test_session.detect_mult * 2):
2186             p = wait_for_bfd_packet(self)
2187             self.assert_equal(p[BFD].state, BFDState.up, BFDState)
2188             self.test_session.send_packet()
2189         self.assert_equal(self.vpp_session.state, BFDState.up, BFDState)
2190         self.assert_equal(len(self.vapi.collect_events()), 0,
2191                           "number of bfd events")
2192
2193     def test_auth_change_key_delayed(self):
2194         """ change auth key without disturbing session state (delayed) """
2195         key1 = self.factory.create_random_key(self)
2196         key1.add_vpp_config()
2197         key2 = self.factory.create_random_key(self)
2198         key2.add_vpp_config()
2199         self.vpp_session = VppBFDUDPSession(self, self.pg0,
2200                                             self.pg0.remote_ip4, sha1_key=key1)
2201         self.vpp_session.add_vpp_config()
2202         self.vpp_session.admin_up()
2203         self.test_session = BFDTestSession(
2204             self, self.pg0, AF_INET, sha1_key=key1,
2205             bfd_key_id=self.vpp_session.bfd_key_id)
2206         bfd_session_up(self)
2207         for dummy in range(self.test_session.detect_mult * 2):
2208             p = wait_for_bfd_packet(self)
2209             self.assert_equal(p[BFD].state, BFDState.up, BFDState)
2210             self.test_session.send_packet()
2211         self.vpp_session.activate_auth(key2, delayed=True)
2212         for dummy in range(self.test_session.detect_mult * 2):
2213             p = wait_for_bfd_packet(self)
2214             self.assert_equal(p[BFD].state, BFDState.up, BFDState)
2215             self.test_session.send_packet()
2216         self.test_session.bfd_key_id = self.vpp_session.bfd_key_id
2217         self.test_session.sha1_key = key2
2218         self.test_session.send_packet()
2219         for dummy in range(self.test_session.detect_mult * 2):
2220             p = wait_for_bfd_packet(self)
2221             self.assert_equal(p[BFD].state, BFDState.up, BFDState)
2222             self.test_session.send_packet()
2223         self.assert_equal(self.vpp_session.state, BFDState.up, BFDState)
2224         self.assert_equal(len(self.vapi.collect_events()), 0,
2225                           "number of bfd events")
2226
2227
2228 @unittest.skipUnless(running_extended_tests, "part of extended tests")
2229 class BFDCLITestCase(VppTestCase):
2230     """Bidirectional Forwarding Detection (BFD) (CLI) """
2231     pg0 = None
2232
2233     @classmethod
2234     def setUpClass(cls):
2235         super(BFDCLITestCase, cls).setUpClass()
2236         cls.vapi.cli("set log class bfd level debug")
2237         try:
2238             cls.create_pg_interfaces((0,))
2239             cls.pg0.config_ip4()
2240             cls.pg0.config_ip6()
2241             cls.pg0.resolve_arp()
2242             cls.pg0.resolve_ndp()
2243
2244         except Exception:
2245             super(BFDCLITestCase, cls).tearDownClass()
2246             raise
2247
2248     def setUp(self):
2249         super(BFDCLITestCase, self).setUp()
2250         self.factory = AuthKeyFactory()
2251         self.pg0.enable_capture()
2252
2253     def tearDown(self):
2254         try:
2255             self.vapi.want_bfd_events(enable_disable=0)
2256         except UnexpectedApiReturnValueError:
2257             # some tests aren't subscribed, so this is not an issue
2258             pass
2259         self.vapi.collect_events()  # clear the event queue
2260         super(BFDCLITestCase, self).tearDown()
2261
2262     def cli_verify_no_response(self, cli):
2263         """ execute a CLI, asserting that the response is empty """
2264         self.assert_equal(self.vapi.cli(cli),
2265                           "",
2266                           "CLI command response")
2267
2268     def cli_verify_response(self, cli, expected):
2269         """ execute a CLI, asserting that the response matches expectation """
2270         self.assert_equal(self.vapi.cli(cli).strip(),
2271                           expected,
2272                           "CLI command response")
2273
2274     def test_show(self):
2275         """ show commands """
2276         k1 = self.factory.create_random_key(self)
2277         k1.add_vpp_config()
2278         k2 = self.factory.create_random_key(
2279             self, auth_type=BFDAuthType.meticulous_keyed_sha1)
2280         k2.add_vpp_config()
2281         s1 = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
2282         s1.add_vpp_config()
2283         s2 = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip6, af=AF_INET6,
2284                               sha1_key=k2)
2285         s2.add_vpp_config()
2286         self.logger.info(self.vapi.ppcli("show bfd keys"))
2287         self.logger.info(self.vapi.ppcli("show bfd sessions"))
2288         self.logger.info(self.vapi.ppcli("show bfd"))
2289
2290     def test_set_del_sha1_key(self):
2291         """ set/delete SHA1 auth key """
2292         k = self.factory.create_random_key(self)
2293         self.registry.register(k, self.logger)
2294         self.cli_verify_no_response(
2295             "bfd key set conf-key-id %s type keyed-sha1 secret %s" %
2296             (k.conf_key_id,
2297                 "".join("{:02x}".format(ord(c)) for c in k.key)))
2298         self.assertTrue(k.query_vpp_config())
2299         self.vpp_session = VppBFDUDPSession(
2300             self, self.pg0, self.pg0.remote_ip4, sha1_key=k)
2301         self.vpp_session.add_vpp_config()
2302         self.test_session = \
2303             BFDTestSession(self, self.pg0, AF_INET, sha1_key=k,
2304                            bfd_key_id=self.vpp_session.bfd_key_id)
2305         self.vapi.want_bfd_events()
2306         bfd_session_up(self)
2307         bfd_session_down(self)
2308         # try to replace the secret for the key - should fail because the key
2309         # is in-use
2310         k2 = self.factory.create_random_key(self)
2311         self.cli_verify_response(
2312             "bfd key set conf-key-id %s type keyed-sha1 secret %s" %
2313             (k.conf_key_id,
2314                 "".join("{:02x}".format(ord(c)) for c in k2.key)),
2315             "bfd key set: `bfd_auth_set_key' API call failed, "
2316             "rv=-103:BFD object in use")
2317         # manipulating the session using old secret should still work
2318         bfd_session_up(self)
2319         bfd_session_down(self)
2320         self.vpp_session.remove_vpp_config()
2321         self.cli_verify_no_response(
2322             "bfd key del conf-key-id %s" % k.conf_key_id)
2323         self.assertFalse(k.query_vpp_config())
2324
2325     def test_set_del_meticulous_sha1_key(self):
2326         """ set/delete meticulous SHA1 auth key """
2327         k = self.factory.create_random_key(
2328             self, auth_type=BFDAuthType.meticulous_keyed_sha1)
2329         self.registry.register(k, self.logger)
2330         self.cli_verify_no_response(
2331             "bfd key set conf-key-id %s type meticulous-keyed-sha1 secret %s" %
2332             (k.conf_key_id,
2333                 "".join("{:02x}".format(ord(c)) for c in k.key)))
2334         self.assertTrue(k.query_vpp_config())
2335         self.vpp_session = VppBFDUDPSession(self, self.pg0,
2336                                             self.pg0.remote_ip6, af=AF_INET6,
2337                                             sha1_key=k)
2338         self.vpp_session.add_vpp_config()
2339         self.vpp_session.admin_up()
2340         self.test_session = \
2341             BFDTestSession(self, self.pg0, AF_INET6, sha1_key=k,
2342                            bfd_key_id=self.vpp_session.bfd_key_id)
2343         self.vapi.want_bfd_events()
2344         bfd_session_up(self)
2345         bfd_session_down(self)
2346         # try to replace the secret for the key - should fail because the key
2347         # is in-use
2348         k2 = self.factory.create_random_key(self)
2349         self.cli_verify_response(
2350             "bfd key set conf-key-id %s type keyed-sha1 secret %s" %
2351             (k.conf_key_id,
2352                 "".join("{:02x}".format(ord(c)) for c in k2.key)),
2353             "bfd key set: `bfd_auth_set_key' API call failed, "
2354             "rv=-103:BFD object in use")
2355         # manipulating the session using old secret should still work
2356         bfd_session_up(self)
2357         bfd_session_down(self)
2358         self.vpp_session.remove_vpp_config()
2359         self.cli_verify_no_response(
2360             "bfd key del conf-key-id %s" % k.conf_key_id)
2361         self.assertFalse(k.query_vpp_config())
2362
2363     def test_add_mod_del_bfd_udp(self):
2364         """ create/modify/delete IPv4 BFD UDP session """
2365         vpp_session = VppBFDUDPSession(
2366             self, self.pg0, self.pg0.remote_ip4)
2367         self.registry.register(vpp_session, self.logger)
2368         cli_add_cmd = "bfd udp session add interface %s local-addr %s " \
2369             "peer-addr %s desired-min-tx %s required-min-rx %s "\
2370             "detect-mult %s" % (self.pg0.name, self.pg0.local_ip4,
2371                                 self.pg0.remote_ip4,
2372                                 vpp_session.desired_min_tx,
2373                                 vpp_session.required_min_rx,
2374                                 vpp_session.detect_mult)
2375         self.cli_verify_no_response(cli_add_cmd)
2376         # 2nd add should fail
2377         self.cli_verify_response(
2378             cli_add_cmd,
2379             "bfd udp session add: `bfd_add_add_session' API call"
2380             " failed, rv=-101:Duplicate BFD object")
2381         verify_bfd_session_config(self, vpp_session)
2382         mod_session = VppBFDUDPSession(
2383             self, self.pg0, self.pg0.remote_ip4,
2384             required_min_rx=2 * vpp_session.required_min_rx,
2385             desired_min_tx=3 * vpp_session.desired_min_tx,
2386             detect_mult=4 * vpp_session.detect_mult)
2387         self.cli_verify_no_response(
2388             "bfd udp session mod interface %s local-addr %s peer-addr %s "
2389             "desired-min-tx %s required-min-rx %s detect-mult %s" %
2390             (self.pg0.name, self.pg0.local_ip4, self.pg0.remote_ip4,
2391              mod_session.desired_min_tx, mod_session.required_min_rx,
2392              mod_session.detect_mult))
2393         verify_bfd_session_config(self, mod_session)
2394         cli_del_cmd = "bfd udp session del interface %s local-addr %s "\
2395             "peer-addr %s" % (self.pg0.name,
2396                               self.pg0.local_ip4, self.pg0.remote_ip4)
2397         self.cli_verify_no_response(cli_del_cmd)
2398         # 2nd del is expected to fail
2399         self.cli_verify_response(
2400             cli_del_cmd, "bfd udp session del: `bfd_udp_del_session' API call"
2401             " failed, rv=-102:No such BFD object")
2402         self.assertFalse(vpp_session.query_vpp_config())
2403
2404     def test_add_mod_del_bfd_udp6(self):
2405         """ create/modify/delete IPv6 BFD UDP session """
2406         vpp_session = VppBFDUDPSession(
2407             self, self.pg0, self.pg0.remote_ip6, af=AF_INET6)
2408         self.registry.register(vpp_session, self.logger)
2409         cli_add_cmd = "bfd udp session add interface %s local-addr %s " \
2410             "peer-addr %s desired-min-tx %s required-min-rx %s "\
2411             "detect-mult %s" % (self.pg0.name, self.pg0.local_ip6,
2412                                 self.pg0.remote_ip6,
2413                                 vpp_session.desired_min_tx,
2414                                 vpp_session.required_min_rx,
2415                                 vpp_session.detect_mult)
2416         self.cli_verify_no_response(cli_add_cmd)
2417         # 2nd add should fail
2418         self.cli_verify_response(
2419             cli_add_cmd,
2420             "bfd udp session add: `bfd_add_add_session' API call"
2421             " failed, rv=-101:Duplicate BFD object")
2422         verify_bfd_session_config(self, vpp_session)
2423         mod_session = VppBFDUDPSession(
2424             self, self.pg0, self.pg0.remote_ip6, af=AF_INET6,
2425             required_min_rx=2 * vpp_session.required_min_rx,
2426             desired_min_tx=3 * vpp_session.desired_min_tx,
2427             detect_mult=4 * vpp_session.detect_mult)
2428         self.cli_verify_no_response(
2429             "bfd udp session mod interface %s local-addr %s peer-addr %s "
2430             "desired-min-tx %s required-min-rx %s detect-mult %s" %
2431             (self.pg0.name, self.pg0.local_ip6, self.pg0.remote_ip6,
2432              mod_session.desired_min_tx,
2433              mod_session.required_min_rx, mod_session.detect_mult))
2434         verify_bfd_session_config(self, mod_session)
2435         cli_del_cmd = "bfd udp session del interface %s local-addr %s "\
2436             "peer-addr %s" % (self.pg0.name,
2437                               self.pg0.local_ip6, self.pg0.remote_ip6)
2438         self.cli_verify_no_response(cli_del_cmd)
2439         # 2nd del is expected to fail
2440         self.cli_verify_response(
2441             cli_del_cmd,
2442             "bfd udp session del: `bfd_udp_del_session' API call"
2443             " failed, rv=-102:No such BFD object")
2444         self.assertFalse(vpp_session.query_vpp_config())
2445
2446     def test_add_mod_del_bfd_udp_auth(self):
2447         """ create/modify/delete IPv4 BFD UDP session (authenticated) """
2448         key = self.factory.create_random_key(self)
2449         key.add_vpp_config()
2450         vpp_session = VppBFDUDPSession(
2451             self, self.pg0, self.pg0.remote_ip4, sha1_key=key)
2452         self.registry.register(vpp_session, self.logger)
2453         cli_add_cmd = "bfd udp session add interface %s local-addr %s " \
2454             "peer-addr %s desired-min-tx %s required-min-rx %s "\
2455             "detect-mult %s conf-key-id %s bfd-key-id %s"\
2456             % (self.pg0.name, self.pg0.local_ip4, self.pg0.remote_ip4,
2457                vpp_session.desired_min_tx, vpp_session.required_min_rx,
2458                vpp_session.detect_mult, key.conf_key_id,
2459                vpp_session.bfd_key_id)
2460         self.cli_verify_no_response(cli_add_cmd)
2461         # 2nd add should fail
2462         self.cli_verify_response(
2463             cli_add_cmd,
2464             "bfd udp session add: `bfd_add_add_session' API call"
2465             " failed, rv=-101:Duplicate BFD object")
2466         verify_bfd_session_config(self, vpp_session)
2467         mod_session = VppBFDUDPSession(
2468             self, self.pg0, self.pg0.remote_ip4, sha1_key=key,
2469             bfd_key_id=vpp_session.bfd_key_id,
2470             required_min_rx=2 * vpp_session.required_min_rx,
2471             desired_min_tx=3 * vpp_session.desired_min_tx,
2472             detect_mult=4 * vpp_session.detect_mult)
2473         self.cli_verify_no_response(
2474             "bfd udp session mod interface %s local-addr %s peer-addr %s "
2475             "desired-min-tx %s required-min-rx %s detect-mult %s" %
2476             (self.pg0.name, self.pg0.local_ip4, self.pg0.remote_ip4,
2477              mod_session.desired_min_tx,
2478              mod_session.required_min_rx, mod_session.detect_mult))
2479         verify_bfd_session_config(self, mod_session)
2480         cli_del_cmd = "bfd udp session del interface %s local-addr %s "\
2481             "peer-addr %s" % (self.pg0.name,
2482                               self.pg0.local_ip4, self.pg0.remote_ip4)
2483         self.cli_verify_no_response(cli_del_cmd)
2484         # 2nd del is expected to fail
2485         self.cli_verify_response(
2486             cli_del_cmd,
2487             "bfd udp session del: `bfd_udp_del_session' API call"
2488             " failed, rv=-102:No such BFD object")
2489         self.assertFalse(vpp_session.query_vpp_config())
2490
2491     def test_add_mod_del_bfd_udp6_auth(self):
2492         """ create/modify/delete IPv6 BFD UDP session (authenticated) """
2493         key = self.factory.create_random_key(
2494             self, auth_type=BFDAuthType.meticulous_keyed_sha1)
2495         key.add_vpp_config()
2496         vpp_session = VppBFDUDPSession(
2497             self, self.pg0, self.pg0.remote_ip6, af=AF_INET6, sha1_key=key)
2498         self.registry.register(vpp_session, self.logger)
2499         cli_add_cmd = "bfd udp session add interface %s local-addr %s " \
2500             "peer-addr %s desired-min-tx %s required-min-rx %s "\
2501             "detect-mult %s conf-key-id %s bfd-key-id %s" \
2502             % (self.pg0.name, self.pg0.local_ip6, self.pg0.remote_ip6,
2503                vpp_session.desired_min_tx, vpp_session.required_min_rx,
2504                vpp_session.detect_mult, key.conf_key_id,
2505                vpp_session.bfd_key_id)
2506         self.cli_verify_no_response(cli_add_cmd)
2507         # 2nd add should fail
2508         self.cli_verify_response(
2509             cli_add_cmd,
2510             "bfd udp session add: `bfd_add_add_session' API call"
2511             " failed, rv=-101:Duplicate BFD object")
2512         verify_bfd_session_config(self, vpp_session)
2513         mod_session = VppBFDUDPSession(
2514             self, self.pg0, self.pg0.remote_ip6, af=AF_INET6, sha1_key=key,
2515             bfd_key_id=vpp_session.bfd_key_id,
2516             required_min_rx=2 * vpp_session.required_min_rx,
2517             desired_min_tx=3 * vpp_session.desired_min_tx,
2518             detect_mult=4 * vpp_session.detect_mult)
2519         self.cli_verify_no_response(
2520             "bfd udp session mod interface %s local-addr %s peer-addr %s "
2521             "desired-min-tx %s required-min-rx %s detect-mult %s" %
2522             (self.pg0.name, self.pg0.local_ip6, self.pg0.remote_ip6,
2523              mod_session.desired_min_tx,
2524              mod_session.required_min_rx, mod_session.detect_mult))
2525         verify_bfd_session_config(self, mod_session)
2526         cli_del_cmd = "bfd udp session del interface %s local-addr %s "\
2527             "peer-addr %s" % (self.pg0.name,
2528                               self.pg0.local_ip6, self.pg0.remote_ip6)
2529         self.cli_verify_no_response(cli_del_cmd)
2530         # 2nd del is expected to fail
2531         self.cli_verify_response(
2532             cli_del_cmd,
2533             "bfd udp session del: `bfd_udp_del_session' API call"
2534             " failed, rv=-102:No such BFD object")
2535         self.assertFalse(vpp_session.query_vpp_config())
2536
2537     def test_auth_on_off(self):
2538         """ turn authentication on and off """
2539         key = self.factory.create_random_key(
2540             self, auth_type=BFDAuthType.meticulous_keyed_sha1)
2541         key.add_vpp_config()
2542         session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
2543         auth_session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4,
2544                                         sha1_key=key)
2545         session.add_vpp_config()
2546         cli_activate = \
2547             "bfd udp session auth activate interface %s local-addr %s "\
2548             "peer-addr %s conf-key-id %s bfd-key-id %s"\
2549             % (self.pg0.name, self.pg0.local_ip4, self.pg0.remote_ip4,
2550                key.conf_key_id, auth_session.bfd_key_id)
2551         self.cli_verify_no_response(cli_activate)
2552         verify_bfd_session_config(self, auth_session)
2553         self.cli_verify_no_response(cli_activate)
2554         verify_bfd_session_config(self, auth_session)
2555         cli_deactivate = \
2556             "bfd udp session auth deactivate interface %s local-addr %s "\
2557             "peer-addr %s "\
2558             % (self.pg0.name, self.pg0.local_ip4, self.pg0.remote_ip4)
2559         self.cli_verify_no_response(cli_deactivate)
2560         verify_bfd_session_config(self, session)
2561         self.cli_verify_no_response(cli_deactivate)
2562         verify_bfd_session_config(self, session)
2563
2564     def test_auth_on_off_delayed(self):
2565         """ turn authentication on and off (delayed) """
2566         key = self.factory.create_random_key(
2567             self, auth_type=BFDAuthType.meticulous_keyed_sha1)
2568         key.add_vpp_config()
2569         session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
2570         auth_session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4,
2571                                         sha1_key=key)
2572         session.add_vpp_config()
2573         cli_activate = \
2574             "bfd udp session auth activate interface %s local-addr %s "\
2575             "peer-addr %s conf-key-id %s bfd-key-id %s delayed yes"\
2576             % (self.pg0.name, self.pg0.local_ip4, self.pg0.remote_ip4,
2577                key.conf_key_id, auth_session.bfd_key_id)
2578         self.cli_verify_no_response(cli_activate)
2579         verify_bfd_session_config(self, auth_session)
2580         self.cli_verify_no_response(cli_activate)
2581         verify_bfd_session_config(self, auth_session)
2582         cli_deactivate = \
2583             "bfd udp session auth deactivate interface %s local-addr %s "\
2584             "peer-addr %s delayed yes"\
2585             % (self.pg0.name, self.pg0.local_ip4, self.pg0.remote_ip4)
2586         self.cli_verify_no_response(cli_deactivate)
2587         verify_bfd_session_config(self, session)
2588         self.cli_verify_no_response(cli_deactivate)
2589         verify_bfd_session_config(self, session)
2590
2591     def test_admin_up_down(self):
2592         """ put session admin-up and admin-down """
2593         session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
2594         session.add_vpp_config()
2595         cli_down = \
2596             "bfd udp session set-flags admin down interface %s local-addr %s "\
2597             "peer-addr %s "\
2598             % (self.pg0.name, self.pg0.local_ip4, self.pg0.remote_ip4)
2599         cli_up = \
2600             "bfd udp session set-flags admin up interface %s local-addr %s "\
2601             "peer-addr %s "\
2602             % (self.pg0.name, self.pg0.local_ip4, self.pg0.remote_ip4)
2603         self.cli_verify_no_response(cli_down)
2604         verify_bfd_session_config(self, session, state=BFDState.admin_down)
2605         self.cli_verify_no_response(cli_up)
2606         verify_bfd_session_config(self, session, state=BFDState.down)
2607
2608     def test_set_del_udp_echo_source(self):
2609         """ set/del udp echo source """
2610         self.create_loopback_interfaces(1)
2611         self.loopback0 = self.lo_interfaces[0]
2612         self.loopback0.admin_up()
2613         self.cli_verify_response("show bfd echo-source",
2614                                  "UDP echo source is not set.")
2615         cli_set = "bfd udp echo-source set interface %s" % self.loopback0.name
2616         self.cli_verify_no_response(cli_set)
2617         self.cli_verify_response("show bfd echo-source",
2618                                  "UDP echo source is: %s\n"
2619                                  "IPv4 address usable as echo source: none\n"
2620                                  "IPv6 address usable as echo source: none" %
2621                                  self.loopback0.name)
2622         self.loopback0.config_ip4()
2623         unpacked = unpack("!L", self.loopback0.local_ip4n)
2624         echo_ip4 = inet_ntop(AF_INET, pack("!L", unpacked[0] ^ 1))
2625         self.cli_verify_response("show bfd echo-source",
2626                                  "UDP echo source is: %s\n"
2627                                  "IPv4 address usable as echo source: %s\n"
2628                                  "IPv6 address usable as echo source: none" %
2629                                  (self.loopback0.name, echo_ip4))
2630         unpacked = unpack("!LLLL", self.loopback0.local_ip6n)
2631         echo_ip6 = inet_ntop(AF_INET6, pack("!LLLL", unpacked[0], unpacked[1],
2632                                             unpacked[2], unpacked[3] ^ 1))
2633         self.loopback0.config_ip6()
2634         self.cli_verify_response("show bfd echo-source",
2635                                  "UDP echo source is: %s\n"
2636                                  "IPv4 address usable as echo source: %s\n"
2637                                  "IPv6 address usable as echo source: %s" %
2638                                  (self.loopback0.name, echo_ip4, echo_ip6))
2639         cli_del = "bfd udp echo-source del"
2640         self.cli_verify_no_response(cli_del)
2641         self.cli_verify_response("show bfd echo-source",
2642                                  "UDP echo source is not set.")
2643
2644 if __name__ == '__main__':
2645     unittest.main(testRunner=VppTestRunner)