4 from __future__ import division
7 from collections import namedtuple
13 from random import randint, shuffle, getrandbits
14 from socket import AF_INET, AF_INET6
17 from scapy.layers.inet import UDP, IP
18 from scapy.layers.inet6 import IPv6
19 from scapy.layers.l2 import Ether, GRE
20 from scapy.packet import Raw
22 from config import config
32 from framework import VppTestCase
33 from asfframework import (
34 tag_fixme_vpp_workers,
40 from vpp_ip import DpoProto
41 from vpp_ip_route import VppIpRoute, VppRoutePath
42 from vpp_lo_interface import VppLoInterface
43 from vpp_papi_provider import UnexpectedApiReturnValueError, CliFailedCommandError
44 from vpp_pg_interface import CaptureTimeoutError, is_ipv6_misc
45 from vpp_gre_interface import VppGreInterface
50 class AuthKeyFactory(object):
51 """Factory class for creating auth keys with unique conf key ID"""
54 self._conf_key_ids = {}
56 def create_random_key(self, test, auth_type=BFDAuthType.keyed_sha1):
57 """create a random key with unique conf key id"""
58 conf_key_id = randint(0, 0xFFFFFFFF)
59 while conf_key_id in self._conf_key_ids:
60 conf_key_id = randint(0, 0xFFFFFFFF)
61 self._conf_key_ids[conf_key_id] = 1
62 key = scapy.compat.raw(
63 bytearray([randint(0, 255) for _ in range(randint(1, 20))])
66 test=test, auth_type=auth_type, conf_key_id=conf_key_id, key=key
70 class BFDAPITestCase(VppTestCase):
71 """Bidirectional Forwarding Detection (BFD) - API"""
78 super(BFDAPITestCase, cls).setUpClass()
79 cls.vapi.cli("set log class bfd level debug")
81 cls.create_pg_interfaces(range(2))
82 for i in cls.pg_interfaces:
88 super(BFDAPITestCase, cls).tearDownClass()
92 def tearDownClass(cls):
93 super(BFDAPITestCase, cls).tearDownClass()
96 super(BFDAPITestCase, self).setUp()
97 self.factory = AuthKeyFactory()
99 def test_add_bfd(self):
100 """create a BFD session"""
101 session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
102 session.add_vpp_config()
103 self.logger.debug("Session state is %s", session.state)
104 session.remove_vpp_config()
105 session.add_vpp_config()
106 self.logger.debug("Session state is %s", session.state)
107 session.remove_vpp_config()
109 def test_double_add(self):
110 """create the same BFD session twice (negative case)"""
111 session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
112 session.add_vpp_config()
114 with self.vapi.assert_negative_api_retval():
115 session.add_vpp_config()
117 session.remove_vpp_config()
119 def test_add_bfd6(self):
120 """create IPv6 BFD session"""
121 session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip6, af=AF_INET6)
122 session.add_vpp_config()
123 self.logger.debug("Session state is %s", session.state)
124 session.remove_vpp_config()
125 session.add_vpp_config()
126 self.logger.debug("Session state is %s", session.state)
127 session.remove_vpp_config()
129 def test_mod_bfd(self):
130 """modify BFD session parameters"""
131 session = VppBFDUDPSession(
135 desired_min_tx=50000,
136 required_min_rx=10000,
139 session.add_vpp_config()
140 s = session.get_bfd_udp_session_dump_entry()
142 session.desired_min_tx, s.desired_min_tx, "desired min transmit interval"
145 session.required_min_rx, s.required_min_rx, "required min receive interval"
147 self.assert_equal(session.detect_mult, s.detect_mult, "detect mult")
148 session.modify_parameters(
149 desired_min_tx=session.desired_min_tx * 2,
150 required_min_rx=session.required_min_rx * 2,
151 detect_mult=session.detect_mult * 2,
153 s = session.get_bfd_udp_session_dump_entry()
155 session.desired_min_tx, s.desired_min_tx, "desired min transmit interval"
158 session.required_min_rx, s.required_min_rx, "required min receive interval"
160 self.assert_equal(session.detect_mult, s.detect_mult, "detect mult")
162 def test_upd_bfd(self):
163 """Create/Modify w/ Update BFD session parameters"""
164 session = VppBFDUDPSession(
168 desired_min_tx=50000,
169 required_min_rx=10000,
172 session.upd_vpp_config()
173 s = session.get_bfd_udp_session_dump_entry()
175 session.desired_min_tx, s.desired_min_tx, "desired min transmit interval"
178 session.required_min_rx, s.required_min_rx, "required min receive interval"
181 self.assert_equal(session.detect_mult, s.detect_mult, "detect mult")
182 session.upd_vpp_config(
183 desired_min_tx=session.desired_min_tx * 2,
184 required_min_rx=session.required_min_rx * 2,
185 detect_mult=session.detect_mult * 2,
187 s = session.get_bfd_udp_session_dump_entry()
189 session.desired_min_tx, s.desired_min_tx, "desired min transmit interval"
192 session.required_min_rx, s.required_min_rx, "required min receive interval"
194 self.assert_equal(session.detect_mult, s.detect_mult, "detect mult")
196 def test_add_sha1_keys(self):
199 keys = [self.factory.create_random_key(self) for i in range(0, key_count)]
201 self.assertFalse(key.query_vpp_config())
205 self.assertTrue(key.query_vpp_config())
207 indexes = list(range(key_count))
212 key.remove_vpp_config()
214 for j in range(key_count):
217 self.assertFalse(key.query_vpp_config())
219 self.assertTrue(key.query_vpp_config())
220 # should be removed now
222 self.assertFalse(key.query_vpp_config())
223 # add back and remove again
227 self.assertTrue(key.query_vpp_config())
229 key.remove_vpp_config()
231 self.assertFalse(key.query_vpp_config())
233 def test_add_bfd_sha1(self):
234 """create a BFD session (SHA1)"""
235 key = self.factory.create_random_key(self)
237 session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4, sha1_key=key)
238 session.add_vpp_config()
239 self.logger.debug("Session state is %s", session.state)
240 session.remove_vpp_config()
241 session.add_vpp_config()
242 self.logger.debug("Session state is %s", session.state)
243 session.remove_vpp_config()
245 def test_double_add_sha1(self):
246 """create the same BFD session twice (negative case) (SHA1)"""
247 key = self.factory.create_random_key(self)
249 session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4, sha1_key=key)
250 session.add_vpp_config()
251 with self.assertRaises(Exception):
252 session.add_vpp_config()
254 def test_add_auth_nonexistent_key(self):
255 """create BFD session using non-existent SHA1 (negative case)"""
256 session = VppBFDUDPSession(
260 sha1_key=self.factory.create_random_key(self),
262 with self.assertRaises(Exception):
263 session.add_vpp_config()
265 def test_shared_sha1_key(self):
266 """single SHA1 key shared by multiple BFD sessions"""
267 key = self.factory.create_random_key(self)
270 VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4, sha1_key=key),
272 self, self.pg0, self.pg0.remote_ip6, sha1_key=key, af=AF_INET6
274 VppBFDUDPSession(self, self.pg1, self.pg1.remote_ip4, sha1_key=key),
276 self, self.pg1, self.pg1.remote_ip6, sha1_key=key, af=AF_INET6
283 e = key.get_bfd_auth_keys_dump_entry()
285 e.use_count, len(sessions) - removed, "Use count for shared key"
287 s.remove_vpp_config()
289 e = key.get_bfd_auth_keys_dump_entry()
291 e.use_count, len(sessions) - removed, "Use count for shared key"
294 def test_activate_auth(self):
295 """activate SHA1 authentication"""
296 key = self.factory.create_random_key(self)
298 session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
299 session.add_vpp_config()
300 session.activate_auth(key)
302 def test_deactivate_auth(self):
303 """deactivate SHA1 authentication"""
304 key = self.factory.create_random_key(self)
306 session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
307 session.add_vpp_config()
308 session.activate_auth(key)
309 session.deactivate_auth()
311 def test_change_key(self):
312 """change SHA1 key"""
313 key1 = self.factory.create_random_key(self)
314 key2 = self.factory.create_random_key(self)
315 while key2.conf_key_id == key1.conf_key_id:
316 key2 = self.factory.create_random_key(self)
317 key1.add_vpp_config()
318 key2.add_vpp_config()
319 session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4, sha1_key=key1)
320 session.add_vpp_config()
321 session.activate_auth(key2)
323 def test_set_del_udp_echo_source(self):
324 """set/del udp echo source"""
325 self.create_loopback_interfaces(1)
326 self.loopback0 = self.lo_interfaces[0]
327 self.loopback0.admin_up()
328 echo_source = self.vapi.bfd_udp_get_echo_source()
329 self.assertFalse(echo_source.is_set)
330 self.assertFalse(echo_source.have_usable_ip4)
331 self.assertFalse(echo_source.have_usable_ip6)
333 self.vapi.bfd_udp_set_echo_source(sw_if_index=self.loopback0.sw_if_index)
334 echo_source = self.vapi.bfd_udp_get_echo_source()
335 self.assertTrue(echo_source.is_set)
336 self.assertEqual(echo_source.sw_if_index, self.loopback0.sw_if_index)
337 self.assertFalse(echo_source.have_usable_ip4)
338 self.assertFalse(echo_source.have_usable_ip6)
340 self.loopback0.config_ip4()
341 echo_ip4 = ipaddress.IPv4Address(
342 int(ipaddress.IPv4Address(self.loopback0.local_ip4)) ^ 1
344 echo_source = self.vapi.bfd_udp_get_echo_source()
345 self.assertTrue(echo_source.is_set)
346 self.assertEqual(echo_source.sw_if_index, self.loopback0.sw_if_index)
347 self.assertTrue(echo_source.have_usable_ip4)
348 self.assertEqual(echo_source.ip4_addr.packed, echo_ip4)
349 self.assertFalse(echo_source.have_usable_ip6)
351 self.loopback0.config_ip6()
352 echo_ip6 = ipaddress.IPv6Address(
353 int(ipaddress.IPv6Address(self.loopback0.local_ip6)) ^ 1
356 echo_source = self.vapi.bfd_udp_get_echo_source()
357 self.assertTrue(echo_source.is_set)
358 self.assertEqual(echo_source.sw_if_index, self.loopback0.sw_if_index)
359 self.assertTrue(echo_source.have_usable_ip4)
360 self.assertEqual(echo_source.ip4_addr.packed, echo_ip4)
361 self.assertTrue(echo_source.have_usable_ip6)
362 self.assertEqual(echo_source.ip6_addr.packed, echo_ip6)
364 self.vapi.bfd_udp_del_echo_source()
365 echo_source = self.vapi.bfd_udp_get_echo_source()
366 self.assertFalse(echo_source.is_set)
367 self.assertFalse(echo_source.have_usable_ip4)
368 self.assertFalse(echo_source.have_usable_ip6)
371 class BFDTestSession(object):
372 """BFD session as seen from test framework side"""
388 self.sha1_key = sha1_key
389 self.bfd_key_id = bfd_key_id
390 self.interface = interface
392 self.phy_interface = phy_interface
394 self.phy_interface = self.interface
395 self.udp_sport = randint(49152, 65535)
396 if our_seq_number is None:
397 self.our_seq_number = randint(0, 40000000)
399 self.our_seq_number = our_seq_number
400 self.vpp_seq_number = None
401 self.my_discriminator = 0
402 self.desired_min_tx = 300000
403 self.required_min_rx = 300000
404 self.required_min_echo_rx = None
405 self.detect_mult = detect_mult
406 self.diag = BFDDiagCode.no_diagnostic
407 self.your_discriminator = None
408 self.state = BFDState.down
409 self.auth_type = BFDAuthType.no_auth
410 self.tunnel_header = tunnel_header
413 self.tx_packets_echo = 0
414 self.rx_packets_echo = 0
416 def inc_seq_num(self):
417 """increment sequence number, wrapping if needed"""
418 if self.our_seq_number == 0xFFFFFFFF:
419 self.our_seq_number = 0
421 self.our_seq_number += 1
425 my_discriminator=None,
426 your_discriminator=None,
428 required_min_rx=None,
429 required_min_echo_rx=None,
435 """update BFD parameters associated with session"""
436 if my_discriminator is not None:
437 self.my_discriminator = my_discriminator
438 if your_discriminator is not None:
439 self.your_discriminator = your_discriminator
440 if required_min_rx is not None:
441 self.required_min_rx = required_min_rx
442 if required_min_echo_rx is not None:
443 self.required_min_echo_rx = required_min_echo_rx
444 if desired_min_tx is not None:
445 self.desired_min_tx = desired_min_tx
446 if detect_mult is not None:
447 self.detect_mult = detect_mult
450 if state is not None:
452 if auth_type is not None:
453 self.auth_type = auth_type
455 def fill_packet_fields(self, packet):
456 """set packet fields with known values in packet"""
458 if self.my_discriminator:
459 self.test.logger.debug(
460 "BFD: setting packet.my_discriminator=%s", self.my_discriminator
462 bfd.my_discriminator = self.my_discriminator
463 if self.your_discriminator:
464 self.test.logger.debug(
465 "BFD: setting packet.your_discriminator=%s", self.your_discriminator
467 bfd.your_discriminator = self.your_discriminator
468 if self.required_min_rx:
469 self.test.logger.debug(
470 "BFD: setting packet.required_min_rx_interval=%s", self.required_min_rx
472 bfd.required_min_rx_interval = self.required_min_rx
473 if self.required_min_echo_rx:
474 self.test.logger.debug(
475 "BFD: setting packet.required_min_echo_rx=%s", self.required_min_echo_rx
477 bfd.required_min_echo_rx_interval = self.required_min_echo_rx
478 if self.desired_min_tx:
479 self.test.logger.debug(
480 "BFD: setting packet.desired_min_tx_interval=%s", self.desired_min_tx
482 bfd.desired_min_tx_interval = self.desired_min_tx
484 self.test.logger.debug(
485 "BFD: setting packet.detect_mult=%s", self.detect_mult
487 bfd.detect_mult = self.detect_mult
489 self.test.logger.debug("BFD: setting packet.diag=%s", self.diag)
492 self.test.logger.debug("BFD: setting packet.state=%s", self.state)
493 bfd.state = self.state
495 # this is used by a negative test-case
496 self.test.logger.debug("BFD: setting packet.auth_type=%s", self.auth_type)
497 bfd.auth_type = self.auth_type
499 def create_packet(self):
500 """create a BFD packet, reflecting the current state of session"""
503 bfd.auth_type = self.sha1_key.auth_type
504 bfd.auth_len = BFD.sha1_auth_len
505 bfd.auth_key_id = self.bfd_key_id
506 bfd.auth_seq_num = self.our_seq_number
507 bfd.length = BFD.sha1_auth_len + BFD.bfd_pkt_len
511 src=self.phy_interface.remote_mac, dst=self.phy_interface.local_mac
513 if self.tunnel_header:
514 packet = packet / self.tunnel_header
515 if self.af == AF_INET6:
519 src=self.interface.remote_ip6,
520 dst=self.interface.local_ip6,
523 / UDP(sport=self.udp_sport, dport=BFD.udp_dport)
530 src=self.interface.remote_ip4, dst=self.interface.local_ip4, ttl=255
532 / UDP(sport=self.udp_sport, dport=BFD.udp_dport)
535 self.test.logger.debug("BFD: Creating packet")
536 self.fill_packet_fields(packet)
539 scapy.compat.raw(packet[BFD])[:32]
541 + b"\0" * (20 - len(self.sha1_key.key))
543 self.test.logger.debug(
544 "BFD: Calculated SHA1 hash: %s"
545 % hashlib.sha1(hash_material).hexdigest()
547 packet[BFD].auth_key_hash = hashlib.sha1(hash_material).digest()
550 def send_packet(self, packet=None, interface=None):
551 """send packet on interface, creating the packet if needed"""
553 packet = self.create_packet()
554 if interface is None:
555 interface = self.phy_interface
556 self.test.logger.debug(ppp("Sending packet:", packet))
557 interface.add_stream(packet)
561 def verify_sha1_auth(self, packet):
562 """Verify correctness of authentication in BFD layer."""
564 self.test.assert_equal(bfd.auth_len, 28, "Auth section length")
565 self.test.assert_equal(bfd.auth_type, self.sha1_key.auth_type, BFDAuthType)
566 self.test.assert_equal(bfd.auth_key_id, self.bfd_key_id, "Key ID")
567 self.test.assert_equal(bfd.auth_reserved, 0, "Reserved")
568 if self.vpp_seq_number is None:
569 self.vpp_seq_number = bfd.auth_seq_num
570 self.test.logger.debug(
571 "Received initial sequence number: %s" % self.vpp_seq_number
574 recvd_seq_num = bfd.auth_seq_num
575 self.test.logger.debug(
576 "Received followup sequence number: %s" % recvd_seq_num
578 if self.vpp_seq_number < 0xFFFFFFFF:
579 if self.sha1_key.auth_type == BFDAuthType.meticulous_keyed_sha1:
580 self.test.assert_equal(
581 recvd_seq_num, self.vpp_seq_number + 1, "BFD sequence number"
584 self.test.assert_in_range(
587 self.vpp_seq_number + 1,
588 "BFD sequence number",
591 if self.sha1_key.auth_type == BFDAuthType.meticulous_keyed_sha1:
592 self.test.assert_equal(recvd_seq_num, 0, "BFD sequence number")
596 (self.vpp_seq_number, 0),
597 "BFD sequence number not one of "
598 "(%s, 0)" % self.vpp_seq_number,
600 self.vpp_seq_number = recvd_seq_num
601 # last 20 bytes represent the hash - so replace them with the key,
602 # pad the result with zeros and hash the result
606 + b"\0" * (20 - len(self.sha1_key.key))
608 expected_hash = hashlib.sha1(hash_material).hexdigest()
609 self.test.assert_equal(
610 binascii.hexlify(bfd.auth_key_hash), expected_hash.encode(), "Auth key hash"
613 def verify_bfd(self, packet):
614 """Verify correctness of BFD layer."""
616 self.test.assert_equal(bfd.version, 1, "BFD version")
617 self.test.assert_equal(
618 bfd.your_discriminator, self.my_discriminator, "BFD - your discriminator"
621 self.verify_sha1_auth(packet)
624 def bfd_session_up(test):
625 """Bring BFD session up"""
626 test.logger.info("BFD: Waiting for slow hello")
627 p = wait_for_bfd_packet(test, 2, is_tunnel=test.vpp_session.is_tunnel)
629 if hasattr(test, "vpp_clock_offset"):
630 old_offset = test.vpp_clock_offset
631 test.vpp_clock_offset = time.time() - float(p.time)
632 test.logger.debug("BFD: Calculated vpp clock offset: %s", test.vpp_clock_offset)
634 test.assertAlmostEqual(
636 test.vpp_clock_offset,
638 msg="vpp clock offset not stable (new: %s, old: %s)"
639 % (test.vpp_clock_offset, old_offset),
641 test.logger.info("BFD: Sending Init")
642 test.test_session.update(
643 my_discriminator=randint(0, 40000000),
644 your_discriminator=p[BFD].my_discriminator,
648 test.test_session.sha1_key
649 and test.test_session.sha1_key.auth_type == BFDAuthType.meticulous_keyed_sha1
651 test.test_session.inc_seq_num()
652 test.test_session.send_packet()
653 test.logger.info("BFD: Waiting for event")
654 e = test.vapi.wait_for_event(1, "bfd_udp_session_event")
655 verify_event(test, e, expected_state=BFDState.up)
656 test.logger.info("BFD: Session is Up")
657 test.test_session.update(state=BFDState.up)
659 test.test_session.sha1_key
660 and test.test_session.sha1_key.auth_type == BFDAuthType.meticulous_keyed_sha1
662 test.test_session.inc_seq_num()
663 test.test_session.send_packet()
664 test.assert_equal(test.vpp_session.state, BFDState.up, BFDState)
667 def bfd_session_down(test):
668 """Bring BFD session down"""
669 test.assert_equal(test.vpp_session.state, BFDState.up, BFDState)
670 test.test_session.update(state=BFDState.down)
672 test.test_session.sha1_key
673 and test.test_session.sha1_key.auth_type == BFDAuthType.meticulous_keyed_sha1
675 test.test_session.inc_seq_num()
676 test.test_session.send_packet()
677 test.logger.info("BFD: Waiting for event")
678 e = test.vapi.wait_for_event(1, "bfd_udp_session_event")
679 verify_event(test, e, expected_state=BFDState.down)
680 test.logger.info("BFD: Session is Down")
681 test.assert_equal(test.vpp_session.state, BFDState.down, BFDState)
684 def verify_bfd_session_config(test, session, state=None):
685 dump = session.get_bfd_udp_session_dump_entry()
686 test.assertIsNotNone(dump)
687 # since dump is not none, we have verified that sw_if_index and addresses
688 # are valid (in get_bfd_udp_session_dump_entry)
690 test.assert_equal(dump.state, state, "session state")
692 dump.required_min_rx, session.required_min_rx, "required min rx interval"
695 dump.desired_min_tx, session.desired_min_tx, "desired min tx interval"
697 test.assert_equal(dump.detect_mult, session.detect_mult, "detect multiplier")
698 if session.sha1_key is None:
699 test.assert_equal(dump.is_authenticated, 0, "is_authenticated flag")
701 test.assert_equal(dump.is_authenticated, 1, "is_authenticated flag")
702 test.assert_equal(dump.bfd_key_id, session.bfd_key_id, "bfd key id")
704 dump.conf_key_id, session.sha1_key.conf_key_id, "config key id"
708 def verify_ip(test, packet):
709 """Verify correctness of IP layer."""
710 if test.vpp_session.af == AF_INET6:
712 local_ip = test.vpp_session.interface.local_ip6
713 remote_ip = test.vpp_session.interface.remote_ip6
714 test.assert_equal(ip.hlim, 255, "IPv6 hop limit")
717 local_ip = test.vpp_session.interface.local_ip4
718 remote_ip = test.vpp_session.interface.remote_ip4
719 test.assert_equal(ip.ttl, 255, "IPv4 TTL")
720 test.assert_equal(ip.src, local_ip, "IP source address")
721 test.assert_equal(ip.dst, remote_ip, "IP destination address")
724 def verify_udp(test, packet):
725 """Verify correctness of UDP layer."""
727 test.assert_equal(udp.dport, BFD.udp_dport, "UDP destination port")
728 test.assert_in_range(
729 udp.sport, BFD.udp_sport_min, BFD.udp_sport_max, "UDP source port"
733 def verify_event(test, event, expected_state):
734 """Verify correctness of event values."""
736 test.logger.debug("BFD: Event: %s" % reprlib.repr(e))
738 e.sw_if_index, test.vpp_session.interface.sw_if_index, "BFD interface index"
742 str(e.local_addr), test.vpp_session.local_addr, "Local IPv6 address"
744 test.assert_equal(str(e.peer_addr), test.vpp_session.peer_addr, "Peer IPv6 address")
745 test.assert_equal(e.state, expected_state, BFDState)
748 def wait_for_bfd_packet(test, timeout=1, pcap_time_min=None, is_tunnel=False):
749 """wait for BFD packet and verify its correctness
751 :param timeout: how long to wait
752 :param pcap_time_min: ignore packets with pcap timestamp lower than this
754 :returns: tuple (packet, time spent waiting for packet)
756 test.logger.info("BFD: Waiting for BFD packet")
757 deadline = time.time() + timeout
762 test.assert_in_range(counter, 0, 100, "number of packets ignored")
763 time_left = deadline - time.time()
765 raise CaptureTimeoutError("Packet did not arrive within timeout")
766 p = test.pg0.wait_for_packet(timeout=time_left)
767 test.test_session.rx_packets += 1
768 test.logger.debug(ppp("BFD: Got packet:", p))
769 if pcap_time_min is not None and p.time < pcap_time_min:
772 "BFD: ignoring packet (pcap time %s < "
773 "pcap time min %s):" % (p.time, pcap_time_min),
779 test.logger.debug(test.vapi.ppcli("show trace"))
781 # strip an IP layer and move to the next
786 raise Exception(ppp("Unexpected or invalid BFD packet:", p))
788 raise Exception(ppp("Unexpected payload in BFD packet:", bfd))
791 test.test_session.verify_bfd(p)
795 BFDStats = namedtuple("BFDStats", "rx rx_echo tx tx_echo")
798 def bfd_grab_stats_snapshot(test, bs_idx=0, thread_index=None):
802 rx = s["/bfd/rx-session-counters"][:, bs_idx].sum_packets()
803 rx_echo = s["/bfd/rx-session-echo-counters"][:, bs_idx].sum_packets()
804 tx = s["/bfd/tx-session-counters"][:, bs_idx].sum_packets()
805 tx_echo = s["/bfd/tx-session-echo-counters"][:, bs_idx].sum_packets()
807 rx = s["/bfd/rx-session-counters"][ti, bs_idx].sum_packets()
808 rx_echo = s["/bfd/rx-session-echo-counters"][ti, bs_idx].sum_packets()
809 tx = s["/bfd/tx-session-counters"][ti, bs_idx].sum_packets()
810 tx_echo = s["/bfd/tx-session-echo-counters"][ti, bs_idx].sum_packets()
811 return BFDStats(rx, rx_echo, tx, tx_echo)
814 def bfd_stats_diff(stats_before, stats_after):
815 rx = stats_after.rx - stats_before.rx
816 rx_echo = stats_after.rx_echo - stats_before.rx_echo
817 tx = stats_after.tx - stats_before.tx
818 tx_echo = stats_after.tx_echo - stats_before.tx_echo
819 return BFDStats(rx, rx_echo, tx, tx_echo)
824 class BFD4TestCase(VppTestCase):
825 """Bidirectional Forwarding Detection (BFD)"""
828 vpp_clock_offset = None
834 super(BFD4TestCase, cls).setUpClass()
835 cls.vapi.cli("set log class bfd level debug")
837 cls.create_pg_interfaces([0])
838 cls.create_loopback_interfaces(1)
839 cls.loopback0 = cls.lo_interfaces[0]
840 cls.loopback0.config_ip4()
841 cls.loopback0.admin_up()
843 cls.pg0.configure_ipv4_neighbors()
845 cls.pg0.resolve_arp()
848 super(BFD4TestCase, cls).tearDownClass()
852 def tearDownClass(cls):
853 super(BFD4TestCase, cls).tearDownClass()
856 super(BFD4TestCase, self).setUp()
857 self.factory = AuthKeyFactory()
858 self.vapi.want_bfd_events()
859 self.pg0.enable_capture()
861 self.bfd_udp4_sessions = self.statistics["/bfd/udp4/sessions"]
862 self.bfd_udp6_sessions = self.statistics["/bfd/udp6/sessions"]
863 self.vapi.cli("trace add bfd-process 500")
864 self.vpp_session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
865 self.vpp_session.add_vpp_config()
866 self.vpp_session.admin_up()
867 self.test_session = BFDTestSession(self, self.pg0, AF_INET)
868 except BaseException:
869 self.vapi.want_bfd_events(enable_disable=0)
873 if not self.vpp_dead:
874 self.vapi.want_bfd_events(enable_disable=0)
875 self.vapi.collect_events() # clear the event queue
876 super(BFD4TestCase, self).tearDown()
878 def test_session_up(self):
879 """bring BFD session up"""
881 bfd_udp4_sessions = self.statistics["/bfd/udp4/sessions"]
882 bfd_udp6_sessions = self.statistics["/bfd/udp6/sessions"]
883 self.assert_equal(bfd_udp4_sessions - self.bfd_udp4_sessions, 1)
884 self.assert_equal(bfd_udp6_sessions, self.bfd_udp6_sessions)
886 def test_session_up_by_ip(self):
887 """bring BFD session up - first frame looked up by address pair"""
888 self.logger.info("BFD: Sending Slow control frame")
889 self.test_session.update(my_discriminator=randint(0, 40000000))
890 self.test_session.send_packet()
891 self.pg0.enable_capture()
892 p = self.pg0.wait_for_packet(1)
894 p[BFD].your_discriminator,
895 self.test_session.my_discriminator,
896 "BFD - your discriminator",
898 self.assert_equal(p[BFD].state, BFDState.init, BFDState)
899 self.test_session.update(
900 your_discriminator=p[BFD].my_discriminator, state=BFDState.up
902 self.logger.info("BFD: Waiting for event")
903 e = self.vapi.wait_for_event(1, "bfd_udp_session_event")
904 verify_event(self, e, expected_state=BFDState.init)
905 self.logger.info("BFD: Sending Up")
906 self.test_session.send_packet()
907 self.logger.info("BFD: Waiting for event")
908 e = self.vapi.wait_for_event(1, "bfd_udp_session_event")
909 verify_event(self, e, expected_state=BFDState.up)
910 self.logger.info("BFD: Session is Up")
911 self.test_session.update(state=BFDState.up)
912 self.test_session.send_packet()
913 self.assert_equal(self.vpp_session.state, BFDState.up, BFDState)
915 def test_session_down(self):
916 """bring BFD session down"""
918 bfd_session_down(self)
920 def test_hold_up(self):
921 """hold BFD session up"""
923 for dummy in range(self.test_session.detect_mult * 2):
924 wait_for_bfd_packet(self)
925 self.test_session.send_packet()
926 self.assert_equal(len(self.vapi.collect_events()), 0, "number of bfd events")
928 def test_slow_timer(self):
929 """verify slow periodic control frames while session down"""
931 self.logger.info("BFD: Waiting for %d BFD packets", packet_count)
932 prev_packet = wait_for_bfd_packet(self, 2)
933 for dummy in range(packet_count):
934 next_packet = wait_for_bfd_packet(self, 2)
935 time_diff = next_packet.time - prev_packet.time
936 # spec says the range should be <0.75, 1>, allow extra 0.05 margin
937 # to work around timing issues
938 self.assert_in_range(time_diff, 0.70, 1.05, "time between slow packets")
939 prev_packet = next_packet
941 def test_zero_remote_min_rx(self):
942 """no packets when zero remote required min rx interval"""
944 self.test_session.update(required_min_rx=0)
945 self.test_session.send_packet()
946 for dummy in range(self.test_session.detect_mult):
948 self.vpp_session.required_min_rx / USEC_IN_SEC,
949 "sleep before transmitting bfd packet",
951 self.test_session.send_packet()
953 p = wait_for_bfd_packet(self, timeout=0)
954 self.logger.error(ppp("Received unexpected packet:", p))
955 except CaptureTimeoutError:
957 self.assert_equal(len(self.vapi.collect_events()), 0, "number of bfd events")
958 self.test_session.update(required_min_rx=300000)
959 for dummy in range(3):
960 self.test_session.send_packet()
962 self, timeout=self.test_session.required_min_rx / USEC_IN_SEC
964 self.assert_equal(len(self.vapi.collect_events()), 0, "number of bfd events")
966 def test_conn_down(self):
967 """verify session goes down after inactivity"""
970 self.test_session.detect_mult
971 * self.vpp_session.required_min_rx
974 self.sleep(detection_time, "waiting for BFD session time-out")
975 e = self.vapi.wait_for_event(1, "bfd_udp_session_event")
976 verify_event(self, e, expected_state=BFDState.down)
978 def test_peer_discr_reset_sess_down(self):
979 """peer discriminator reset after session goes down"""
982 self.test_session.detect_mult
983 * self.vpp_session.required_min_rx
986 self.sleep(detection_time, "waiting for BFD session time-out")
987 self.test_session.my_discriminator = 0
988 wait_for_bfd_packet(self, pcap_time_min=time.time() - self.vpp_clock_offset)
990 def test_large_required_min_rx(self):
991 """large remote required min rx interval"""
993 p = wait_for_bfd_packet(self)
995 self.test_session.update(required_min_rx=interval)
996 self.test_session.send_packet()
997 time_mark = time.time()
999 # busy wait here, trying to collect a packet or event, vpp is not
1000 # allowed to send packets and the session will timeout first - so the
1001 # Up->Down event must arrive before any packets do
1002 while time.time() < time_mark + interval / USEC_IN_SEC:
1004 p = wait_for_bfd_packet(self, timeout=0)
1005 # if vpp managed to send a packet before we did the session
1006 # session update, then that's fine, ignore it
1007 if p.time < time_mark - self.vpp_clock_offset:
1009 self.logger.error(ppp("Received unexpected packet:", p))
1011 except CaptureTimeoutError:
1013 events = self.vapi.collect_events()
1015 verify_event(self, events[0], BFDState.down)
1017 self.assert_equal(count, 0, "number of packets received")
1019 def test_immediate_remote_min_rx_reduction(self):
1020 """immediately honor remote required min rx reduction"""
1021 self.vpp_session.remove_vpp_config()
1022 self.vpp_session = VppBFDUDPSession(
1023 self, self.pg0, self.pg0.remote_ip4, desired_min_tx=10000
1025 self.pg0.enable_capture()
1026 self.vpp_session.add_vpp_config()
1027 self.test_session.update(desired_min_tx=1000000, required_min_rx=1000000)
1028 bfd_session_up(self)
1029 reference_packet = wait_for_bfd_packet(self)
1030 time_mark = time.time()
1032 self.test_session.update(required_min_rx=interval)
1033 self.test_session.send_packet()
1034 extra_time = time.time() - time_mark
1035 p = wait_for_bfd_packet(self)
1036 # first packet is allowed to be late by time we spent doing the update
1037 # calculated in extra_time
1038 self.assert_in_range(
1039 p.time - reference_packet.time,
1040 0.95 * 0.75 * interval / USEC_IN_SEC,
1041 1.05 * interval / USEC_IN_SEC + extra_time,
1042 "time between BFD packets",
1044 reference_packet = p
1045 for dummy in range(3):
1046 p = wait_for_bfd_packet(self)
1047 diff = p.time - reference_packet.time
1048 self.assert_in_range(
1050 0.95 * 0.75 * interval / USEC_IN_SEC,
1051 1.05 * interval / USEC_IN_SEC,
1052 "time between BFD packets",
1054 reference_packet = p
1056 def test_modify_req_min_rx_double(self):
1057 """modify session - double required min rx"""
1058 bfd_session_up(self)
1059 p = wait_for_bfd_packet(self)
1060 self.test_session.update(desired_min_tx=10000, required_min_rx=10000)
1061 self.test_session.send_packet()
1062 # double required min rx
1063 self.vpp_session.modify_parameters(
1064 required_min_rx=2 * self.vpp_session.required_min_rx
1066 p = wait_for_bfd_packet(self, pcap_time_min=time.time() - self.vpp_clock_offset)
1067 # poll bit needs to be set
1068 self.assertIn("P", p.sprintf("%BFD.flags%"), "Poll bit not set in BFD packet")
1069 # finish poll sequence with final packet
1070 final = self.test_session.create_packet()
1071 final[BFD].flags = "F"
1073 self.test_session.detect_mult
1074 * max(self.test_session.desired_min_tx, self.vpp_session.required_min_rx)
1077 self.test_session.send_packet(final)
1078 time_mark = time.time()
1079 e = self.vapi.wait_for_event(2 * timeout, "bfd_udp_session_event")
1080 verify_event(self, e, expected_state=BFDState.down)
1081 time_to_event = time.time() - time_mark
1082 self.assert_in_range(
1083 time_to_event, 0.9 * timeout, 1.1 * timeout, "session timeout"
1086 def test_modify_req_min_rx_halve(self):
1087 """modify session - halve required min rx"""
1088 self.vpp_session.modify_parameters(
1089 required_min_rx=2 * self.vpp_session.required_min_rx
1091 bfd_session_up(self)
1092 p = wait_for_bfd_packet(self)
1093 self.test_session.update(desired_min_tx=10000, required_min_rx=10000)
1094 self.test_session.send_packet()
1095 p = wait_for_bfd_packet(self, pcap_time_min=time.time() - self.vpp_clock_offset)
1096 # halve required min rx
1097 old_required_min_rx = self.vpp_session.required_min_rx
1098 self.vpp_session.modify_parameters(
1099 required_min_rx=self.vpp_session.required_min_rx // 2
1101 # now we wait 0.8*3*old-req-min-rx and the session should still be up
1103 0.8 * self.vpp_session.detect_mult * old_required_min_rx / USEC_IN_SEC,
1104 "wait before finishing poll sequence",
1106 self.assert_equal(len(self.vapi.collect_events()), 0, "number of bfd events")
1107 p = wait_for_bfd_packet(self)
1108 # poll bit needs to be set
1109 self.assertIn("P", p.sprintf("%BFD.flags%"), "Poll bit not set in BFD packet")
1110 # finish poll sequence with final packet
1111 final = self.test_session.create_packet()
1112 final[BFD].flags = "F"
1113 self.test_session.send_packet(final)
1114 # now the session should time out under new conditions
1116 self.test_session.detect_mult
1117 * self.vpp_session.required_min_rx
1120 before = time.time()
1121 e = self.vapi.wait_for_event(2 * detection_time, "bfd_udp_session_event")
1123 self.assert_in_range(
1125 0.9 * detection_time,
1126 1.1 * detection_time,
1127 "time before bfd session goes down",
1129 verify_event(self, e, expected_state=BFDState.down)
1131 def test_modify_detect_mult(self):
1132 """modify detect multiplier"""
1133 bfd_session_up(self)
1134 p = wait_for_bfd_packet(self)
1135 self.vpp_session.modify_parameters(detect_mult=1)
1136 p = wait_for_bfd_packet(self, pcap_time_min=time.time() - self.vpp_clock_offset)
1138 self.vpp_session.detect_mult, p[BFD].detect_mult, "detect mult"
1140 # poll bit must not be set
1142 "P", p.sprintf("%BFD.flags%"), "Poll bit not set in BFD packet"
1144 self.vpp_session.modify_parameters(detect_mult=10)
1145 p = wait_for_bfd_packet(self, pcap_time_min=time.time() - self.vpp_clock_offset)
1147 self.vpp_session.detect_mult, p[BFD].detect_mult, "detect mult"
1149 # poll bit must not be set
1151 "P", p.sprintf("%BFD.flags%"), "Poll bit not set in BFD packet"
1154 def test_queued_poll(self):
1155 """test poll sequence queueing"""
1156 bfd_session_up(self)
1157 p = wait_for_bfd_packet(self)
1158 self.vpp_session.modify_parameters(
1159 required_min_rx=2 * self.vpp_session.required_min_rx
1161 p = wait_for_bfd_packet(self)
1162 poll_sequence_start = time.time()
1163 poll_sequence_length_min = 0.5
1164 send_final_after = time.time() + poll_sequence_length_min
1165 # poll bit needs to be set
1166 self.assertIn("P", p.sprintf("%BFD.flags%"), "Poll bit not set in BFD packet")
1168 p[BFD].required_min_rx_interval,
1169 self.vpp_session.required_min_rx,
1170 "BFD required min rx interval",
1172 self.vpp_session.modify_parameters(
1173 required_min_rx=2 * self.vpp_session.required_min_rx
1175 # 2nd poll sequence should be queued now
1176 # don't send the reply back yet, wait for some time to emulate
1177 # longer round-trip time
1179 while time.time() < send_final_after:
1180 self.test_session.send_packet()
1181 p = wait_for_bfd_packet(self)
1183 len(self.vapi.collect_events()), 0, "number of bfd events"
1186 p[BFD].required_min_rx_interval,
1187 self.vpp_session.required_min_rx,
1188 "BFD required min rx interval",
1191 # poll bit must be set
1193 "P", p.sprintf("%BFD.flags%"), "Poll bit not set in BFD packet"
1195 final = self.test_session.create_packet()
1196 final[BFD].flags = "F"
1197 self.test_session.send_packet(final)
1198 # finish 1st with final
1199 poll_sequence_length = time.time() - poll_sequence_start
1200 # vpp must wait for some time before starting new poll sequence
1201 poll_no_2_started = False
1202 for dummy in range(2 * packet_count):
1203 p = wait_for_bfd_packet(self)
1205 len(self.vapi.collect_events()), 0, "number of bfd events"
1207 if "P" in p.sprintf("%BFD.flags%"):
1208 poll_no_2_started = True
1209 if time.time() < poll_sequence_start + poll_sequence_length:
1210 raise Exception("VPP started 2nd poll sequence too soon")
1211 final = self.test_session.create_packet()
1212 final[BFD].flags = "F"
1213 self.test_session.send_packet(final)
1216 self.test_session.send_packet()
1217 self.assertTrue(poll_no_2_started, "2nd poll sequence not performed")
1218 # finish 2nd with final
1219 final = self.test_session.create_packet()
1220 final[BFD].flags = "F"
1221 self.test_session.send_packet(final)
1222 p = wait_for_bfd_packet(self)
1223 # poll bit must not be set
1224 self.assertNotIn("P", p.sprintf("%BFD.flags%"), "Poll bit set in BFD packet")
1226 # returning inconsistent results requiring retries in per-patch tests
1227 @unittest.skipUnless(config.extended, "part of extended tests")
1228 def test_poll_response(self):
1229 """test correct response to control frame with poll bit set"""
1230 bfd_session_up(self)
1231 poll = self.test_session.create_packet()
1232 poll[BFD].flags = "P"
1233 self.test_session.send_packet(poll)
1234 final = wait_for_bfd_packet(
1235 self, pcap_time_min=time.time() - self.vpp_clock_offset
1237 self.assertIn("F", final.sprintf("%BFD.flags%"))
1239 def test_no_periodic_if_remote_demand(self):
1240 """no periodic frames outside poll sequence if remote demand set"""
1241 bfd_session_up(self)
1242 demand = self.test_session.create_packet()
1243 demand[BFD].flags = "D"
1244 self.test_session.send_packet(demand)
1247 * max(self.vpp_session.required_min_rx, self.test_session.desired_min_tx)
1251 for dummy in range(self.test_session.detect_mult * 2):
1252 self.sleep(transmit_time)
1253 self.test_session.send_packet(demand)
1255 p = wait_for_bfd_packet(self, timeout=0)
1256 self.logger.error(ppp("Received unexpected packet:", p))
1258 except CaptureTimeoutError:
1260 events = self.vapi.collect_events()
1262 self.logger.error("Received unexpected event: %s", e)
1263 self.assert_equal(count, 0, "number of packets received")
1264 self.assert_equal(len(events), 0, "number of events received")
1266 def test_echo_looped_back(self):
1267 """echo packets looped back"""
1268 bfd_session_up(self)
1269 stats_before = bfd_grab_stats_snapshot(self)
1270 self.pg0.enable_capture()
1271 echo_packet_count = 10
1272 # random source port low enough to increment a few times..
1273 udp_sport_tx = randint(1, 50000)
1274 udp_sport_rx = udp_sport_tx
1276 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
1277 / IP(src=self.pg0.remote_ip4, dst=self.pg0.remote_ip4)
1278 / UDP(dport=BFD.udp_dport_echo)
1279 / Raw("this should be looped back")
1281 for dummy in range(echo_packet_count):
1282 self.sleep(0.01, "delay between echo packets")
1283 echo_packet[UDP].sport = udp_sport_tx
1285 self.logger.debug(ppp("Sending packet:", echo_packet))
1286 self.pg0.add_stream(echo_packet)
1288 self.logger.debug(self.vapi.ppcli("show trace"))
1290 bfd_control_packets_rx = 0
1291 while counter < echo_packet_count:
1292 p = self.pg0.wait_for_packet(1)
1293 self.logger.debug(ppp("Got packet:", p))
1295 self.assert_equal(self.pg0.remote_mac, ether.dst, "Destination MAC")
1296 self.assert_equal(self.pg0.local_mac, ether.src, "Source MAC")
1298 self.assert_equal(self.pg0.remote_ip4, ip.dst, "Destination IP")
1300 if udp.dport == BFD.udp_dport:
1301 bfd_control_packets_rx += 1
1303 self.assert_equal(self.pg0.remote_ip4, ip.src, "Source IP")
1304 self.assert_equal(udp.dport, BFD.udp_dport_echo, "UDP destination port")
1305 self.assert_equal(udp.sport, udp_sport_rx, "UDP source port")
1307 # need to compare the hex payload here, otherwise BFD_vpp_echo
1310 scapy.compat.raw(p[UDP].payload),
1311 scapy.compat.raw(echo_packet[UDP].payload),
1312 "Received packet is not the echo packet sent",
1318 "UDP source port (== ECHO packet identifier for test purposes)",
1320 stats_after = bfd_grab_stats_snapshot(self)
1321 diff = bfd_stats_diff(stats_before, stats_after)
1322 self.assertEqual(0, diff.rx, "RX counter bumped but no BFD packets sent")
1324 0, diff.rx_echo, "RX echo counter bumped but no BFD session exists"
1327 0, diff.tx_echo, "TX echo counter bumped but no BFD session exists"
1330 def test_echo(self):
1332 stats_before = bfd_grab_stats_snapshot(self)
1333 bfd_session_up(self)
1334 self.test_session.update(required_min_echo_rx=150000)
1335 self.test_session.send_packet()
1337 self.test_session.detect_mult
1338 * self.vpp_session.required_min_rx
1341 # echo shouldn't work without echo source set
1342 for dummy in range(10):
1343 sleep = self.vpp_session.required_min_rx / USEC_IN_SEC
1344 self.sleep(sleep, "delay before sending bfd packet")
1345 self.test_session.send_packet()
1346 p = wait_for_bfd_packet(self, pcap_time_min=time.time() - self.vpp_clock_offset)
1348 p[BFD].required_min_rx_interval,
1349 self.vpp_session.required_min_rx,
1350 "BFD required min rx interval",
1352 self.test_session.send_packet()
1353 self.vapi.bfd_udp_set_echo_source(sw_if_index=self.loopback0.sw_if_index)
1355 # should be turned on - loopback echo packets
1356 for dummy in range(3):
1357 loop_until = time.time() + 0.75 * detection_time
1358 while time.time() < loop_until:
1359 p = self.pg0.wait_for_packet(1)
1360 self.logger.debug(ppp("Got packet:", p))
1361 if p[UDP].dport == BFD.udp_dport_echo:
1362 self.assert_equal(p[IP].dst, self.pg0.local_ip4, "BFD ECHO dst IP")
1363 self.assertNotEqual(
1365 self.loopback0.local_ip4,
1366 "BFD ECHO src IP equal to loopback IP",
1368 self.logger.debug(ppp("Looping back packet:", p))
1371 self.pg0.remote_mac,
1372 "ECHO packet destination MAC address",
1374 p[Ether].dst = self.pg0.local_mac
1375 self.pg0.add_stream(p)
1376 self.test_session.rx_packets_echo += 1
1377 self.test_session.tx_packets_echo += 1
1380 elif p.haslayer(BFD):
1381 self.test_session.rx_packets += 1
1383 self.assertGreaterEqual(
1384 p[BFD].required_min_rx_interval, 1000000
1386 if "P" in p.sprintf("%BFD.flags%"):
1387 final = self.test_session.create_packet()
1388 final[BFD].flags = "F"
1389 self.test_session.send_packet(final)
1391 raise Exception(ppp("Received unknown packet:", p))
1394 len(self.vapi.collect_events()), 0, "number of bfd events"
1396 self.test_session.send_packet()
1397 self.assertTrue(echo_seen, "No echo packets received")
1399 stats_after = bfd_grab_stats_snapshot(self)
1400 diff = bfd_stats_diff(stats_before, stats_after)
1401 # our rx is vpp tx and vice versa, also tolerate one packet off
1402 self.assert_in_range(
1403 self.test_session.tx_packets, diff.rx - 1, diff.rx + 1, "RX counter"
1405 self.assert_in_range(
1406 self.test_session.rx_packets, diff.tx - 1, diff.tx + 1, "TX counter"
1408 self.assert_in_range(
1409 self.test_session.tx_packets_echo,
1414 self.assert_in_range(
1415 self.test_session.rx_packets_echo,
1421 def test_echo_fail(self):
1422 """session goes down if echo function fails"""
1423 bfd_session_up(self)
1424 self.test_session.update(required_min_echo_rx=150000)
1425 self.test_session.send_packet()
1427 self.test_session.detect_mult
1428 * self.vpp_session.required_min_rx
1431 self.vapi.bfd_udp_set_echo_source(sw_if_index=self.loopback0.sw_if_index)
1432 # echo function should be used now, but we will drop the echo packets
1433 verified_diag = False
1434 for dummy in range(3):
1435 loop_until = time.time() + 0.75 * detection_time
1436 while time.time() < loop_until:
1437 p = self.pg0.wait_for_packet(1)
1438 self.logger.debug(ppp("Got packet:", p))
1439 if p[UDP].dport == BFD.udp_dport_echo:
1442 elif p.haslayer(BFD):
1443 if "P" in p.sprintf("%BFD.flags%"):
1444 self.assertGreaterEqual(
1445 p[BFD].required_min_rx_interval, 1000000
1447 final = self.test_session.create_packet()
1448 final[BFD].flags = "F"
1449 self.test_session.send_packet(final)
1450 if p[BFD].state == BFDState.down:
1452 p[BFD].diag, BFDDiagCode.echo_function_failed, BFDDiagCode
1454 verified_diag = True
1456 raise Exception(ppp("Received unknown packet:", p))
1457 self.test_session.send_packet()
1458 events = self.vapi.collect_events()
1459 self.assert_equal(len(events), 1, "number of bfd events")
1460 self.assert_equal(events[0].state, BFDState.down, BFDState)
1461 self.assertTrue(verified_diag, "Incorrect diagnostics code received")
1463 def test_echo_stop(self):
1464 """echo function stops if peer sets required min echo rx zero"""
1465 bfd_session_up(self)
1466 self.test_session.update(required_min_echo_rx=150000)
1467 self.test_session.send_packet()
1468 self.vapi.bfd_udp_set_echo_source(sw_if_index=self.loopback0.sw_if_index)
1469 # wait for first echo packet
1471 p = self.pg0.wait_for_packet(1)
1472 self.logger.debug(ppp("Got packet:", p))
1473 if p[UDP].dport == BFD.udp_dport_echo:
1474 self.logger.debug(ppp("Looping back packet:", p))
1475 p[Ether].dst = self.pg0.local_mac
1476 self.pg0.add_stream(p)
1479 elif p.haslayer(BFD):
1483 raise Exception(ppp("Received unknown packet:", p))
1484 self.test_session.update(required_min_echo_rx=0)
1485 self.test_session.send_packet()
1486 # echo packets shouldn't arrive anymore
1487 for dummy in range(5):
1488 wait_for_bfd_packet(self, pcap_time_min=time.time() - self.vpp_clock_offset)
1489 self.test_session.send_packet()
1490 events = self.vapi.collect_events()
1491 self.assert_equal(len(events), 0, "number of bfd events")
1493 def test_echo_source_removed(self):
1494 """echo function stops if echo source is removed"""
1495 bfd_session_up(self)
1496 self.test_session.update(required_min_echo_rx=150000)
1497 self.test_session.send_packet()
1498 self.vapi.bfd_udp_set_echo_source(sw_if_index=self.loopback0.sw_if_index)
1499 # wait for first echo packet
1501 p = self.pg0.wait_for_packet(1)
1502 self.logger.debug(ppp("Got packet:", p))
1503 if p[UDP].dport == BFD.udp_dport_echo:
1504 self.logger.debug(ppp("Looping back packet:", p))
1505 p[Ether].dst = self.pg0.local_mac
1506 self.pg0.add_stream(p)
1509 elif p.haslayer(BFD):
1513 raise Exception(ppp("Received unknown packet:", p))
1514 self.vapi.bfd_udp_del_echo_source()
1515 self.test_session.send_packet()
1516 # echo packets shouldn't arrive anymore
1517 for dummy in range(5):
1518 wait_for_bfd_packet(self, pcap_time_min=time.time() - self.vpp_clock_offset)
1519 self.test_session.send_packet()
1520 events = self.vapi.collect_events()
1521 self.assert_equal(len(events), 0, "number of bfd events")
1523 def test_stale_echo(self):
1524 """stale echo packets don't keep a session up"""
1525 bfd_session_up(self)
1526 self.test_session.update(required_min_echo_rx=150000)
1527 self.vapi.bfd_udp_set_echo_source(sw_if_index=self.loopback0.sw_if_index)
1528 self.test_session.send_packet()
1529 # should be turned on - loopback echo packets
1533 for dummy in range(10 * self.vpp_session.detect_mult):
1534 p = self.pg0.wait_for_packet(1)
1535 if p[UDP].dport == BFD.udp_dport_echo:
1536 if echo_packet is None:
1537 self.logger.debug(ppp("Got first echo packet:", p))
1541 + self.vpp_session.detect_mult
1542 * self.test_session.required_min_echo_rx
1546 self.logger.debug(ppp("Got followup echo packet:", p))
1547 self.logger.debug(ppp("Looping back first echo packet:", p))
1548 echo_packet[Ether].dst = self.pg0.local_mac
1549 self.pg0.add_stream(echo_packet)
1551 elif p.haslayer(BFD):
1552 self.logger.debug(ppp("Got packet:", p))
1553 if "P" in p.sprintf("%BFD.flags%"):
1554 final = self.test_session.create_packet()
1555 final[BFD].flags = "F"
1556 self.test_session.send_packet(final)
1557 if p[BFD].state == BFDState.down:
1558 self.assertIsNotNone(
1560 "Session went down before first echo packet received",
1563 self.assertGreaterEqual(
1566 "Session timeout at %s, but is expected at %s"
1567 % (now, timeout_at),
1570 p[BFD].diag, BFDDiagCode.echo_function_failed, BFDDiagCode
1572 events = self.vapi.collect_events()
1573 self.assert_equal(len(events), 1, "number of bfd events")
1574 self.assert_equal(events[0].state, BFDState.down, BFDState)
1578 raise Exception(ppp("Received unknown packet:", p))
1579 self.test_session.send_packet()
1580 self.assertTrue(timeout_ok, "Expected timeout event didn't occur")
1582 def test_invalid_echo_checksum(self):
1583 """echo packets with invalid checksum don't keep a session up"""
1584 bfd_session_up(self)
1585 self.test_session.update(required_min_echo_rx=150000)
1586 self.vapi.bfd_udp_set_echo_source(sw_if_index=self.loopback0.sw_if_index)
1587 self.test_session.send_packet()
1588 # should be turned on - loopback echo packets
1591 for dummy in range(10 * self.vpp_session.detect_mult):
1592 p = self.pg0.wait_for_packet(1)
1593 if p[UDP].dport == BFD.udp_dport_echo:
1594 self.logger.debug(ppp("Got echo packet:", p))
1595 if timeout_at is None:
1598 + self.vpp_session.detect_mult
1599 * self.test_session.required_min_echo_rx
1602 p[BFD_vpp_echo].checksum = getrandbits(64)
1603 p[Ether].dst = self.pg0.local_mac
1604 self.logger.debug(ppp("Looping back modified echo packet:", p))
1605 self.pg0.add_stream(p)
1607 elif p.haslayer(BFD):
1608 self.logger.debug(ppp("Got packet:", p))
1609 if "P" in p.sprintf("%BFD.flags%"):
1610 final = self.test_session.create_packet()
1611 final[BFD].flags = "F"
1612 self.test_session.send_packet(final)
1613 if p[BFD].state == BFDState.down:
1614 self.assertIsNotNone(
1616 "Session went down before first echo packet received",
1619 self.assertGreaterEqual(
1622 "Session timeout at %s, but is expected at %s"
1623 % (now, timeout_at),
1626 p[BFD].diag, BFDDiagCode.echo_function_failed, BFDDiagCode
1628 events = self.vapi.collect_events()
1629 self.assert_equal(len(events), 1, "number of bfd events")
1630 self.assert_equal(events[0].state, BFDState.down, BFDState)
1634 raise Exception(ppp("Received unknown packet:", p))
1635 self.test_session.send_packet()
1636 self.assertTrue(timeout_ok, "Expected timeout event didn't occur")
1638 def test_admin_up_down(self):
1639 """put session admin-up and admin-down"""
1640 bfd_session_up(self)
1641 self.vpp_session.admin_down()
1642 self.pg0.enable_capture()
1643 e = self.vapi.wait_for_event(1, "bfd_udp_session_event")
1644 verify_event(self, e, expected_state=BFDState.admin_down)
1645 for dummy in range(2):
1646 p = wait_for_bfd_packet(self)
1647 self.assert_equal(p[BFD].state, BFDState.admin_down, BFDState)
1648 # try to bring session up - shouldn't be possible
1649 self.test_session.update(state=BFDState.init)
1650 self.test_session.send_packet()
1651 for dummy in range(2):
1652 p = wait_for_bfd_packet(self)
1653 self.assert_equal(p[BFD].state, BFDState.admin_down, BFDState)
1654 self.vpp_session.admin_up()
1655 self.test_session.update(state=BFDState.down)
1656 e = self.vapi.wait_for_event(1, "bfd_udp_session_event")
1657 verify_event(self, e, expected_state=BFDState.down)
1658 p = wait_for_bfd_packet(self, pcap_time_min=time.time() - self.vpp_clock_offset)
1659 self.assert_equal(p[BFD].state, BFDState.down, BFDState)
1660 self.test_session.send_packet()
1661 p = wait_for_bfd_packet(self, pcap_time_min=time.time() - self.vpp_clock_offset)
1662 self.assert_equal(p[BFD].state, BFDState.init, BFDState)
1663 e = self.vapi.wait_for_event(1, "bfd_udp_session_event")
1664 verify_event(self, e, expected_state=BFDState.init)
1665 self.test_session.update(state=BFDState.up)
1666 self.test_session.send_packet()
1667 p = wait_for_bfd_packet(self, pcap_time_min=time.time() - self.vpp_clock_offset)
1668 self.assert_equal(p[BFD].state, BFDState.up, BFDState)
1669 e = self.vapi.wait_for_event(1, "bfd_udp_session_event")
1670 verify_event(self, e, expected_state=BFDState.up)
1672 def test_config_change_remote_demand(self):
1673 """configuration change while peer in demand mode"""
1674 bfd_session_up(self)
1675 demand = self.test_session.create_packet()
1676 demand[BFD].flags = "D"
1677 self.test_session.send_packet(demand)
1678 self.vpp_session.modify_parameters(
1679 required_min_rx=2 * self.vpp_session.required_min_rx
1681 p = wait_for_bfd_packet(self, pcap_time_min=time.time() - self.vpp_clock_offset)
1682 # poll bit must be set
1683 self.assertIn("P", p.sprintf("%BFD.flags%"), "Poll bit not set")
1684 # terminate poll sequence
1685 final = self.test_session.create_packet()
1686 final[BFD].flags = "D+F"
1687 self.test_session.send_packet(final)
1688 # vpp should be quiet now again
1691 * max(self.vpp_session.required_min_rx, self.test_session.desired_min_tx)
1695 for dummy in range(self.test_session.detect_mult * 2):
1696 self.sleep(transmit_time)
1697 self.test_session.send_packet(demand)
1699 p = wait_for_bfd_packet(self, timeout=0)
1700 self.logger.error(ppp("Received unexpected packet:", p))
1702 except CaptureTimeoutError:
1704 events = self.vapi.collect_events()
1706 self.logger.error("Received unexpected event: %s", e)
1707 self.assert_equal(count, 0, "number of packets received")
1708 self.assert_equal(len(events), 0, "number of events received")
1710 def test_intf_deleted(self):
1711 """interface with bfd session deleted"""
1712 intf = VppLoInterface(self)
1715 sw_if_index = intf.sw_if_index
1716 vpp_session = VppBFDUDPSession(self, intf, intf.remote_ip4)
1717 vpp_session.add_vpp_config()
1718 vpp_session.admin_up()
1719 intf.remove_vpp_config()
1720 e = self.vapi.wait_for_event(1, "bfd_udp_session_event")
1721 self.assert_equal(e.sw_if_index, sw_if_index, "sw_if_index")
1722 self.assertFalse(vpp_session.query_vpp_config())
1726 @tag_fixme_vpp_workers
1727 class BFD6TestCase(VppTestCase):
1728 """Bidirectional Forwarding Detection (BFD) (IPv6)"""
1731 vpp_clock_offset = None
1736 def setUpClass(cls):
1737 super(BFD6TestCase, cls).setUpClass()
1738 cls.vapi.cli("set log class bfd level debug")
1740 cls.create_pg_interfaces([0])
1741 cls.pg0.config_ip6()
1742 cls.pg0.configure_ipv6_neighbors()
1744 cls.pg0.resolve_ndp()
1745 cls.create_loopback_interfaces(1)
1746 cls.loopback0 = cls.lo_interfaces[0]
1747 cls.loopback0.config_ip6()
1748 cls.loopback0.admin_up()
1751 super(BFD6TestCase, cls).tearDownClass()
1755 def tearDownClass(cls):
1756 super(BFD6TestCase, cls).tearDownClass()
1759 super(BFD6TestCase, self).setUp()
1760 self.factory = AuthKeyFactory()
1761 self.vapi.want_bfd_events()
1762 self.pg0.enable_capture()
1764 self.bfd_udp4_sessions = self.statistics["/bfd/udp4/sessions"]
1765 self.bfd_udp6_sessions = self.statistics["/bfd/udp6/sessions"]
1766 self.vpp_session = VppBFDUDPSession(
1767 self, self.pg0, self.pg0.remote_ip6, af=AF_INET6
1769 self.vpp_session.add_vpp_config()
1770 self.vpp_session.admin_up()
1771 self.test_session = BFDTestSession(self, self.pg0, AF_INET6)
1772 self.logger.debug(self.vapi.cli("show adj nbr"))
1773 except BaseException:
1774 self.vapi.want_bfd_events(enable_disable=0)
1778 if not self.vpp_dead:
1779 self.vapi.want_bfd_events(enable_disable=0)
1780 self.vapi.collect_events() # clear the event queue
1781 super(BFD6TestCase, self).tearDown()
1783 def test_session_up(self):
1784 """bring BFD session up"""
1785 bfd_session_up(self)
1786 bfd_udp4_sessions = self.statistics["/bfd/udp4/sessions"]
1787 bfd_udp6_sessions = self.statistics["/bfd/udp6/sessions"]
1788 self.assert_equal(bfd_udp4_sessions, self.bfd_udp4_sessions)
1789 self.assert_equal(bfd_udp6_sessions - self.bfd_udp6_sessions, 1)
1791 def test_session_up_by_ip(self):
1792 """bring BFD session up - first frame looked up by address pair"""
1793 self.logger.info("BFD: Sending Slow control frame")
1794 self.test_session.update(my_discriminator=randint(0, 40000000))
1795 self.test_session.send_packet()
1796 self.pg0.enable_capture()
1797 p = self.pg0.wait_for_packet(1)
1799 p[BFD].your_discriminator,
1800 self.test_session.my_discriminator,
1801 "BFD - your discriminator",
1803 self.assert_equal(p[BFD].state, BFDState.init, BFDState)
1804 self.test_session.update(
1805 your_discriminator=p[BFD].my_discriminator, state=BFDState.up
1807 self.logger.info("BFD: Waiting for event")
1808 e = self.vapi.wait_for_event(1, "bfd_udp_session_event")
1809 verify_event(self, e, expected_state=BFDState.init)
1810 self.logger.info("BFD: Sending Up")
1811 self.test_session.send_packet()
1812 self.logger.info("BFD: Waiting for event")
1813 e = self.vapi.wait_for_event(1, "bfd_udp_session_event")
1814 verify_event(self, e, expected_state=BFDState.up)
1815 self.logger.info("BFD: Session is Up")
1816 self.test_session.update(state=BFDState.up)
1817 self.test_session.send_packet()
1818 self.assert_equal(self.vpp_session.state, BFDState.up, BFDState)
1820 def test_hold_up(self):
1821 """hold BFD session up"""
1822 bfd_session_up(self)
1823 for dummy in range(self.test_session.detect_mult * 2):
1824 wait_for_bfd_packet(self)
1825 self.test_session.send_packet()
1826 self.assert_equal(len(self.vapi.collect_events()), 0, "number of bfd events")
1827 self.assert_equal(self.vpp_session.state, BFDState.up, BFDState)
1829 def test_echo_looped_back(self):
1830 """echo packets looped back"""
1831 bfd_session_up(self)
1832 stats_before = bfd_grab_stats_snapshot(self)
1833 self.pg0.enable_capture()
1834 echo_packet_count = 10
1835 # random source port low enough to increment a few times..
1836 udp_sport_tx = randint(1, 50000)
1837 udp_sport_rx = udp_sport_tx
1839 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
1840 / IPv6(src=self.pg0.remote_ip6, dst=self.pg0.remote_ip6)
1841 / UDP(dport=BFD.udp_dport_echo)
1842 / Raw("this should be looped back")
1844 for dummy in range(echo_packet_count):
1845 self.sleep(0.01, "delay between echo packets")
1846 echo_packet[UDP].sport = udp_sport_tx
1848 self.logger.debug(ppp("Sending packet:", echo_packet))
1849 self.pg0.add_stream(echo_packet)
1852 bfd_control_packets_rx = 0
1853 while counter < echo_packet_count:
1854 p = self.pg0.wait_for_packet(1)
1855 self.logger.debug(ppp("Got packet:", p))
1857 self.assert_equal(self.pg0.remote_mac, ether.dst, "Destination MAC")
1858 self.assert_equal(self.pg0.local_mac, ether.src, "Source MAC")
1860 self.assert_equal(self.pg0.remote_ip6, ip.dst, "Destination IP")
1862 if udp.dport == BFD.udp_dport:
1863 bfd_control_packets_rx += 1
1865 self.assert_equal(self.pg0.remote_ip6, ip.src, "Source IP")
1866 self.assert_equal(udp.dport, BFD.udp_dport_echo, "UDP destination port")
1867 self.assert_equal(udp.sport, udp_sport_rx, "UDP source port")
1869 # need to compare the hex payload here, otherwise BFD_vpp_echo
1872 scapy.compat.raw(p[UDP].payload),
1873 scapy.compat.raw(echo_packet[UDP].payload),
1874 "Received packet is not the echo packet sent",
1880 "UDP source port (== ECHO packet identifier for test purposes)",
1882 stats_after = bfd_grab_stats_snapshot(self)
1883 diff = bfd_stats_diff(stats_before, stats_after)
1884 self.assertEqual(0, diff.rx, "RX counter bumped but no BFD packets sent")
1886 0, diff.rx_echo, "RX echo counter bumped but no BFD session exists"
1889 0, diff.tx_echo, "TX echo counter bumped but no BFD session exists"
1892 def test_echo(self):
1894 stats_before = bfd_grab_stats_snapshot(self)
1895 bfd_session_up(self)
1896 self.test_session.update(required_min_echo_rx=150000)
1897 self.test_session.send_packet()
1899 self.test_session.detect_mult
1900 * self.vpp_session.required_min_rx
1903 # echo shouldn't work without echo source set
1904 for dummy in range(10):
1905 sleep = self.vpp_session.required_min_rx / USEC_IN_SEC
1906 self.sleep(sleep, "delay before sending bfd packet")
1907 self.test_session.send_packet()
1908 p = wait_for_bfd_packet(self, pcap_time_min=time.time() - self.vpp_clock_offset)
1910 p[BFD].required_min_rx_interval,
1911 self.vpp_session.required_min_rx,
1912 "BFD required min rx interval",
1914 self.test_session.send_packet()
1915 self.vapi.bfd_udp_set_echo_source(sw_if_index=self.loopback0.sw_if_index)
1917 # should be turned on - loopback echo packets
1918 for dummy in range(3):
1919 loop_until = time.time() + 0.75 * detection_time
1920 while time.time() < loop_until:
1921 p = self.pg0.wait_for_packet(1)
1922 self.logger.debug(ppp("Got packet:", p))
1923 if p[UDP].dport == BFD.udp_dport_echo:
1925 p[IPv6].dst, self.pg0.local_ip6, "BFD ECHO dst IP"
1927 self.assertNotEqual(
1929 self.loopback0.local_ip6,
1930 "BFD ECHO src IP equal to loopback IP",
1932 self.logger.debug(ppp("Looping back packet:", p))
1935 self.pg0.remote_mac,
1936 "ECHO packet destination MAC address",
1938 self.test_session.rx_packets_echo += 1
1939 self.test_session.tx_packets_echo += 1
1940 p[Ether].dst = self.pg0.local_mac
1941 self.pg0.add_stream(p)
1944 elif p.haslayer(BFD):
1945 self.test_session.rx_packets += 1
1947 self.assertGreaterEqual(
1948 p[BFD].required_min_rx_interval, 1000000
1950 if "P" in p.sprintf("%BFD.flags%"):
1951 final = self.test_session.create_packet()
1952 final[BFD].flags = "F"
1953 self.test_session.send_packet(final)
1955 raise Exception(ppp("Received unknown packet:", p))
1958 len(self.vapi.collect_events()), 0, "number of bfd events"
1960 self.test_session.send_packet()
1961 self.assertTrue(echo_seen, "No echo packets received")
1963 stats_after = bfd_grab_stats_snapshot(self)
1964 diff = bfd_stats_diff(stats_before, stats_after)
1965 # our rx is vpp tx and vice versa, also tolerate one packet off
1966 self.assert_in_range(
1967 self.test_session.tx_packets, diff.rx - 1, diff.rx + 1, "RX counter"
1969 self.assert_in_range(
1970 self.test_session.rx_packets, diff.tx - 1, diff.tx + 1, "TX counter"
1972 self.assert_in_range(
1973 self.test_session.tx_packets_echo,
1978 self.assert_in_range(
1979 self.test_session.rx_packets_echo,
1985 def test_intf_deleted(self):
1986 """interface with bfd session deleted"""
1987 intf = VppLoInterface(self)
1990 sw_if_index = intf.sw_if_index
1991 vpp_session = VppBFDUDPSession(self, intf, intf.remote_ip6, af=AF_INET6)
1992 vpp_session.add_vpp_config()
1993 vpp_session.admin_up()
1994 intf.remove_vpp_config()
1995 e = self.vapi.wait_for_event(1, "bfd_udp_session_event")
1996 self.assert_equal(e.sw_if_index, sw_if_index, "sw_if_index")
1997 self.assertFalse(vpp_session.query_vpp_config())
2001 class BFDFIBTestCase(VppTestCase):
2002 """BFD-FIB interactions (IPv6)"""
2008 def setUpClass(cls):
2009 super(BFDFIBTestCase, cls).setUpClass()
2012 def tearDownClass(cls):
2013 super(BFDFIBTestCase, cls).tearDownClass()
2016 super(BFDFIBTestCase, self).setUp()
2017 self.create_pg_interfaces(range(1))
2019 self.vapi.want_bfd_events()
2020 self.pg0.enable_capture()
2022 for i in self.pg_interfaces:
2025 i.configure_ipv6_neighbors()
2028 if not self.vpp_dead:
2029 self.vapi.want_bfd_events(enable_disable=False)
2031 super(BFDFIBTestCase, self).tearDown()
2034 def pkt_is_not_data_traffic(p):
2035 """not data traffic implies BFD or the usual IPv6 ND/RA"""
2036 if p.haslayer(BFD) or is_ipv6_misc(p):
2040 def test_session_with_fib(self):
2041 """BFD-FIB interactions"""
2043 # packets to match against both of the routes
2046 Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
2047 / IPv6(src="3001::1", dst="2001::1")
2048 / UDP(sport=1234, dport=1234)
2049 / Raw(b"\xa5" * 100)
2052 Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
2053 / IPv6(src="3001::1", dst="2002::1")
2054 / UDP(sport=1234, dport=1234)
2055 / Raw(b"\xa5" * 100)
2059 # A recursive and a non-recursive route via a next-hop that
2060 # will have a BFD session
2061 ip_2001_s_64 = VppIpRoute(
2065 [VppRoutePath(self.pg0.remote_ip6, self.pg0.sw_if_index)],
2067 ip_2002_s_64 = VppIpRoute(
2068 self, "2002::", 64, [VppRoutePath(self.pg0.remote_ip6, 0xFFFFFFFF)]
2070 ip_2001_s_64.add_vpp_config()
2071 ip_2002_s_64.add_vpp_config()
2073 # bring the session up now the routes are present
2074 self.vpp_session = VppBFDUDPSession(
2075 self, self.pg0, self.pg0.remote_ip6, af=AF_INET6
2077 self.vpp_session.add_vpp_config()
2078 self.vpp_session.admin_up()
2079 self.test_session = BFDTestSession(self, self.pg0, AF_INET6)
2081 # session is up - traffic passes
2082 bfd_session_up(self)
2084 self.pg0.add_stream(p)
2087 captured = self.pg0.wait_for_packet(
2088 1, filter_out_fn=self.pkt_is_not_data_traffic
2090 self.assertEqual(captured[IPv6].dst, packet[IPv6].dst)
2092 # session is up - traffic is dropped
2093 bfd_session_down(self)
2095 self.pg0.add_stream(p)
2097 with self.assertRaises(CaptureTimeoutError):
2098 self.pg0.wait_for_packet(1, self.pkt_is_not_data_traffic)
2100 # session is up - traffic passes
2101 bfd_session_up(self)
2103 self.pg0.add_stream(p)
2106 captured = self.pg0.wait_for_packet(
2107 1, filter_out_fn=self.pkt_is_not_data_traffic
2109 self.assertEqual(captured[IPv6].dst, packet[IPv6].dst)
2112 @unittest.skipUnless(config.extended, "part of extended tests")
2113 class BFDTunTestCase(VppTestCase):
2114 """BFD over GRE tunnel"""
2120 def setUpClass(cls):
2121 super(BFDTunTestCase, cls).setUpClass()
2124 def tearDownClass(cls):
2125 super(BFDTunTestCase, cls).tearDownClass()
2128 super(BFDTunTestCase, self).setUp()
2129 self.create_pg_interfaces(range(1))
2131 self.vapi.want_bfd_events()
2132 self.pg0.enable_capture()
2134 for i in self.pg_interfaces:
2140 if not self.vpp_dead:
2141 self.vapi.want_bfd_events(enable_disable=0)
2143 super(BFDTunTestCase, self).tearDown()
2146 def pkt_is_not_data_traffic(p):
2147 """not data traffic implies BFD or the usual IPv6 ND/RA"""
2148 if p.haslayer(BFD) or is_ipv6_misc(p):
2152 def test_bfd_o_gre(self):
2155 # A GRE interface over which to run a BFD session
2156 gre_if = VppGreInterface(self, self.pg0.local_ip4, self.pg0.remote_ip4)
2157 gre_if.add_vpp_config()
2161 # bring the session up now the routes are present
2162 self.vpp_session = VppBFDUDPSession(
2163 self, gre_if, gre_if.remote_ip4, is_tunnel=True
2165 self.vpp_session.add_vpp_config()
2166 self.vpp_session.admin_up()
2168 self.test_session = BFDTestSession(
2172 tunnel_header=(IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) / GRE()),
2173 phy_interface=self.pg0,
2176 # packets to match against both of the routes
2179 Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
2180 / IP(src=self.pg0.remote_ip4, dst=gre_if.remote_ip4)
2181 / UDP(sport=1234, dport=1234)
2182 / Raw(b"\xa5" * 100)
2186 # session is up - traffic passes
2187 bfd_session_up(self)
2189 self.send_and_expect(self.pg0, p, self.pg0)
2191 # bring session down
2192 bfd_session_down(self)
2196 class BFDSHA1TestCase(VppTestCase):
2197 """Bidirectional Forwarding Detection (BFD) (SHA1 auth)"""
2200 vpp_clock_offset = None
2205 def setUpClass(cls):
2206 super(BFDSHA1TestCase, cls).setUpClass()
2207 cls.vapi.cli("set log class bfd level debug")
2209 cls.create_pg_interfaces([0])
2210 cls.pg0.config_ip4()
2212 cls.pg0.resolve_arp()
2215 super(BFDSHA1TestCase, cls).tearDownClass()
2219 def tearDownClass(cls):
2220 super(BFDSHA1TestCase, cls).tearDownClass()
2223 super(BFDSHA1TestCase, self).setUp()
2224 self.factory = AuthKeyFactory()
2225 self.vapi.want_bfd_events()
2226 self.pg0.enable_capture()
2229 if not self.vpp_dead:
2230 self.vapi.want_bfd_events(enable_disable=False)
2231 self.vapi.collect_events() # clear the event queue
2232 super(BFDSHA1TestCase, self).tearDown()
2234 def test_session_up(self):
2235 """bring BFD session up"""
2236 key = self.factory.create_random_key(self)
2237 key.add_vpp_config()
2238 self.vpp_session = VppBFDUDPSession(
2239 self, self.pg0, self.pg0.remote_ip4, sha1_key=key
2241 self.vpp_session.add_vpp_config()
2242 self.vpp_session.admin_up()
2243 self.test_session = BFDTestSession(
2248 bfd_key_id=self.vpp_session.bfd_key_id,
2250 bfd_session_up(self)
2252 def test_hold_up(self):
2253 """hold BFD session up"""
2254 key = self.factory.create_random_key(self)
2255 key.add_vpp_config()
2256 self.vpp_session = VppBFDUDPSession(
2257 self, self.pg0, self.pg0.remote_ip4, sha1_key=key
2259 self.vpp_session.add_vpp_config()
2260 self.vpp_session.admin_up()
2261 self.test_session = BFDTestSession(
2266 bfd_key_id=self.vpp_session.bfd_key_id,
2268 bfd_session_up(self)
2269 for dummy in range(self.test_session.detect_mult * 2):
2270 wait_for_bfd_packet(self)
2271 self.test_session.send_packet()
2272 self.assert_equal(self.vpp_session.state, BFDState.up, BFDState)
2274 def test_hold_up_meticulous(self):
2275 """hold BFD session up - meticulous auth"""
2276 key = self.factory.create_random_key(self, BFDAuthType.meticulous_keyed_sha1)
2277 key.add_vpp_config()
2278 self.vpp_session = VppBFDUDPSession(
2279 self, self.pg0, self.pg0.remote_ip4, sha1_key=key
2281 self.vpp_session.add_vpp_config()
2282 self.vpp_session.admin_up()
2283 # specify sequence number so that it wraps
2284 self.test_session = BFDTestSession(
2289 bfd_key_id=self.vpp_session.bfd_key_id,
2290 our_seq_number=0xFFFFFFFF - 4,
2292 bfd_session_up(self)
2293 for dummy in range(30):
2294 wait_for_bfd_packet(self)
2295 self.test_session.inc_seq_num()
2296 self.test_session.send_packet()
2297 self.assert_equal(self.vpp_session.state, BFDState.up, BFDState)
2299 def test_send_bad_seq_number(self):
2300 """session is not kept alive by msgs with bad sequence numbers"""
2301 key = self.factory.create_random_key(self, BFDAuthType.meticulous_keyed_sha1)
2302 key.add_vpp_config()
2303 self.vpp_session = VppBFDUDPSession(
2304 self, self.pg0, self.pg0.remote_ip4, sha1_key=key
2306 self.vpp_session.add_vpp_config()
2307 self.test_session = BFDTestSession(
2312 bfd_key_id=self.vpp_session.bfd_key_id,
2314 bfd_session_up(self)
2316 self.test_session.detect_mult
2317 * self.vpp_session.required_min_rx
2320 send_until = time.time() + 2 * detection_time
2321 while time.time() < send_until:
2322 self.test_session.send_packet()
2324 0.7 * self.vpp_session.required_min_rx / USEC_IN_SEC,
2325 "time between bfd packets",
2327 e = self.vapi.collect_events()
2328 # session should be down now, because the sequence numbers weren't
2330 self.assert_equal(len(e), 1, "number of bfd events")
2331 verify_event(self, e[0], expected_state=BFDState.down)
2333 def execute_rogue_session_scenario(
2335 vpp_bfd_udp_session,
2336 legitimate_test_session,
2338 rogue_bfd_values=None,
2340 """execute a rogue session interaction scenario
2342 1. create vpp session, add config
2343 2. bring the legitimate session up
2344 3. copy the bfd values from legitimate session to rogue session
2345 4. apply rogue_bfd_values to rogue session
2346 5. set rogue session state to down
2347 6. send message to take the session down from the rogue session
2348 7. assert that the legitimate session is unaffected
2351 self.vpp_session = vpp_bfd_udp_session
2352 self.vpp_session.add_vpp_config()
2353 self.test_session = legitimate_test_session
2354 # bring vpp session up
2355 bfd_session_up(self)
2356 # send packet from rogue session
2357 rogue_test_session.update(
2358 my_discriminator=self.test_session.my_discriminator,
2359 your_discriminator=self.test_session.your_discriminator,
2360 desired_min_tx=self.test_session.desired_min_tx,
2361 required_min_rx=self.test_session.required_min_rx,
2362 detect_mult=self.test_session.detect_mult,
2363 diag=self.test_session.diag,
2364 state=self.test_session.state,
2365 auth_type=self.test_session.auth_type,
2367 if rogue_bfd_values:
2368 rogue_test_session.update(**rogue_bfd_values)
2369 rogue_test_session.update(state=BFDState.down)
2370 rogue_test_session.send_packet()
2371 wait_for_bfd_packet(self)
2372 self.assert_equal(self.vpp_session.state, BFDState.up, BFDState)
2374 def test_mismatch_auth(self):
2375 """session is not brought down by unauthenticated msg"""
2376 key = self.factory.create_random_key(self)
2377 key.add_vpp_config()
2378 vpp_session = VppBFDUDPSession(
2379 self, self.pg0, self.pg0.remote_ip4, sha1_key=key
2381 legitimate_test_session = BFDTestSession(
2382 self, self.pg0, AF_INET, sha1_key=key, bfd_key_id=vpp_session.bfd_key_id
2384 rogue_test_session = BFDTestSession(self, self.pg0, AF_INET)
2385 self.execute_rogue_session_scenario(
2386 vpp_session, legitimate_test_session, rogue_test_session
2389 def test_mismatch_bfd_key_id(self):
2390 """session is not brought down by msg with non-existent key-id"""
2391 key = self.factory.create_random_key(self)
2392 key.add_vpp_config()
2393 vpp_session = VppBFDUDPSession(
2394 self, self.pg0, self.pg0.remote_ip4, sha1_key=key
2396 # pick a different random bfd key id
2398 while x == vpp_session.bfd_key_id:
2400 legitimate_test_session = BFDTestSession(
2401 self, self.pg0, AF_INET, sha1_key=key, bfd_key_id=vpp_session.bfd_key_id
2403 rogue_test_session = BFDTestSession(
2404 self, self.pg0, AF_INET, sha1_key=key, bfd_key_id=x
2406 self.execute_rogue_session_scenario(
2407 vpp_session, legitimate_test_session, rogue_test_session
2410 def test_mismatched_auth_type(self):
2411 """session is not brought down by msg with wrong auth type"""
2412 key = self.factory.create_random_key(self)
2413 key.add_vpp_config()
2414 vpp_session = VppBFDUDPSession(
2415 self, self.pg0, self.pg0.remote_ip4, sha1_key=key
2417 legitimate_test_session = BFDTestSession(
2418 self, self.pg0, AF_INET, sha1_key=key, bfd_key_id=vpp_session.bfd_key_id
2420 rogue_test_session = BFDTestSession(
2421 self, self.pg0, AF_INET, sha1_key=key, bfd_key_id=vpp_session.bfd_key_id
2423 self.execute_rogue_session_scenario(
2425 legitimate_test_session,
2427 {"auth_type": BFDAuthType.keyed_md5},
2430 def test_restart(self):
2431 """simulate remote peer restart and resynchronization"""
2432 key = self.factory.create_random_key(self, BFDAuthType.meticulous_keyed_sha1)
2433 key.add_vpp_config()
2434 self.vpp_session = VppBFDUDPSession(
2435 self, self.pg0, self.pg0.remote_ip4, sha1_key=key
2437 self.vpp_session.add_vpp_config()
2438 self.test_session = BFDTestSession(
2443 bfd_key_id=self.vpp_session.bfd_key_id,
2446 bfd_session_up(self)
2447 # don't send any packets for 2*detection_time
2449 self.test_session.detect_mult
2450 * self.vpp_session.required_min_rx
2453 self.sleep(2 * detection_time, "simulating peer restart")
2454 events = self.vapi.collect_events()
2455 self.assert_equal(len(events), 1, "number of bfd events")
2456 verify_event(self, events[0], expected_state=BFDState.down)
2457 self.test_session.update(state=BFDState.down)
2458 # reset sequence number
2459 self.test_session.our_seq_number = 0
2460 self.test_session.vpp_seq_number = None
2461 # now throw away any pending packets
2462 self.pg0.enable_capture()
2463 self.test_session.my_discriminator = 0
2464 bfd_session_up(self)
2468 class BFDAuthOnOffTestCase(VppTestCase):
2469 """Bidirectional Forwarding Detection (BFD) (changing auth)"""
2476 def setUpClass(cls):
2477 super(BFDAuthOnOffTestCase, cls).setUpClass()
2478 cls.vapi.cli("set log class bfd level debug")
2480 cls.create_pg_interfaces([0])
2481 cls.pg0.config_ip4()
2483 cls.pg0.resolve_arp()
2486 super(BFDAuthOnOffTestCase, cls).tearDownClass()
2490 def tearDownClass(cls):
2491 super(BFDAuthOnOffTestCase, cls).tearDownClass()
2494 super(BFDAuthOnOffTestCase, self).setUp()
2495 self.factory = AuthKeyFactory()
2496 self.vapi.want_bfd_events()
2497 self.pg0.enable_capture()
2500 if not self.vpp_dead:
2501 self.vapi.want_bfd_events(enable_disable=False)
2502 self.vapi.collect_events() # clear the event queue
2503 super(BFDAuthOnOffTestCase, self).tearDown()
2505 def test_auth_on_immediate(self):
2506 """turn auth on without disturbing session state (immediate)"""
2507 key = self.factory.create_random_key(self)
2508 key.add_vpp_config()
2509 self.vpp_session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
2510 self.vpp_session.add_vpp_config()
2511 self.test_session = BFDTestSession(self, self.pg0, AF_INET)
2512 bfd_session_up(self)
2513 for dummy in range(self.test_session.detect_mult * 2):
2514 p = wait_for_bfd_packet(self)
2515 self.assert_equal(p[BFD].state, BFDState.up, BFDState)
2516 self.test_session.send_packet()
2517 self.vpp_session.activate_auth(key)
2518 self.test_session.bfd_key_id = self.vpp_session.bfd_key_id
2519 self.test_session.sha1_key = key
2520 for dummy in range(self.test_session.detect_mult * 2):
2521 p = wait_for_bfd_packet(self)
2522 self.assert_equal(p[BFD].state, BFDState.up, BFDState)
2523 self.test_session.send_packet()
2524 self.assert_equal(self.vpp_session.state, BFDState.up, BFDState)
2525 self.assert_equal(len(self.vapi.collect_events()), 0, "number of bfd events")
2527 def test_auth_off_immediate(self):
2528 """turn auth off without disturbing session state (immediate)"""
2529 key = self.factory.create_random_key(self)
2530 key.add_vpp_config()
2531 self.vpp_session = VppBFDUDPSession(
2532 self, self.pg0, self.pg0.remote_ip4, sha1_key=key
2534 self.vpp_session.add_vpp_config()
2535 self.test_session = BFDTestSession(
2540 bfd_key_id=self.vpp_session.bfd_key_id,
2542 bfd_session_up(self)
2543 # self.vapi.want_bfd_events(enable_disable=0)
2544 for dummy in range(self.test_session.detect_mult * 2):
2545 p = wait_for_bfd_packet(self)
2546 self.assert_equal(p[BFD].state, BFDState.up, BFDState)
2547 self.test_session.inc_seq_num()
2548 self.test_session.send_packet()
2549 self.vpp_session.deactivate_auth()
2550 self.test_session.bfd_key_id = None
2551 self.test_session.sha1_key = None
2552 for dummy in range(self.test_session.detect_mult * 2):
2553 p = wait_for_bfd_packet(self)
2554 self.assert_equal(p[BFD].state, BFDState.up, BFDState)
2555 self.test_session.inc_seq_num()
2556 self.test_session.send_packet()
2557 self.assert_equal(self.vpp_session.state, BFDState.up, BFDState)
2558 self.assert_equal(len(self.vapi.collect_events()), 0, "number of bfd events")
2560 def test_auth_change_key_immediate(self):
2561 """change auth key without disturbing session state (immediate)"""
2562 key1 = self.factory.create_random_key(self)
2563 key1.add_vpp_config()
2564 key2 = self.factory.create_random_key(self)
2565 key2.add_vpp_config()
2566 self.vpp_session = VppBFDUDPSession(
2567 self, self.pg0, self.pg0.remote_ip4, sha1_key=key1
2569 self.vpp_session.add_vpp_config()
2570 self.test_session = BFDTestSession(
2575 bfd_key_id=self.vpp_session.bfd_key_id,
2577 bfd_session_up(self)
2578 for dummy in range(self.test_session.detect_mult * 2):
2579 p = wait_for_bfd_packet(self)
2580 self.assert_equal(p[BFD].state, BFDState.up, BFDState)
2581 self.test_session.send_packet()
2582 self.vpp_session.activate_auth(key2)
2583 self.test_session.bfd_key_id = self.vpp_session.bfd_key_id
2584 self.test_session.sha1_key = key2
2585 for dummy in range(self.test_session.detect_mult * 2):
2586 p = wait_for_bfd_packet(self)
2587 self.assert_equal(p[BFD].state, BFDState.up, BFDState)
2588 self.test_session.send_packet()
2589 self.assert_equal(self.vpp_session.state, BFDState.up, BFDState)
2590 self.assert_equal(len(self.vapi.collect_events()), 0, "number of bfd events")
2592 def test_auth_on_delayed(self):
2593 """turn auth on without disturbing session state (delayed)"""
2594 key = self.factory.create_random_key(self)
2595 key.add_vpp_config()
2596 self.vpp_session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
2597 self.vpp_session.add_vpp_config()
2598 self.test_session = BFDTestSession(self, self.pg0, AF_INET)
2599 bfd_session_up(self)
2600 for dummy in range(self.test_session.detect_mult * 2):
2601 wait_for_bfd_packet(self)
2602 self.test_session.send_packet()
2603 self.vpp_session.activate_auth(key, delayed=True)
2604 for dummy in range(self.test_session.detect_mult * 2):
2605 p = wait_for_bfd_packet(self)
2606 self.assert_equal(p[BFD].state, BFDState.up, BFDState)
2607 self.test_session.send_packet()
2608 self.test_session.bfd_key_id = self.vpp_session.bfd_key_id
2609 self.test_session.sha1_key = key
2610 self.test_session.send_packet()
2611 for dummy in range(self.test_session.detect_mult * 2):
2612 p = wait_for_bfd_packet(self)
2613 self.assert_equal(p[BFD].state, BFDState.up, BFDState)
2614 self.test_session.send_packet()
2615 self.assert_equal(self.vpp_session.state, BFDState.up, BFDState)
2616 self.assert_equal(len(self.vapi.collect_events()), 0, "number of bfd events")
2618 def test_auth_off_delayed(self):
2619 """turn auth off without disturbing session state (delayed)"""
2620 key = self.factory.create_random_key(self)
2621 key.add_vpp_config()
2622 self.vpp_session = VppBFDUDPSession(
2623 self, self.pg0, self.pg0.remote_ip4, sha1_key=key
2625 self.vpp_session.add_vpp_config()
2626 self.test_session = BFDTestSession(
2631 bfd_key_id=self.vpp_session.bfd_key_id,
2633 bfd_session_up(self)
2634 for dummy in range(self.test_session.detect_mult * 2):
2635 p = wait_for_bfd_packet(self)
2636 self.assert_equal(p[BFD].state, BFDState.up, BFDState)
2637 self.test_session.send_packet()
2638 self.vpp_session.deactivate_auth(delayed=True)
2639 for dummy in range(self.test_session.detect_mult * 2):
2640 p = wait_for_bfd_packet(self)
2641 self.assert_equal(p[BFD].state, BFDState.up, BFDState)
2642 self.test_session.send_packet()
2643 self.test_session.bfd_key_id = None
2644 self.test_session.sha1_key = None
2645 self.test_session.send_packet()
2646 for dummy in range(self.test_session.detect_mult * 2):
2647 p = wait_for_bfd_packet(self)
2648 self.assert_equal(p[BFD].state, BFDState.up, BFDState)
2649 self.test_session.send_packet()
2650 self.assert_equal(self.vpp_session.state, BFDState.up, BFDState)
2651 self.assert_equal(len(self.vapi.collect_events()), 0, "number of bfd events")
2653 def test_auth_change_key_delayed(self):
2654 """change auth key without disturbing session state (delayed)"""
2655 key1 = self.factory.create_random_key(self)
2656 key1.add_vpp_config()
2657 key2 = self.factory.create_random_key(self)
2658 key2.add_vpp_config()
2659 self.vpp_session = VppBFDUDPSession(
2660 self, self.pg0, self.pg0.remote_ip4, sha1_key=key1
2662 self.vpp_session.add_vpp_config()
2663 self.vpp_session.admin_up()
2664 self.test_session = BFDTestSession(
2669 bfd_key_id=self.vpp_session.bfd_key_id,
2671 bfd_session_up(self)
2672 for dummy in range(self.test_session.detect_mult * 2):
2673 p = wait_for_bfd_packet(self)
2674 self.assert_equal(p[BFD].state, BFDState.up, BFDState)
2675 self.test_session.send_packet()
2676 self.vpp_session.activate_auth(key2, delayed=True)
2677 for dummy in range(self.test_session.detect_mult * 2):
2678 p = wait_for_bfd_packet(self)
2679 self.assert_equal(p[BFD].state, BFDState.up, BFDState)
2680 self.test_session.send_packet()
2681 self.test_session.bfd_key_id = self.vpp_session.bfd_key_id
2682 self.test_session.sha1_key = key2
2683 self.test_session.send_packet()
2684 for dummy in range(self.test_session.detect_mult * 2):
2685 p = wait_for_bfd_packet(self)
2686 self.assert_equal(p[BFD].state, BFDState.up, BFDState)
2687 self.test_session.send_packet()
2688 self.assert_equal(self.vpp_session.state, BFDState.up, BFDState)
2689 self.assert_equal(len(self.vapi.collect_events()), 0, "number of bfd events")
2693 class BFDCLITestCase(VppTestCase):
2694 """Bidirectional Forwarding Detection (BFD) (CLI)"""
2699 def setUpClass(cls):
2700 super(BFDCLITestCase, cls).setUpClass()
2701 cls.vapi.cli("set log class bfd level debug")
2703 cls.create_pg_interfaces((0,))
2704 cls.pg0.config_ip4()
2705 cls.pg0.config_ip6()
2706 cls.pg0.resolve_arp()
2707 cls.pg0.resolve_ndp()
2710 super(BFDCLITestCase, cls).tearDownClass()
2714 def tearDownClass(cls):
2715 super(BFDCLITestCase, cls).tearDownClass()
2718 super(BFDCLITestCase, self).setUp()
2719 self.factory = AuthKeyFactory()
2720 self.pg0.enable_capture()
2724 self.vapi.want_bfd_events(enable_disable=False)
2725 except UnexpectedApiReturnValueError:
2726 # some tests aren't subscribed, so this is not an issue
2728 self.vapi.collect_events() # clear the event queue
2729 super(BFDCLITestCase, self).tearDown()
2731 def cli_verify_no_response(self, cli):
2732 """execute a CLI, asserting that the response is empty"""
2733 self.assert_equal(self.vapi.cli(cli), "", "CLI command response")
2735 def cli_verify_response(self, cli, expected):
2736 """execute a CLI, asserting that the response matches expectation"""
2738 reply = self.vapi.cli(cli)
2739 except CliFailedCommandError as cli_error:
2740 reply = str(cli_error)
2741 self.assert_equal(reply.strip(), expected, "CLI command response")
2743 def test_show(self):
2745 k1 = self.factory.create_random_key(self)
2747 k2 = self.factory.create_random_key(
2748 self, auth_type=BFDAuthType.meticulous_keyed_sha1
2751 s1 = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
2753 s2 = VppBFDUDPSession(
2754 self, self.pg0, self.pg0.remote_ip6, af=AF_INET6, sha1_key=k2
2757 self.logger.info(self.vapi.ppcli("show bfd keys"))
2758 self.logger.info(self.vapi.ppcli("show bfd sessions"))
2759 self.logger.info(self.vapi.ppcli("show bfd"))
2761 def test_set_del_sha1_key(self):
2762 """set/delete SHA1 auth key"""
2763 k = self.factory.create_random_key(self)
2764 self.registry.register(k, self.logger)
2765 self.cli_verify_no_response(
2766 "bfd key set conf-key-id %s type keyed-sha1 secret %s"
2769 "".join("{:02x}".format(scapy.compat.orb(c)) for c in k.key),
2772 self.assertTrue(k.query_vpp_config())
2773 self.vpp_session = VppBFDUDPSession(
2774 self, self.pg0, self.pg0.remote_ip4, sha1_key=k
2776 self.vpp_session.add_vpp_config()
2777 self.test_session = BFDTestSession(
2778 self, self.pg0, AF_INET, sha1_key=k, bfd_key_id=self.vpp_session.bfd_key_id
2780 self.vapi.want_bfd_events()
2781 bfd_session_up(self)
2782 bfd_session_down(self)
2783 # try to replace the secret for the key - should fail because the key
2785 k2 = self.factory.create_random_key(self)
2786 self.cli_verify_response(
2787 "bfd key set conf-key-id %s type keyed-sha1 secret %s"
2790 "".join("{:02x}".format(scapy.compat.orb(c)) for c in k2.key),
2792 "bfd key set: `bfd_auth_set_key' API call failed, "
2793 "rv=-103:BFD object in use",
2795 # manipulating the session using old secret should still work
2796 bfd_session_up(self)
2797 bfd_session_down(self)
2798 self.vpp_session.remove_vpp_config()
2799 self.cli_verify_no_response("bfd key del conf-key-id %s" % k.conf_key_id)
2800 self.assertFalse(k.query_vpp_config())
2802 def test_set_del_meticulous_sha1_key(self):
2803 """set/delete meticulous SHA1 auth key"""
2804 k = self.factory.create_random_key(
2805 self, auth_type=BFDAuthType.meticulous_keyed_sha1
2807 self.registry.register(k, self.logger)
2808 self.cli_verify_no_response(
2809 "bfd key set conf-key-id %s type meticulous-keyed-sha1 secret %s"
2812 "".join("{:02x}".format(scapy.compat.orb(c)) for c in k.key),
2815 self.assertTrue(k.query_vpp_config())
2816 self.vpp_session = VppBFDUDPSession(
2817 self, self.pg0, self.pg0.remote_ip6, af=AF_INET6, sha1_key=k
2819 self.vpp_session.add_vpp_config()
2820 self.vpp_session.admin_up()
2821 self.test_session = BFDTestSession(
2822 self, self.pg0, AF_INET6, sha1_key=k, bfd_key_id=self.vpp_session.bfd_key_id
2824 self.vapi.want_bfd_events()
2825 bfd_session_up(self)
2826 bfd_session_down(self)
2827 # try to replace the secret for the key - should fail because the key
2829 k2 = self.factory.create_random_key(self)
2830 self.cli_verify_response(
2831 "bfd key set conf-key-id %s type keyed-sha1 secret %s"
2834 "".join("{:02x}".format(scapy.compat.orb(c)) for c in k2.key),
2836 "bfd key set: `bfd_auth_set_key' API call failed, "
2837 "rv=-103:BFD object in use",
2839 # manipulating the session using old secret should still work
2840 bfd_session_up(self)
2841 bfd_session_down(self)
2842 self.vpp_session.remove_vpp_config()
2843 self.cli_verify_no_response("bfd key del conf-key-id %s" % k.conf_key_id)
2844 self.assertFalse(k.query_vpp_config())
2846 def test_add_mod_del_bfd_udp(self):
2847 """create/modify/delete IPv4 BFD UDP session"""
2848 vpp_session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
2849 self.registry.register(vpp_session, self.logger)
2851 "bfd udp session add interface %s local-addr %s "
2852 "peer-addr %s desired-min-tx %s required-min-rx %s "
2857 self.pg0.remote_ip4,
2858 vpp_session.desired_min_tx,
2859 vpp_session.required_min_rx,
2860 vpp_session.detect_mult,
2863 self.cli_verify_no_response(cli_add_cmd)
2864 # 2nd add should fail
2865 self.cli_verify_response(
2867 "bfd udp session add: `bfd_add_add_session' API call"
2868 " failed, rv=-101:Duplicate BFD object",
2870 verify_bfd_session_config(self, vpp_session)
2871 mod_session = VppBFDUDPSession(
2874 self.pg0.remote_ip4,
2875 required_min_rx=2 * vpp_session.required_min_rx,
2876 desired_min_tx=3 * vpp_session.desired_min_tx,
2877 detect_mult=4 * vpp_session.detect_mult,
2879 self.cli_verify_no_response(
2880 "bfd udp session mod interface %s local-addr %s peer-addr %s "
2881 "desired-min-tx %s required-min-rx %s detect-mult %s"
2885 self.pg0.remote_ip4,
2886 mod_session.desired_min_tx,
2887 mod_session.required_min_rx,
2888 mod_session.detect_mult,
2891 verify_bfd_session_config(self, mod_session)
2893 "bfd udp session del interface %s local-addr %s "
2894 "peer-addr %s" % (self.pg0.name, self.pg0.local_ip4, self.pg0.remote_ip4)
2896 self.cli_verify_no_response(cli_del_cmd)
2897 # 2nd del is expected to fail
2898 self.cli_verify_response(
2900 "bfd udp session del: `bfd_udp_del_session' API call"
2901 " failed, rv=-102:No such BFD object",
2903 self.assertFalse(vpp_session.query_vpp_config())
2905 def test_add_mod_del_bfd_udp6(self):
2906 """create/modify/delete IPv6 BFD UDP session"""
2907 vpp_session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip6, af=AF_INET6)
2908 self.registry.register(vpp_session, self.logger)
2910 "bfd udp session add interface %s local-addr %s "
2911 "peer-addr %s desired-min-tx %s required-min-rx %s "
2916 self.pg0.remote_ip6,
2917 vpp_session.desired_min_tx,
2918 vpp_session.required_min_rx,
2919 vpp_session.detect_mult,
2922 self.cli_verify_no_response(cli_add_cmd)
2923 # 2nd add should fail
2924 self.cli_verify_response(
2926 "bfd udp session add: `bfd_add_add_session' API call"
2927 " failed, rv=-101:Duplicate BFD object",
2929 verify_bfd_session_config(self, vpp_session)
2930 mod_session = VppBFDUDPSession(
2933 self.pg0.remote_ip6,
2935 required_min_rx=2 * vpp_session.required_min_rx,
2936 desired_min_tx=3 * vpp_session.desired_min_tx,
2937 detect_mult=4 * vpp_session.detect_mult,
2939 self.cli_verify_no_response(
2940 "bfd udp session mod interface %s local-addr %s peer-addr %s "
2941 "desired-min-tx %s required-min-rx %s detect-mult %s"
2945 self.pg0.remote_ip6,
2946 mod_session.desired_min_tx,
2947 mod_session.required_min_rx,
2948 mod_session.detect_mult,
2951 verify_bfd_session_config(self, mod_session)
2953 "bfd udp session del interface %s local-addr %s "
2954 "peer-addr %s" % (self.pg0.name, self.pg0.local_ip6, self.pg0.remote_ip6)
2956 self.cli_verify_no_response(cli_del_cmd)
2957 # 2nd del is expected to fail
2958 self.cli_verify_response(
2960 "bfd udp session del: `bfd_udp_del_session' API call"
2961 " failed, rv=-102:No such BFD object",
2963 self.assertFalse(vpp_session.query_vpp_config())
2965 def test_add_mod_del_bfd_udp_auth(self):
2966 """create/modify/delete IPv4 BFD UDP session (authenticated)"""
2967 key = self.factory.create_random_key(self)
2968 key.add_vpp_config()
2969 vpp_session = VppBFDUDPSession(
2970 self, self.pg0, self.pg0.remote_ip4, sha1_key=key
2972 self.registry.register(vpp_session, self.logger)
2974 "bfd udp session add interface %s local-addr %s "
2975 "peer-addr %s desired-min-tx %s required-min-rx %s "
2976 "detect-mult %s conf-key-id %s bfd-key-id %s"
2980 self.pg0.remote_ip4,
2981 vpp_session.desired_min_tx,
2982 vpp_session.required_min_rx,
2983 vpp_session.detect_mult,
2985 vpp_session.bfd_key_id,
2988 self.cli_verify_no_response(cli_add_cmd)
2989 # 2nd add should fail
2990 self.cli_verify_response(
2992 "bfd udp session add: `bfd_add_add_session' API call"
2993 " failed, rv=-101:Duplicate BFD object",
2995 verify_bfd_session_config(self, vpp_session)
2996 mod_session = VppBFDUDPSession(
2999 self.pg0.remote_ip4,
3001 bfd_key_id=vpp_session.bfd_key_id,
3002 required_min_rx=2 * vpp_session.required_min_rx,
3003 desired_min_tx=3 * vpp_session.desired_min_tx,
3004 detect_mult=4 * vpp_session.detect_mult,
3006 self.cli_verify_no_response(
3007 "bfd udp session mod interface %s local-addr %s peer-addr %s "
3008 "desired-min-tx %s required-min-rx %s detect-mult %s"
3012 self.pg0.remote_ip4,
3013 mod_session.desired_min_tx,
3014 mod_session.required_min_rx,
3015 mod_session.detect_mult,
3018 verify_bfd_session_config(self, mod_session)
3020 "bfd udp session del interface %s local-addr %s "
3021 "peer-addr %s" % (self.pg0.name, self.pg0.local_ip4, self.pg0.remote_ip4)
3023 self.cli_verify_no_response(cli_del_cmd)
3024 # 2nd del is expected to fail
3025 self.cli_verify_response(
3027 "bfd udp session del: `bfd_udp_del_session' API call"
3028 " failed, rv=-102:No such BFD object",
3030 self.assertFalse(vpp_session.query_vpp_config())
3032 def test_add_mod_del_bfd_udp6_auth(self):
3033 """create/modify/delete IPv6 BFD UDP session (authenticated)"""
3034 key = self.factory.create_random_key(
3035 self, auth_type=BFDAuthType.meticulous_keyed_sha1
3037 key.add_vpp_config()
3038 vpp_session = VppBFDUDPSession(
3039 self, self.pg0, self.pg0.remote_ip6, af=AF_INET6, sha1_key=key
3041 self.registry.register(vpp_session, self.logger)
3043 "bfd udp session add interface %s local-addr %s "
3044 "peer-addr %s desired-min-tx %s required-min-rx %s "
3045 "detect-mult %s conf-key-id %s bfd-key-id %s"
3049 self.pg0.remote_ip6,
3050 vpp_session.desired_min_tx,
3051 vpp_session.required_min_rx,
3052 vpp_session.detect_mult,
3054 vpp_session.bfd_key_id,
3057 self.cli_verify_no_response(cli_add_cmd)
3058 # 2nd add should fail
3059 self.cli_verify_response(
3061 "bfd udp session add: `bfd_add_add_session' API call"
3062 " failed, rv=-101:Duplicate BFD object",
3064 verify_bfd_session_config(self, vpp_session)
3065 mod_session = VppBFDUDPSession(
3068 self.pg0.remote_ip6,
3071 bfd_key_id=vpp_session.bfd_key_id,
3072 required_min_rx=2 * vpp_session.required_min_rx,
3073 desired_min_tx=3 * vpp_session.desired_min_tx,
3074 detect_mult=4 * vpp_session.detect_mult,
3076 self.cli_verify_no_response(
3077 "bfd udp session mod interface %s local-addr %s peer-addr %s "
3078 "desired-min-tx %s required-min-rx %s detect-mult %s"
3082 self.pg0.remote_ip6,
3083 mod_session.desired_min_tx,
3084 mod_session.required_min_rx,
3085 mod_session.detect_mult,
3088 verify_bfd_session_config(self, mod_session)
3090 "bfd udp session del interface %s local-addr %s "
3091 "peer-addr %s" % (self.pg0.name, self.pg0.local_ip6, self.pg0.remote_ip6)
3093 self.cli_verify_no_response(cli_del_cmd)
3094 # 2nd del is expected to fail
3095 self.cli_verify_response(
3097 "bfd udp session del: `bfd_udp_del_session' API call"
3098 " failed, rv=-102:No such BFD object",
3100 self.assertFalse(vpp_session.query_vpp_config())
3102 def test_auth_on_off(self):
3103 """turn authentication on and off"""
3104 key = self.factory.create_random_key(
3105 self, auth_type=BFDAuthType.meticulous_keyed_sha1
3107 key.add_vpp_config()
3108 session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
3109 auth_session = VppBFDUDPSession(
3110 self, self.pg0, self.pg0.remote_ip4, sha1_key=key
3112 session.add_vpp_config()
3114 "bfd udp session auth activate interface %s local-addr %s "
3115 "peer-addr %s conf-key-id %s bfd-key-id %s"
3119 self.pg0.remote_ip4,
3121 auth_session.bfd_key_id,
3124 self.cli_verify_no_response(cli_activate)
3125 verify_bfd_session_config(self, auth_session)
3126 self.cli_verify_no_response(cli_activate)
3127 verify_bfd_session_config(self, auth_session)
3129 "bfd udp session auth deactivate interface %s local-addr %s "
3130 "peer-addr %s " % (self.pg0.name, self.pg0.local_ip4, self.pg0.remote_ip4)
3132 self.cli_verify_no_response(cli_deactivate)
3133 verify_bfd_session_config(self, session)
3134 self.cli_verify_no_response(cli_deactivate)
3135 verify_bfd_session_config(self, session)
3137 def test_auth_on_off_delayed(self):
3138 """turn authentication on and off (delayed)"""
3139 key = self.factory.create_random_key(
3140 self, auth_type=BFDAuthType.meticulous_keyed_sha1
3142 key.add_vpp_config()
3143 session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
3144 auth_session = VppBFDUDPSession(
3145 self, self.pg0, self.pg0.remote_ip4, sha1_key=key
3147 session.add_vpp_config()
3149 "bfd udp session auth activate interface %s local-addr %s "
3150 "peer-addr %s conf-key-id %s bfd-key-id %s delayed yes"
3154 self.pg0.remote_ip4,
3156 auth_session.bfd_key_id,
3159 self.cli_verify_no_response(cli_activate)
3160 verify_bfd_session_config(self, auth_session)
3161 self.cli_verify_no_response(cli_activate)
3162 verify_bfd_session_config(self, auth_session)
3164 "bfd udp session auth deactivate interface %s local-addr %s "
3165 "peer-addr %s delayed yes"
3166 % (self.pg0.name, self.pg0.local_ip4, self.pg0.remote_ip4)
3168 self.cli_verify_no_response(cli_deactivate)
3169 verify_bfd_session_config(self, session)
3170 self.cli_verify_no_response(cli_deactivate)
3171 verify_bfd_session_config(self, session)
3173 def test_admin_up_down(self):
3174 """put session admin-up and admin-down"""
3175 session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
3176 session.add_vpp_config()
3178 "bfd udp session set-flags admin down interface %s local-addr %s "
3179 "peer-addr %s " % (self.pg0.name, self.pg0.local_ip4, self.pg0.remote_ip4)
3182 "bfd udp session set-flags admin up interface %s local-addr %s "
3183 "peer-addr %s " % (self.pg0.name, self.pg0.local_ip4, self.pg0.remote_ip4)
3185 self.cli_verify_no_response(cli_down)
3186 verify_bfd_session_config(self, session, state=BFDState.admin_down)
3187 self.cli_verify_no_response(cli_up)
3188 verify_bfd_session_config(self, session, state=BFDState.down)
3190 def test_set_del_udp_echo_source(self):
3191 """set/del udp echo source"""
3192 self.create_loopback_interfaces(1)
3193 self.loopback0 = self.lo_interfaces[0]
3194 self.loopback0.admin_up()
3195 self.cli_verify_response("show bfd echo-source", "UDP echo source is not set.")
3196 cli_set = "bfd udp echo-source set interface %s" % self.loopback0.name
3197 self.cli_verify_no_response(cli_set)
3198 self.cli_verify_response(
3199 "show bfd echo-source",
3200 "UDP echo source is: %s\n"
3201 "IPv4 address usable as echo source: none\n"
3202 "IPv6 address usable as echo source: none" % self.loopback0.name,
3204 self.loopback0.config_ip4()
3206 ipaddress.IPv4Address(
3207 int(ipaddress.IPv4Address(self.loopback0.local_ip4)) ^ 1
3210 self.cli_verify_response(
3211 "show bfd echo-source",
3212 "UDP echo source is: %s\n"
3213 "IPv4 address usable as echo source: %s\n"
3214 "IPv6 address usable as echo source: none"
3215 % (self.loopback0.name, echo_ip4),
3218 ipaddress.IPv6Address(
3219 int(ipaddress.IPv6Address(self.loopback0.local_ip6)) ^ 1
3222 self.loopback0.config_ip6()
3223 self.cli_verify_response(
3224 "show bfd echo-source",
3225 "UDP echo source is: %s\n"
3226 "IPv4 address usable as echo source: %s\n"
3227 "IPv6 address usable as echo source: %s"
3228 % (self.loopback0.name, echo_ip4, echo_ip6),
3230 cli_del = "bfd udp echo-source del"
3231 self.cli_verify_no_response(cli_del)
3232 self.cli_verify_response("show bfd echo-source", "UDP echo source is not set.")
3235 if __name__ == "__main__":
3236 unittest.main(testRunner=VppTestRunner)