bfd: use transport local address for BFD

docs/sources/bfd.md

@@ -25,6 +25,8 @@ Supported behavior:
 - peer-group BFD configuration inherited by neighbors;
 - default destination UDP port `3784`;
 - source UDP port selected from the RFC 5881 dynamic range `49152..65535`;
+- source IP selected from the neighbor transport `local-address` when configured,
+  or by the kernel when unset;
 - outgoing BFD packets sent with TTL/Hop Limit `255`;
 - BFD states `DOWN`, `INIT`, `UP`, and `ADMIN_DOWN`;
 - Poll/Final handling in control packets;
@@ -242,6 +244,8 @@ If the BFD session does not come up:
   neighbor address;
 - verify that UDP `3784` is reachable in both directions;
 - verify that the remote system accepts source ports from `49152..65535`;
+- verify that the neighbor transport `local-address`, when configured, exists on
+  the host and is compatible with `bind-interface`;
 - avoid setting a non-default `port` unless the remote peer is known to listen
   there;
 - check JSON peer output for `state.bfd_state.session_state`;

pkg/server/bfd_peer.go

@@ -43,6 +43,7 @@ type bfdPeer struct {
 	logger        *slog.Logger
 	peerAddress   netip.Addr
 	peerPort      int
+	localAddress  netip.Addr
 	bindInterface string
 
 	udpClient *net.UDPConn
@@ -69,6 +70,10 @@ type bfdPeer struct {
 }
 
 func NewBfdPeer(ps peerState, logger *slog.Logger, peerAddress netip.Addr, config oc.BfdConfig, bindInterface string) *bfdPeer {
+	return newBfdPeer(ps, logger, peerAddress, config, netip.Addr{}, bindInterface)
+}
+
+func newBfdPeer(ps peerState, logger *slog.Logger, peerAddress netip.Addr, config oc.BfdConfig, localAddress netip.Addr, bindInterface string) *bfdPeer {
 	peerPort := int(config.Port)
 	if peerPort == 0 {
 		peerPort = BfdServerPort
@@ -79,6 +84,7 @@ func NewBfdPeer(ps peerState, logger *slog.Logger, peerAddress netip.Addr, confi
 		logger:        logger,
 		peerAddress:   peerAddress,
 		peerPort:      peerPort,
+		localAddress:  localAddress,
 		bindInterface: bindInterface,
 
 		myDiscriminator: randomBFDMyDiscriminator(),
@@ -202,11 +208,22 @@ func (p *bfdPeer) remoteUDPAddr() *net.UDPAddr {
 	}
 }
 
-func (p *bfdPeer) startClient() {
-	localAddress := &net.UDPAddr{
+func (p *bfdPeer) localUDPAddr() *net.UDPAddr {
+	addr := &net.UDPAddr{
 		Port: randRange(bfdSourcePortMin, bfdSourcePortMax),
 	}
 
+	if p.localAddress.IsValid() && !p.localAddress.IsUnspecified() {
+		addr.IP = p.localAddress.AsSlice()
+		addr.Zone = p.localAddress.Zone()
+	}
+
+	return addr
+}
+
+func (p *bfdPeer) startClient() {
+	localAddress := p.localUDPAddr()
+
 	remoteAddress := p.remoteUDPAddr()
 
 	var err error

pkg/server/bfd_peer_test.go

@@ -3,6 +3,7 @@ package server
 import (
 	"fmt"
 	"log/slog"
+	"net"
 	"net/netip"
 	"sync/atomic"
 	"testing"
@@ -70,6 +71,86 @@ func Test_BfdPeerRemoteUDPAddrZone(t *testing.T) {
 	assert.Empty(g.remoteUDPAddr().Zone)
 }
 
+func Test_BfdPeerLocalUDPAddr(t *testing.T) {
+	tests := []struct {
+		name         string
+		localAddress netip.Addr
+		expectedIP   string
+		expectedZone string
+	}{
+		{name: "unset"},
+		{name: "unspecified IPv4", localAddress: netip.IPv4Unspecified()},
+		{name: "unspecified IPv6", localAddress: netip.IPv6Unspecified()},
+		{name: "IPv4", localAddress: netip.MustParseAddr("192.0.2.1"), expectedIP: "192.0.2.1"},
+		{name: "IPv6", localAddress: netip.MustParseAddr("2001:db8::1"), expectedIP: "2001:db8::1"},
+		{name: "link-local IPv6", localAddress: netip.MustParseAddr("fe80::1%eth0"), expectedIP: "fe80::1", expectedZone: "eth0"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			addr := (&bfdPeer{localAddress: tt.localAddress}).localUDPAddr()
+
+			if tt.expectedIP == "" {
+				assert.Nil(t, addr.IP)
+			} else {
+				assert.Equal(t, tt.expectedIP, addr.IP.String())
+			}
+
+			assert.Equal(t, tt.expectedZone, addr.Zone)
+			assert.GreaterOrEqual(t, addr.Port, bfdSourcePortMin)
+			assert.LessOrEqual(t, addr.Port, bfdSourcePortMax)
+		})
+	}
+}
+
+func Test_BfdPeerStartClientUsesLocalAddress(t *testing.T) {
+	for _, localAddress := range []netip.Addr{
+		netip.MustParseAddr("127.0.0.1"),
+		netip.MustParseAddr("::1"),
+	} {
+		t.Run(localAddress.String(), func(t *testing.T) {
+			listener, err := net.ListenUDP("udp", &net.UDPAddr{IP: localAddress.AsSlice()})
+			if !assert.NoError(t, err) {
+				return
+			}
+			defer listener.Close()
+
+			peer := &bfdPeer{
+				logger:       slog.Default(),
+				peerAddress:  localAddress,
+				peerPort:     listener.LocalAddr().(*net.UDPAddr).Port,
+				localAddress: localAddress,
+			}
+			peer.startClient()
+			if !assert.NotNil(t, peer.udpClient) {
+				return
+			}
+			defer peer.udpClient.Close()
+
+			actual := peer.udpClient.LocalAddr().(*net.UDPAddr)
+			assert.Equal(t, localAddress.String(), actual.IP.String())
+			assert.GreaterOrEqual(t, actual.Port, bfdSourcePortMin)
+			assert.LessOrEqual(t, actual.Port, bfdSourcePortMax)
+		})
+	}
+}
+
+func Test_NewBfdPeerLocalAddressAndBindInterface(t *testing.T) {
+	localAddress := netip.MustParseAddr("2001:db8::1")
+	p := newBfdPeer(
+		&mockPeerState{},
+		slog.Default(),
+		netip.MustParseAddr("2001:db8::2"),
+		oc.BfdConfig{Enabled: true},
+		localAddress,
+		"eth0",
+	)
+	defer p.Stop()
+
+	assert.Equal(t, localAddress, p.localAddress)
+	assert.Equal(t, "eth0", p.bindInterface)
+}
+
 func Test_BfdPeerStopIdempotent(t *testing.T) {
 	assert := assert.New(t)
 

pkg/server/bfd_server.go

@@ -32,6 +32,7 @@ type bfdEventPeerUpdate struct {
 	isAdd         bool
 	peerAddress   netip.Addr
 	config        oc.BfdConfig
+	localAddress  netip.Addr
 	bindInterface string
 }
 
@@ -117,6 +118,10 @@ func (s *bfdServer) Stop() {
 }
 
 func (s *bfdServer) AddPeer(ctx context.Context, peerAddress netip.Addr, config oc.BfdConfig, bindInterface string) error {
+	return s.addPeer(ctx, peerAddress, config, netip.Addr{}, bindInterface)
+}
+
+func (s *bfdServer) addPeer(ctx context.Context, peerAddress netip.Addr, config oc.BfdConfig, localAddress netip.Addr, bindInterface string) error {
 	if s.stopped.Load() {
 		return errors.New("bfd server stopped")
 	}
@@ -126,7 +131,7 @@ func (s *bfdServer) AddPeer(ctx context.Context, peerAddress netip.Addr, config
 	}
 
 	select {
-	case s.eventPeerUpdate <- &bfdEventPeerUpdate{isAdd: true, peerAddress: peerAddress, config: config, bindInterface: bindInterface}:
+	case s.eventPeerUpdate <- &bfdEventPeerUpdate{isAdd: true, peerAddress: peerAddress, config: config, localAddress: localAddress, bindInterface: bindInterface}:
 		if s.stopped.Load() {
 			return errors.New("bfd server stopped")
 		}
@@ -205,7 +210,7 @@ func (s *bfdServer) loop() {
 			s.config = ev
 		case ev := <-s.eventPeerUpdate:
 			if ev.isAdd {
-				s.addBfdPeer(ev.peerAddress, ev.config, ev.bindInterface)
+				s.addBfdPeer(ev.peerAddress, ev.config, ev.localAddress, ev.bindInterface)
 			} else {
 				s.deleteBfdPeer(ev.peerAddress)
 			}
@@ -292,7 +297,7 @@ func (s *bfdServer) stop() {
 	)
 }
 
-func (s *bfdServer) addBfdPeer(peerAddress netip.Addr, config oc.BfdConfig, bindInterface string) {
+func (s *bfdServer) addBfdPeer(peerAddress netip.Addr, config oc.BfdConfig, localAddress netip.Addr, bindInterface string) {
 	s.peersMutex.RLock()
 	_, ok := s.peers[peerAddress]
 	s.peersMutex.RUnlock()
@@ -306,7 +311,7 @@ func (s *bfdServer) addBfdPeer(peerAddress netip.Addr, config oc.BfdConfig, bind
 		return
 	}
 
-	bfdPeer := NewBfdPeer(s.peerState, s.logger, peerAddress, config, bindInterface)
+	bfdPeer := newBfdPeer(s.peerState, s.logger, peerAddress, config, localAddress, bindInterface)
 	if bfdPeer != nil {
 		s.logger.Info("Insert BFD peer",
 			slog.String("Topic", "bfd"),

pkg/server/bfd_server_test.go

@@ -146,6 +146,82 @@ func addPeer(s *bfdServer, port uint16) error {
 	}, "")
 }
 
+func Test_AddPeerLocalAddressAndBindInterface(t *testing.T) {
+	s := newServer(0)
+	defer s.Stop()
+
+	peerAddress := netip.MustParseAddr("127.0.0.1")
+	localAddress := netip.MustParseAddr("127.0.0.2")
+	err := s.addPeer(context.Background(), peerAddress, oc.BfdConfig{Enabled: true}, localAddress, "lo")
+	assert.NoError(t, err)
+
+	err = eventually(time.Second, func() error {
+		s.peersMutex.RLock()
+		defer s.peersMutex.RUnlock()
+
+		peer := s.peers[peerAddress]
+		if peer == nil {
+			return fmt.Errorf("BFD peer not created")
+		}
+
+		if peer.localAddress != localAddress || peer.bindInterface != "lo" {
+			return fmt.Errorf("unexpected BFD source configuration: address=%s interface=%q", peer.localAddress, peer.bindInterface)
+		}
+
+		return nil
+	})
+	assert.NoError(t, err)
+}
+
+func Test_UpdateBfdPeerLocalAddress(t *testing.T) {
+	bfd := newServer(0)
+	defer bfd.Stop()
+
+	s := &BgpServer{bfdServer: bfd}
+	peerAddress := netip.MustParseAddr("127.0.0.1")
+	oldLocalAddress := netip.MustParseAddr("127.0.0.2")
+	newLocalAddress := netip.MustParseAddr("127.0.0.3")
+	config := oc.BfdConfig{Enabled: true}
+
+	assert.NoError(t, bfd.addPeer(context.Background(), peerAddress, config, oldLocalAddress, "lo"))
+
+	var oldPeer *bfdPeer
+	assert.NoError(t, eventually(time.Second, func() error {
+		bfd.peersMutex.RLock()
+		defer bfd.peersMutex.RUnlock()
+
+		oldPeer = bfd.peers[peerAddress]
+		if oldPeer == nil {
+			return fmt.Errorf("BFD peer not created")
+		}
+
+		return nil
+	}))
+
+	assert.NoError(t, s.updateBfdPeer(peerAddress.String(), config, config, oldLocalAddress, newLocalAddress, "lo", "lo"))
+	assert.NoError(t, eventually(time.Second, func() error {
+		bfd.peersMutex.RLock()
+		defer bfd.peersMutex.RUnlock()
+
+		peer := bfd.peers[peerAddress]
+		if peer == nil || peer == oldPeer || peer.localAddress != newLocalAddress {
+			return fmt.Errorf("BFD peer was not recreated with local address %s", newLocalAddress)
+		}
+
+		return nil
+	}))
+
+	bfd.peersMutex.RLock()
+	unchangedPeer := bfd.peers[peerAddress]
+	bfd.peersMutex.RUnlock()
+
+	assert.NoError(t, s.updateBfdPeer(peerAddress.String(), config, config, newLocalAddress, newLocalAddress, "lo", "lo"))
+
+	bfd.peersMutex.RLock()
+	assert.Same(t, unchangedPeer, bfd.peers[peerAddress])
+	bfd.peersMutex.RUnlock()
+}
+
 func Test_AddDeletePeer(t *testing.T) {
 	assert := assert.New(t)
 
@@ -265,6 +341,7 @@ func Test_BgpAddDeletePeer(t *testing.T) {
 	assert.NoError(err)
 	defer s.Stop()
 
+	localAddress := netip.MustParseAddr("127.0.0.10")
 	nConf1 := &oc.Neighbor{
 		Config: oc.NeighborConfig{
 			NeighborAddress: netip.MustParseAddr("127.0.0.1"),
@@ -281,6 +358,11 @@ func Test_BgpAddDeletePeer(t *testing.T) {
 		Config: oc.PeerGroupConfig{
 			PeerGroupName: "group_on",
 		},
+		Transport: oc.Transport{
+			Config: oc.TransportConfig{
+				LocalAddress: localAddress,
+			},
+		},
 		Bfd: oc.Bfd{
 			Config: oc.BfdConfig{
 				Enabled:                  true,
@@ -318,6 +400,10 @@ func Test_BgpAddDeletePeer(t *testing.T) {
 	})
 	assert.Equal(count, 1)
 
+	s.bfdServer.peersMutex.RLock()
+	assert.Equal(localAddress, s.bfdServer.peers[nConf1.Config.NeighborAddress].localAddress)
+	s.bfdServer.peersMutex.RUnlock()
+
 	// Delete 1 peer
 	err = s.DeletePeer(context.Background(), &api.DeletePeerRequest{
 		Address: "127.0.0.1",

pkg/server/server.go

@@ -372,7 +372,7 @@ func (s *BgpServer) passConnToPeer(conn net.Conn) {
 		// register BFD for the dynamic neighbor too (explicit neighbors do this in addNeighbor): the
 		// BFD config is inherited from the peer group. Without this, BFD never runs for dynamic peers.
 		if s.bfdServer != nil && conf.Bfd.Config.Enabled {
-			if err := s.bfdServer.AddPeer(context.Background(), addr, conf.Bfd.Config, s.bgpConfig.Global.Config.BindToDevice); err != nil {
+			if err := s.bfdServer.addPeer(context.Background(), addr, conf.Bfd.Config, conf.Transport.Config.LocalAddress, s.bgpConfig.Global.Config.BindToDevice); err != nil {
 				s.logger.Warn("failed to add BFD peer for dynamic neighbor",
 					slog.String("Topic", "Peer"),
 					slog.String("Key", addr.String()),
@@ -3536,7 +3536,7 @@ func (s *BgpServer) addNeighbor(c *oc.Neighbor) error {
 		if err != nil {
 			return fmt.Errorf("failed to parse IP address: %v", err)
 		}
-		if err := s.bfdServer.AddPeer(context.Background(), ipAddr, c.Bfd.Config, c.Transport.Config.BindInterface); err != nil {
+		if err := s.bfdServer.addPeer(context.Background(), ipAddr, c.Bfd.Config, c.Transport.Config.LocalAddress, c.Transport.Config.BindInterface); err != nil {
 			s.logger.Warn("failed to add BFD peer",
 				slog.String("Topic", "Peer"),
 				slog.String("Key", addr),
@@ -3565,9 +3565,10 @@ func apiBfdSessionStateToOC(state api.BfdSessionState) oc.BfdSessionState {
 func (s *BgpServer) updateBfdPeer(
 	addr string,
 	oldConfig, newConfig oc.BfdConfig,
+	oldLocalAddress, newLocalAddress netip.Addr,
 	oldBindInterface, newBindInterface string,
 ) error {
-	if s.bfdServer == nil || oldConfig.Equal(&newConfig) && oldBindInterface == newBindInterface {
+	if s.bfdServer == nil || oldConfig.Equal(&newConfig) && oldLocalAddress == newLocalAddress && oldBindInterface == newBindInterface {
 		return nil
 	}
 
@@ -3583,7 +3584,7 @@ func (s *BgpServer) updateBfdPeer(
 	}
 
 	if newConfig.Enabled {
-		if err := s.bfdServer.AddPeer(context.Background(), ipAddr, newConfig, newBindInterface); err != nil {
+		if err := s.bfdServer.addPeer(context.Background(), ipAddr, newConfig, newLocalAddress, newBindInterface); err != nil {
 			return err
 		}
 	}
@@ -3860,6 +3861,11 @@ func (s *BgpServer) updateNeighbor(c *oc.Neighbor) (needsSoftResetIn bool, err e
 		conf.Bfd.Config = c.Bfd.Config
 	}
 
+	if original.Transport.Config.LocalAddress != c.Transport.Config.LocalAddress {
+		peer.fsm.logger.Info("Update BFD local address")
+		bfdConfigChanged = true
+	}
+
 	if original.Transport.Config.BindInterface != c.Transport.Config.BindInterface {
 		peer.fsm.logger.Info("Update BFD interface binding")
 		bfdConfigChanged = true
@@ -3912,6 +3918,7 @@ func (s *BgpServer) updateNeighbor(c *oc.Neighbor) (needsSoftResetIn bool, err e
 			err = s.updateBfdPeer(
 				addr,
 				original.Bfd.Config, c.Bfd.Config,
+				original.Transport.Config.LocalAddress, c.Transport.Config.LocalAddress,
 				original.Transport.Config.BindInterface, c.Transport.Config.BindInterface,
 			)
 		}