netbird/management/internals/modules/peers/manager.go

package peers

//go:generate go tool mockgen -package peers -destination=manager_mock.go -source=./manager.go -build_flags=-mod=mod

import (
	"context"
	"fmt"
	"net"
	"time"

	"github.com/hashicorp/go-multierror"
	"github.com/rs/xid"
	log "github.com/sirupsen/logrus"

	nberrors "github.com/netbirdio/netbird/client/errors"
	"github.com/netbirdio/netbird/management/internals/controllers/network_map"
	"github.com/netbirdio/netbird/management/internals/modules/peers/ephemeral"
	"github.com/netbirdio/netbird/management/server/account"
	"github.com/netbirdio/netbird/management/server/activity"
	"github.com/netbirdio/netbird/management/server/integrations/integrated_validator"
	"github.com/netbirdio/netbird/management/server/peer"
	"github.com/netbirdio/netbird/management/server/permissions"
	"github.com/netbirdio/netbird/management/server/permissions/modules"
	"github.com/netbirdio/netbird/management/server/permissions/operations"
	"github.com/netbirdio/netbird/management/server/store"
	"github.com/netbirdio/netbird/management/server/types"
	"github.com/netbirdio/netbird/shared/management/status"
)

type Manager interface {
	GetPeer(ctx context.Context, accountID, userID, peerID string) (*peer.Peer, error)
	GetPeerAccountID(ctx context.Context, peerID string) (string, error)
	GetAllPeers(ctx context.Context, accountID, userID string) ([]*peer.Peer, error)
	GetPeersByGroupIDs(ctx context.Context, accountID string, groupsIDs []string) ([]*peer.Peer, error)
	// DeletePeers removes the given peers along with their group memberships and
	// policies. Every peer that was not deleted is returned in skipped, so the
	// caller can retry it later: with checkConnected, a peer that is still
	// connected or was seen too recently is left in place, and a peer whose
	// deletion failed is skipped with the failure aggregated into err.
	DeletePeers(ctx context.Context, accountID string, peerIDs []string, userID string, checkConnected bool) (skipped []string, err error)
	SetNetworkMapController(networkMapController network_map.Controller)
	SetIntegratedPeerValidator(integratedPeerValidator integrated_validator.IntegratedValidator)
	SetAccountManager(accountManager account.Manager)
	GetPeerID(ctx context.Context, peerKey string) (string, error)
	CreateProxyPeer(ctx context.Context, accountID string, peerKey string, cluster string) error
	// GetPeerByTunnelIP looks up a peer in accountID by its WireGuard tunnel IP.
	// Returns nil with an error when no match exists. No permission check;
	// callers (the proxy's ValidateTunnelPeer RPC) are trusted server components.
	GetPeerByTunnelIP(ctx context.Context, accountID string, ip net.IP) (*peer.Peer, error)
	// GetPeerWithGroups returns the peer and the list of *types.Group it belongs
	// to. Used by the proxy's auth path to authorise a request by the calling
	// peer's group memberships.
	GetPeerWithGroups(ctx context.Context, accountID, peerID string) (*peer.Peer, []*types.Group, error)
}

type managerImpl struct {
	store                   store.Store
	permissionsManager      permissions.Manager
	integratedPeerValidator integrated_validator.IntegratedValidator
	accountManager          account.Manager

	networkMapController network_map.Controller
}

func NewManager(store store.Store, permissionsManager permissions.Manager) Manager {
	return &managerImpl{
		store:              store,
		permissionsManager: permissionsManager,
	}
}

func (m *managerImpl) SetNetworkMapController(networkMapController network_map.Controller) {
	m.networkMapController = networkMapController
}

func (m *managerImpl) SetIntegratedPeerValidator(integratedPeerValidator integrated_validator.IntegratedValidator) {
	m.integratedPeerValidator = integratedPeerValidator
}

func (m *managerImpl) SetAccountManager(accountManager account.Manager) {
	m.accountManager = accountManager
}

func (m *managerImpl) GetPeer(ctx context.Context, accountID, userID, peerID string) (*peer.Peer, error) {
	allowed, ctx, err := m.permissionsManager.ValidateUserPermissions(ctx, accountID, userID, modules.Peers, operations.Read)
	if err != nil {
		return nil, fmt.Errorf("failed to validate user permissions: %w", err)
	}

	if !allowed {
		return nil, status.NewPermissionDeniedError()
	}

	return m.store.GetPeerByID(ctx, store.LockingStrengthNone, accountID, peerID)
}

func (m *managerImpl) GetAllPeers(ctx context.Context, accountID, userID string) ([]*peer.Peer, error) {
	allowed, ctx, err := m.permissionsManager.ValidateUserPermissions(ctx, accountID, userID, modules.Peers, operations.Read)
	if err != nil {
		return nil, fmt.Errorf("failed to validate user permissions: %w", err)
	}

	if !allowed {
		return m.store.GetUserPeers(ctx, store.LockingStrengthNone, accountID, userID)
	}

	return m.store.GetAccountPeers(ctx, store.LockingStrengthNone, accountID, "", "")
}

func (m *managerImpl) GetPeerAccountID(ctx context.Context, peerID string) (string, error) {
	return m.store.GetAccountIDByPeerID(ctx, store.LockingStrengthNone, peerID)
}

func (m *managerImpl) GetPeersByGroupIDs(ctx context.Context, accountID string, groupsIDs []string) ([]*peer.Peer, error) {
	return m.store.GetPeersByGroupIDs(ctx, accountID, groupsIDs)
}

// GetPeerByTunnelIP delegates to the store's indexed lookup.
func (m *managerImpl) GetPeerByTunnelIP(ctx context.Context, accountID string, ip net.IP) (*peer.Peer, error) {
	return m.store.GetPeerByIP(ctx, store.LockingStrengthNone, accountID, ip)
}

// GetPeerWithGroups returns the peer plus its group memberships. Any store
// error returns (nil, nil, err) so callers never receive a valid peer
// alongside a non-nil error.
func (m *managerImpl) GetPeerWithGroups(ctx context.Context, accountID, peerID string) (*peer.Peer, []*types.Group, error) {
	p, err := m.store.GetPeerByID(ctx, store.LockingStrengthNone, accountID, peerID)
	if err != nil {
		return nil, nil, err
	}
	groups, err := m.store.GetPeerGroups(ctx, store.LockingStrengthNone, accountID, peerID)
	if err != nil {
		return nil, nil, err
	}
	return p, groups, nil
}

// deletePeerOutcome is the per-peer result of a DeletePeers pass.
type deletePeerOutcome int

const (
	// peerMissing means the peer no longer exists, so nothing changed.
	peerMissing deletePeerOutcome = iota
	// peerVetoed means the peer cannot be deleted yet: it is still connected or
	// was seen too recently.
	peerVetoed
	// peerDeleted means the peer and its objects were removed.
	peerDeleted
)

func (m *managerImpl) DeletePeers(ctx context.Context, accountID string, peerIDs []string, userID string, checkConnected bool) ([]string, error) {
	settings, err := m.store.GetAccountSettings(ctx, store.LockingStrengthNone, accountID)
	if err != nil {
		return peerIDs, err
	}
	dnsDomain := m.networkMapController.GetDNSDomain(settings)

	var skipped []string
	var merr *multierror.Error
	deletedAny := false
	for _, peerID := range peerIDs {
		outcome, events, err := m.deleteSinglePeer(ctx, accountID, peerID, userID, checkConnected, dnsDomain)
		if err != nil {
			merr = multierror.Append(merr, fmt.Errorf("delete peer %s: %w", peerID, err))
			skipped = append(skipped, peerID)
			continue
		}

		switch outcome {
		case peerVetoed:
			skipped = append(skipped, peerID)
		case peerDeleted:
			deletedAny = true
			m.notifyPeerDeleted(ctx, accountID, peerID, settings.Extra, events)
		case peerMissing:
		}
	}

	// Skipped or missing peers changed nothing, so an update would push an
	// identical map to every peer in the account.
	if deletedAny {
		m.accountManager.UpdateAccountPeers(ctx, accountID, types.UpdateReason{Resource: types.UpdateResourcePeer, Operation: types.UpdateOperationDelete})
	}

	return skipped, nberrors.FormatErrorOrNil(merr)
}

// deleteSinglePeer deletes one peer along with its group memberships and
// policies in a single transaction. With checkConnected, a peer that is still
// connected or was seen too recently is left untouched and reported as vetoed.
// The returned events must be stored by the caller once the deletion is final.
func (m *managerImpl) deleteSinglePeer(ctx context.Context, accountID, peerID, userID string, checkConnected bool, dnsDomain string) (deletePeerOutcome, []func(), error) {
	outcome := peerMissing
	var eventsToStore []func()
	err := m.store.ExecuteInTransaction(ctx, func(transaction store.Store) error {
		outcome = peerMissing
		eventsToStore = nil

		p, err := transaction.GetPeerByID(ctx, store.LockingStrengthNone, accountID, peerID)
		if err != nil {
			if e, ok := status.FromError(err); ok && e.Type() == status.NotFound {
				log.WithContext(ctx).Tracef("DeletePeers: peer %s not found, skipping", peerID)
				return nil
			}
			return err
		}

		if checkConnected && (p.Status.Connected || p.Status.LastSeen.After(time.Now().Add(-(ephemeral.EphemeralLifeTime - 10*time.Second)))) {
			log.WithContext(ctx).Tracef("DeletePeers: peer %s skipped (connected=%t, lastSeen=%s, threshold=%s, ephemeral=%t)",
				peerID, p.Status.Connected,
				p.Status.LastSeen.Format(time.RFC3339),
				time.Now().Add(-(ephemeral.EphemeralLifeTime - 10*time.Second)).Format(time.RFC3339),
				p.Ephemeral)
			outcome = peerVetoed
			return nil
		}

		eventsToStore, err = m.deletePeerObjects(ctx, transaction, accountID, userID, dnsDomain, p)
		if err != nil {
			return err
		}
		outcome = peerDeleted
		return nil
	})
	if err != nil {
		return outcome, nil, err
	}
	return outcome, eventsToStore, nil
}

// deletePeerObjects removes the peer's group memberships, its policies and the
// peer itself within the given transaction, returning the activity events to
// store once the transaction commits.
func (m *managerImpl) deletePeerObjects(ctx context.Context, transaction store.Store, accountID, userID, dnsDomain string, p *peer.Peer) ([]func(), error) {
	if err := transaction.RemovePeerFromAllGroups(ctx, p.ID); err != nil {
		return nil, fmt.Errorf("remove peer %s from groups: %w", p.ID, err)
	}

	var eventsToStore []func()
	peerPolicyRules, err := transaction.GetPolicyRulesByResourceID(ctx, store.LockingStrengthNone, accountID, p.ID)
	if err != nil {
		return nil, err
	}
	for _, rule := range peerPolicyRules {
		policy, err := transaction.GetPolicyByID(ctx, store.LockingStrengthNone, accountID, rule.PolicyID)
		if err != nil {
			return nil, err
		}

		if err := transaction.DeletePolicy(ctx, accountID, rule.PolicyID); err != nil {
			return nil, err
		}

		eventsToStore = append(eventsToStore, func() {
			m.accountManager.StoreEvent(ctx, userID, p.ID, accountID, activity.PolicyRemoved, policy.EventMeta())
		})
	}

	if err := transaction.DeletePeer(ctx, accountID, p.ID); err != nil {
		return nil, err
	}

	log.WithContext(ctx).Debugf("DeletePeers: deleted peer %s", p.ID)

	if !(p.ProxyMeta.Embedded || p.Meta.KernelVersion == "wasm") {
		eventsToStore = append(eventsToStore, func() {
			m.accountManager.StoreEvent(ctx, userID, p.ID, accountID, activity.PeerRemovedByUser, p.EventMeta(dnsDomain))
		})
	}

	return eventsToStore, nil
}

// notifyPeerDeleted reports a completed deletion to the integrated validator and
// stores the deletion's activity events.
func (m *managerImpl) notifyPeerDeleted(ctx context.Context, accountID, peerID string, extraSettings *types.ExtraSettings, events []func()) {
	if m.integratedPeerValidator != nil {
		if err := m.integratedPeerValidator.PeerDeleted(ctx, accountID, peerID, extraSettings); err != nil {
			log.WithContext(ctx).Errorf("failed to delete peer %s from integrated validator: %v", peerID, err)
		}
	}
	for _, event := range events {
		event()
	}
}

func (m *managerImpl) GetPeerID(ctx context.Context, peerKey string) (string, error) {
	return m.store.GetPeerIDByKey(ctx, store.LockingStrengthNone, peerKey)
}

func (m *managerImpl) CreateProxyPeer(ctx context.Context, accountID string, peerKey string, cluster string) error {
	existingPeerID, err := m.store.GetPeerIDByKey(ctx, store.LockingStrengthNone, peerKey)
	if err == nil && existingPeerID != "" {
		// Same pubkey already registered — idempotent.
		return nil
	}

	name := fmt.Sprintf("proxy-%s", xid.New().String())
	newPeer := &peer.Peer{
		Ephemeral: true,
		ProxyMeta: peer.ProxyMeta{
			Cluster:  cluster,
			Embedded: true,
		},
		Name:                        name,
		Key:                         peerKey,
		LoginExpirationEnabled:      false,
		InactivityExpirationEnabled: false,
		Meta: peer.PeerSystemMeta{
			Hostname: name,
			GoOS:     "proxy",
			OS:       "proxy",
		},
	}

	_, _, _, _, err = m.accountManager.AddPeer(ctx, accountID, "", "", newPeer, true)
	if err != nil {
		return fmt.Errorf("failed to create proxy peer: %w", err)
	}

	return nil
}