Fix possible race when syncing peer

client/firewall/uspfilter/uspfilter.go

@@ -337,6 +337,7 @@ func validateRule(ip net.IP, packetData []byte, rules map[string]Rule, d *decode
 			if rule.dPort != 0 && rule.dPort == uint16(d.udp.DstPort) {
 				return rule.drop, true
 			}
+			return rule.drop, true
 		case layers.LayerTypeICMPv4, layers.LayerTypeICMPv6:
 			return rule.drop, true
 		}

client/internal/networkmonitor/monitor_windows.go

@@ -99,11 +99,6 @@ func routeChanged(nexthop systemops.Nexthop, intf *net.Interface, routes []syste
 		return false
 	}
 
-	if isSoftInterface(nexthop.Intf.Name) {
-		log.Tracef("network monitor: ignoring default route change for soft interface %s", nexthop.Intf.Name)
-		return false
-	}
-
 	unspec := getUnspecifiedPrefix(nexthop.IP)
 	defaultRoutes, foundMatchingRoute := processRoutes(nexthop, intf, routes, unspec)
 
@@ -124,7 +119,7 @@ func getUnspecifiedPrefix(ip netip.Addr) netip.Prefix {
 	return netip.PrefixFrom(netip.IPv4Unspecified(), 0)
 }
 
-func processRoutes(nexthop systemops.Nexthop, nexthopIntf *net.Interface, routes []systemops.Route, unspec netip.Prefix) ([]string, bool) {
+func processRoutes(nexthop systemops.Nexthop, intf *net.Interface, routes []systemops.Route, unspec netip.Prefix) ([]string, bool) {
 	var defaultRoutes []string
 	foundMatchingRoute := false
 
@@ -133,7 +128,7 @@ func processRoutes(nexthop systemops.Nexthop, nexthopIntf *net.Interface, routes
 			routeInfo := formatRouteInfo(r)
 			defaultRoutes = append(defaultRoutes, routeInfo)
 
-			if r.Nexthop == nexthop.IP && compareIntf(r.Interface, nexthopIntf) == 0 {
+			if r.Nexthop == nexthop.IP && compareIntf(r.Interface, intf) == 0 {
 				foundMatchingRoute = true
 				log.Debugf("network monitor: found matching default route: %s", routeInfo)
 			}
@@ -244,11 +239,13 @@ func compareIntf(a, b *net.Interface) int {
 		return -1
 	case b == nil:
 		return 1
+	case isIsatapInterface(a.Name) && isIsatapInterface(b.Name):
+		return 0
 	default:
 		return a.Index - b.Index
 	}
 }
 
-func isSoftInterface(name string) bool {
-	return strings.Contains(strings.ToLower(name), "isatap") || strings.Contains(strings.ToLower(name), "teredo")
+func isIsatapInterface(name string) bool {
+	return strings.HasPrefix(strings.ToLower(name), "isatap")
 }

management/client/grpc.go

@@ -334,11 +334,8 @@ func (c *GrpcClient) login(serverKey wgtypes.Key, req *proto.LoginRequest) (*pro
 			Body:     loginReq,
 		})
 		if err != nil {
-			// retry only on context canceled
-			if s, ok := gstatus.FromError(err); ok && s.Code() == codes.Canceled {
-				return err
-			}
-			return backoff.Permanent(err)
+			log.Printf("Login error: %v", err)
+			return err
 		}
 
 		return nil

management/server/account.go

@@ -1870,7 +1870,7 @@ func (am *DefaultAccountManager) SyncAndMarkPeer(ctx context.Context, peerPubKey
 		return nil, nil, nil, err
 	}
 
-	unlock := am.Store.AcquireAccountReadLock(ctx, accountID)
+	unlock := am.Store.AcquireAccountWriteLock(ctx, accountID)
 	defer unlock()
 
 	account, err := am.Store.GetAccount(ctx, accountID)

management/server/file_store.go

@@ -666,26 +666,6 @@ func (s *FileStore) SaveInstallationID(ctx context.Context, ID string) error {
 	return s.persist(ctx, s.storeFile)
 }
 
-// SavePeer saves the peer in the account
-func (s *FileStore) SavePeer(_ context.Context, accountID string, peer *nbpeer.Peer) error {
-	s.mux.Lock()
-	defer s.mux.Unlock()
-
-	account, err := s.getAccount(accountID)
-	if err != nil {
-		return err
-	}
-
-	newPeer := peer.Copy()
-
-	account.Peers[peer.ID] = newPeer
-
-	s.PeerKeyID2AccountID[peer.Key] = accountID
-	s.PeerID2AccountID[peer.ID] = accountID
-
-	return nil
-}
-
 // SavePeerStatus stores the PeerStatus in memory. It doesn't attempt to persist data to speed up things.
 // PeerStatus will be saved eventually when some other changes occur.
 func (s *FileStore) SavePeerStatus(accountID, peerID string, peerStatus nbpeer.PeerStatus) error {

management/server/peer.go

@@ -7,11 +7,10 @@ import (
 	"strings"
 	"time"
 
+	"github.com/netbirdio/netbird/management/server/posture"
 	"github.com/rs/xid"
 	log "github.com/sirupsen/logrus"
 
-	"github.com/netbirdio/netbird/management/server/posture"
-
 	"github.com/netbirdio/netbird/management/proto"
 	"github.com/netbirdio/netbird/management/server/activity"
 	nbpeer "github.com/netbirdio/netbird/management/server/peer"
@@ -540,7 +539,7 @@ func (am *DefaultAccountManager) SyncPeer(ctx context.Context, sync PeerSync, ac
 
 	peer, updated := updatePeerMeta(peer, sync.Meta, account)
 	if updated {
-		err = am.Store.SavePeer(ctx, account.Id, peer)
+		err = am.Store.SaveAccount(ctx, account)
 		if err != nil {
 			return nil, nil, nil, err
 		}

management/server/sql_store.go

@@ -31,10 +31,8 @@ import (
 )
 
 const (
-	storeSqliteFileName        = "store.db"
-	idQueryCondition           = "id = ?"
-	accountAndIDQueryCondition = "account_id = ? and id = ?"
-	peerNotFoundFMT            = "peer %s not found"
+	storeSqliteFileName = "store.db"
+	idQueryCondition    = "id = ?"
 )
 
 // SqlStore represents an account storage backed by a Sql DB persisted to disk
@@ -273,62 +271,24 @@ func (s *SqlStore) GetInstallationID() string {
 	return installation.InstallationIDValue
 }
 
-func (s *SqlStore) SavePeer(ctx context.Context, accountID string, peer *nbpeer.Peer) error {
-	// To maintain data integrity, we create a copy of the peer's to prevent unintended updates to other fields.
-	peerCopy := peer.Copy()
-	peerCopy.AccountID = accountID
-
-	err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
-		// check if peer exists before saving
-		var peerID string
-		result := tx.Model(&nbpeer.Peer{}).Select("id").Find(&peerID, accountAndIDQueryCondition, accountID, peer.ID)
-		if result.Error != nil {
-			return result.Error
-		}
-
-		if peerID == "" {
-			return status.Errorf(status.NotFound, peerNotFoundFMT, peer.ID)
-		}
-
-		result = tx.Model(&nbpeer.Peer{}).Where(accountAndIDQueryCondition, accountID, peer.ID).Save(peerCopy)
-		if result.Error != nil {
-			return result.Error
-		}
-
-		return nil
-	})
-
-	if err != nil {
-		return err
-	}
-
-	return nil
-}
-
 func (s *SqlStore) SavePeerStatus(accountID, peerID string, peerStatus nbpeer.PeerStatus) error {
 	var peerCopy nbpeer.Peer
 	peerCopy.Status = &peerStatus
 
-	if !peerStatus.Connected {
-		log.WithContext(context.Background()).Tracef("saving peer: %s with false status. PeerCopy %v, Arg: %v, Trace: %s",
-			peerID, peerCopy.Status.Connected, peerStatus.Connected,
-			debug.Stack())
-	}
-
 	fieldsToUpdate := []string{
 		"peer_status_last_seen", "peer_status_connected",
 		"peer_status_login_expired", "peer_status_required_approval",
 	}
 	result := s.db.Model(&nbpeer.Peer{}).
 		Select(fieldsToUpdate).
-		Where(accountAndIDQueryCondition, accountID, peerID).
+		Where("account_id = ? AND id = ?", accountID, peerID).
 		Updates(&peerCopy)
 	if result.Error != nil {
 		return result.Error
 	}
 
 	if result.RowsAffected == 0 {
-		return status.Errorf(status.NotFound, peerNotFoundFMT, peerID)
+		return status.Errorf(status.NotFound, "peer %s not found", peerID)
 	}
 
 	return nil
@@ -342,7 +302,7 @@ func (s *SqlStore) SavePeerLocation(accountID string, peerWithLocation *nbpeer.P
 	peerCopy.Location = peerWithLocation.Location
 
 	result := s.db.Model(&nbpeer.Peer{}).
-		Where(accountAndIDQueryCondition, accountID, peerWithLocation.ID).
+		Where("account_id = ? and id = ?", accountID, peerWithLocation.ID).
 		Updates(peerCopy)
 
 	if result.Error != nil {
@@ -350,7 +310,7 @@ func (s *SqlStore) SavePeerLocation(accountID string, peerWithLocation *nbpeer.P
 	}
 
 	if result.RowsAffected == 0 {
-		return status.Errorf(status.NotFound, peerNotFoundFMT, peerWithLocation.ID)
+		return status.Errorf(status.NotFound, "peer %s not found", peerWithLocation.ID)
 	}
 
 	return nil
@@ -684,7 +644,7 @@ func (s *SqlStore) GetAccountSettings(ctx context.Context, accountID string) (*S
 func (s *SqlStore) SaveUserLastLogin(accountID, userID string, lastLogin time.Time) error {
 	var user User
 
-	result := s.db.First(&user, accountAndIDQueryCondition, accountID, userID)
+	result := s.db.First(&user, "account_id = ? and id = ?", accountID, userID)
 	if result.Error != nil {
 		if errors.Is(result.Error, gorm.ErrRecordNotFound) {
 			return status.Errorf(status.NotFound, "user %s not found", userID)

management/server/sql_store_test.go

@@ -362,54 +362,6 @@ func TestSqlite_GetAccount(t *testing.T) {
 	require.Equal(t, status.NotFound, parsedErr.Type(), "should return not found error")
 }
 
-func TestSqlite_SavePeer(t *testing.T) {
-	if runtime.GOOS == "windows" {
-		t.Skip("The SQLite store is not properly supported by Windows yet")
-	}
-
-	store := newSqliteStoreFromFile(t, "testdata/store.json")
-
-	account, err := store.GetAccount(context.Background(), "bf1c8084-ba50-4ce7-9439-34653001fc3b")
-	require.NoError(t, err)
-
-	// save status of non-existing peer
-	peer := &nbpeer.Peer{
-		Key:      "peerkey",
-		ID:       "testpeer",
-		SetupKey: "peerkeysetupkey",
-		IP:       net.IP{127, 0, 0, 1},
-		Meta:     nbpeer.PeerSystemMeta{Hostname: "testingpeer"},
-		Name:     "peer name",
-		Status:   &nbpeer.PeerStatus{Connected: true, LastSeen: time.Now().UTC()},
-	}
-	ctx := context.Background()
-	err = store.SavePeer(ctx, account.Id, peer)
-	assert.Error(t, err)
-	parsedErr, ok := status.FromError(err)
-	require.True(t, ok)
-	require.Equal(t, status.NotFound, parsedErr.Type(), "should return not found error")
-
-	// save new status of existing peer
-	account.Peers[peer.ID] = peer
-
-	err = store.SaveAccount(context.Background(), account)
-	require.NoError(t, err)
-
-	updatedPeer := peer.Copy()
-	updatedPeer.Status.Connected = false
-	updatedPeer.Meta.Hostname = "updatedpeer"
-
-	err = store.SavePeer(ctx, account.Id, updatedPeer)
-	require.NoError(t, err)
-
-	account, err = store.GetAccount(context.Background(), account.Id)
-	require.NoError(t, err)
-
-	actual := account.Peers[peer.ID]
-	assert.Equal(t, updatedPeer.Status, actual.Status)
-	assert.Equal(t, updatedPeer.Meta, actual.Meta)
-}
-
 func TestSqlite_SavePeerStatus(t *testing.T) {
 	if runtime.GOOS == "windows" {
 		t.Skip("The SQLite store is not properly supported by Windows yet")
@@ -450,19 +402,7 @@ func TestSqlite_SavePeerStatus(t *testing.T) {
 
 	actual := account.Peers["testpeer"].Status
 	assert.Equal(t, newStatus, *actual)
-
-	newStatus.Connected = true
-
-	err = store.SavePeerStatus(account.Id, "testpeer", newStatus)
-	require.NoError(t, err)
-
-	account, err = store.GetAccount(context.Background(), account.Id)
-	require.NoError(t, err)
-
-	actual = account.Peers["testpeer"].Status
-	assert.Equal(t, newStatus, *actual)
 }
-
 func TestSqlite_SavePeerLocation(t *testing.T) {
 	if runtime.GOOS == "windows" {
 		t.Skip("The SQLite store is not properly supported by Windows yet")

management/server/store.go

@@ -12,11 +12,10 @@ import (
 	"strings"
 	"time"
 
+	nbgroup "github.com/netbirdio/netbird/management/server/group"
 	log "github.com/sirupsen/logrus"
 	"gorm.io/gorm"
 
-	nbgroup "github.com/netbirdio/netbird/management/server/group"
-
 	"github.com/netbirdio/netbird/management/server/telemetry"
 	"github.com/netbirdio/netbird/util"
 
@@ -55,7 +54,6 @@ type Store interface {
 	AcquireAccountReadLock(ctx context.Context, accountID string) func()
 	// AcquireGlobalLock should attempt to acquire a global lock and return a function that releases the lock
 	AcquireGlobalLock(ctx context.Context) func()
-	SavePeer(ctx context.Context, accountID string, peer *nbpeer.Peer) error
 	SavePeerStatus(accountID, peerID string, status nbpeer.PeerStatus) error
 	SavePeerLocation(accountID string, peer *nbpeer.Peer) error
 	SaveUserLastLogin(accountID, userID string, lastLogin time.Time) error

signal/cmd/run.go

@@ -90,7 +90,7 @@ var (
 				return err
 			}
 
-			metricsServer, err := metrics.NewServer(metricsPort, "")
+			metricsServer := metrics.NewServer(metricsPort, "")
 			if err != nil {
 				return fmt.Errorf("setup metrics: %v", err)
 			}

signal/metrics/metrics.go

@@ -26,10 +26,10 @@ type Metrics struct {
 }
 
 // NewServer initializes and returns a new Metrics instance
-func NewServer(port int, endpoint string) (*Metrics, error) {
+func NewServer(port int, endpoint string) *Metrics {
 	exporter, err := prometheus.New()
 	if err != nil {
-		return nil, err
+		return nil
 	}
 
 	provider := metric.NewMeterProvider(metric.WithReader(exporter))
@@ -57,7 +57,7 @@ func NewServer(port int, endpoint string) (*Metrics, error) {
 		provider: provider,
 		Endpoint: endpoint,
 		Server:   server,
-	}, nil
+	}
 }
 
 // Shutdown stops the metrics server