diff --git a/internal/controllers/ephemeral_peers/controller.go b/internal/controllers/ephemeral_peers/controller.go index 5b634be..47fb7ae 100644 --- a/internal/controllers/ephemeral_peers/controller.go +++ b/internal/controllers/ephemeral_peers/controller.go @@ -7,7 +7,6 @@ import ( log "github.com/sirupsen/logrus" - nbAccount "github.com/netbirdio/netbird/management/server/account" "github.com/netbirdio/netbird/management/server/activity" nbpeer "github.com/netbirdio/netbird/management/server/peer" "github.com/netbirdio/netbird/management/server/store" diff --git a/internal/modules/accounts/api.go b/internal/modules/accounts/api.go index 710f75d..4acfaac 100644 --- a/internal/modules/accounts/api.go +++ b/internal/modules/accounts/api.go @@ -27,7 +27,7 @@ func newHandler(manager *Manager, permissionsManager permissions.Manager) *handl } } -func (h *handler) RegisterAPI(router *mux.Router) { +func (h *handler) RegisterEndpoints(router *mux.Router) { router.HandleFunc("/accounts/{accountId}", h.updateAccount).Methods("PUT", "OPTIONS") router.HandleFunc("/accounts/{accountId}", h.deleteAccount).Methods("DELETE", "OPTIONS") router.HandleFunc("/accounts", h.getAllAccounts).Methods("GET", "OPTIONS") diff --git a/internal/modules/accounts/settings/api.go b/internal/modules/accounts/settings/api.go index 53077b5..900c663 100644 --- a/internal/modules/accounts/settings/api.go +++ b/internal/modules/accounts/settings/api.go @@ -11,6 +11,8 @@ import ( "management/internal/shared/db" "management/internal/shared/errors" "management/internal/shared/permissions" + "management/internal/shared/permissions/modules" + "management/internal/shared/permissions/operations" ) type handler struct { @@ -25,19 +27,22 @@ func newHandler(manager *Manager, permissionsManager permissions.Manager) *handl } } -func (h *handler) RegisterAPI(router *mux.Router) { - router.HandleFunc("/account/{accountID}/settings", h.GetAllUsers).Methods("GET", "OPTIONS") - router.HandleFunc("/account/{accountID}/settings", h.GetUser).Methods("PUT", "OPTIONS") +func (h *handler) RegisterEndpoints(router *mux.Router) { + router.HandleFunc("/account/{accountID}/settings", h.getSettings).Methods("GET", "OPTIONS") + router.HandleFunc("/account/{accountID}/settings", h.updateSettings).Methods("PUT", "OPTIONS") } -func (h *handler) GetAllUsers(w http.ResponseWriter, r *http.Request) { +func (h *handler) getSettings(w http.ResponseWriter, r *http.Request) { userAuth, err := nbcontext.GetUserAuthFromContext(r.Context()) if err != nil { util.WriteError(r.Context(), err, w) return } - allowed, err := h.permissionsManager.ValidateUserPermissions(r.Context(), userAuth.AccountId, userAuth.UserId, modules.Users, operations.Read) + vars := mux.Vars(r) + accountId := vars["accountID"] + + allowed, err := h.permissionsManager.ValidateUserPermissions(r.Context(), accountId, userAuth.UserId, modules.Settings, operations.Read) if err != nil { util.WriteError(r.Context(), errors.NewPermissionValidationError(err), w) return @@ -46,7 +51,7 @@ func (h *handler) GetAllUsers(w http.ResponseWriter, r *http.Request) { util.WriteError(r.Context(), errors.NewPermissionDeniedError(), w) } - users, err := h.manager.GetAllUsers(r.Context(), nil, db.LockingStrengthShare, userAuth.AccountId) + users, err := h.manager.GetSettings(r.Context(), nil, db.LockingStrengthShare, accountId) if err != nil { http.Error(w, "Internal Server Error", http.StatusInternalServerError) return @@ -54,14 +59,17 @@ func (h *handler) GetAllUsers(w http.ResponseWriter, r *http.Request) { _ = json.NewEncoder(w).Encode(users) } -func (h *handler) GetUser(w http.ResponseWriter, r *http.Request) { +func (h *handler) updateSettings(w http.ResponseWriter, r *http.Request) { userAuth, err := nbcontext.GetUserAuthFromContext(r.Context()) if err != nil { util.WriteError(r.Context(), err, w) return } - allowed, err := h.permissionsManager.ValidateUserPermissions(r.Context(), userAuth.AccountId, userAuth.UserId, modules.Users, operations.Read) + vars := mux.Vars(r) + accountId := vars["accountID"] + + allowed, err := h.permissionsManager.ValidateUserPermissions(r.Context(), accountId, userAuth.UserId, modules.Settings, operations.Write) if err != nil { util.WriteError(r.Context(), errors.NewPermissionValidationError(err), w) return @@ -70,13 +78,10 @@ func (h *handler) GetUser(w http.ResponseWriter, r *http.Request) { util.WriteError(r.Context(), errors.NewPermissionDeniedError(), w) } - vars := mux.Vars(r) - userId := vars["userId"] - - user, err := h.manager.GetUserByID(r.Context(), nil, db.LockingStrengthShare, userId) + settings, err := h.manager.UpdateSettings(r.Context(), nil, db.LockingStrengthShare, accountId) if err != nil { http.Error(w, "Not Found", http.StatusNotFound) return } - _ = json.NewEncoder(w).Encode(user) + _ = json.NewEncoder(w).Encode(settings) } diff --git a/internal/modules/accounts/settings/manager.go b/internal/modules/accounts/settings/manager.go index 1e35d6f..aae981e 100644 --- a/internal/modules/accounts/settings/manager.go +++ b/internal/modules/accounts/settings/manager.go @@ -4,11 +4,15 @@ import ( "context" "fmt" + "github.com/gorilla/mux" "github.com/netbirdio/netbird/management/server/integrations/extra_settings" + types2 "github.com/netbirdio/netbird/management/server/types" "management/internal/modules/accounts/settings/types" "management/internal/modules/users" + "management/internal/shared/activity" "management/internal/shared/db" + "management/internal/shared/permissions" "management/pkg/logging" ) @@ -18,14 +22,20 @@ type Manager struct { repository Repository extraSettingsManager extra_settings.Manager userManager *users.Manager + eventManager *activity.Manager } -func NewManager(store *db.Store, userManager *users.Manager, extraSettingsManager extra_settings.Manager) *Manager { - return &Manager{ - repository: newRepository(store), +func NewManager(store *db.Store, router *mux.Router, eventManager *activity.Manager, permissionsManager permissions.Manager, userManager *users.Manager, extraSettingsManager extra_settings.Manager) *Manager { + repo := newRepository(store) + m := &Manager{ + repository: repo, extraSettingsManager: extraSettingsManager, userManager: userManager, + eventManager: eventManager, } + api := newHandler(m, permissionsManager) + api.RegisterEndpoints(router) + return m } func (m *Manager) GetExtraSettingsManager() extra_settings.Manager { @@ -76,5 +86,9 @@ func (m *Manager) GetExtraSettings(ctx context.Context, tx db.Transaction, accou } func (m *Manager) UpdateExtraSettings(ctx context.Context, accountID, userID string, extraSettings *types.ExtraSettings) (bool, error) { - return m.extraSettingsManager.UpdateExtraSettings(ctx, accountID, userID, extraSettings) + return m.extraSettingsManager.UpdateExtraSettings(ctx, accountID, userID, (*types2.ExtraSettings)(extraSettings)) +} + +func (m *Manager) UpdateSettings(ctx context.Context, tx db.Transaction, settings *types.Settings) (*types.Settings, error) { + return m.repository.UpdateSettings(tx, settings) } diff --git a/internal/modules/accounts/settings/repository.go b/internal/modules/accounts/settings/repository.go index 4323fa8..7d6d033 100644 --- a/internal/modules/accounts/settings/repository.go +++ b/internal/modules/accounts/settings/repository.go @@ -8,6 +8,7 @@ import ( type Repository interface { RunInTx(fn func(tx db.Transaction) error) error GetAccountSettings(tx db.Transaction, strength db.LockingStrength, accountID string) (*types.Settings, error) + UpdateSettings(tx db.Transaction, settings *types.Settings) (*types.Settings, error) } type repository struct { @@ -27,3 +28,11 @@ func (r *repository) GetAccountSettings(tx db.Transaction, strength db.LockingSt err := r.store.GetOne(tx, strength, &settings, "account_id = ?", accountID) return &settings, err } + +func (r *repository) UpdateSettings(tx db.Transaction, settings *types.Settings) (*types.Settings, error) { + err := r.store.Update(tx, settings) + if err != nil { + return nil, err + } + return settings, nil +} diff --git a/internal/modules/groups/api.go b/internal/modules/groups/api.go new file mode 100644 index 0000000..992d699 --- /dev/null +++ b/internal/modules/groups/api.go @@ -0,0 +1,87 @@ +package groups + +import ( + "encoding/json" + "net/http" + + "github.com/gorilla/mux" + nbcontext "github.com/netbirdio/netbird/management/server/context" + "github.com/netbirdio/netbird/management/server/http/util" + + "management/internal/shared/db" + "management/internal/shared/errors" + "management/internal/shared/permissions" + "management/internal/shared/permissions/modules" + "management/internal/shared/permissions/operations" +) + +type handler struct { + manager *Manager + permissionsManager permissions.Manager +} + +func newHandler(manager *Manager, permissionsManager permissions.Manager) *handler { + return &handler{ + manager: manager, + permissionsManager: permissionsManager, + } +} + +func (h *handler) RegisterEndpoints(router *mux.Router) { + router.HandleFunc("/groups", h.getAllGroups).Methods("GET", "OPTIONS") + router.HandleFunc("/groups", h.createGroup).Methods("POST", "OPTIONS") + router.HandleFunc("/groups/{groupId}", h.updateGroup).Methods("PUT", "OPTIONS") + router.HandleFunc("/groups/{groupId}", h.getGroup).Methods("GET", "OPTIONS") + router.HandleFunc("/groups/{groupId}", h.deleteGroup).Methods("DELETE", "OPTIONS") +} + +func (h *handler) getAllUsers(w http.ResponseWriter, r *http.Request) { + userAuth, err := nbcontext.GetUserAuthFromContext(r.Context()) + if err != nil { + util.WriteError(r.Context(), err, w) + return + } + + allowed, err := h.permissionsManager.ValidateUserPermissions(r.Context(), userAuth.AccountId, userAuth.UserId, modules.Users, operations.Read) + if err != nil { + util.WriteError(r.Context(), errors.NewPermissionValidationError(err), w) + return + } + if !allowed { + util.WriteError(r.Context(), errors.NewPermissionDeniedError(), w) + } + + users, err := h.manager.GetAllUsers(r.Context(), nil, db.LockingStrengthShare, userAuth.AccountId) + if err != nil { + http.Error(w, "Internal Server Error", http.StatusInternalServerError) + return + } + _ = json.NewEncoder(w).Encode(users) +} + +func (h *handler) getUser(w http.ResponseWriter, r *http.Request) { + userAuth, err := nbcontext.GetUserAuthFromContext(r.Context()) + if err != nil { + util.WriteError(r.Context(), err, w) + return + } + + allowed, err := h.permissionsManager.ValidateUserPermissions(r.Context(), userAuth.AccountId, userAuth.UserId, modules.Users, operations.Read) + if err != nil { + util.WriteError(r.Context(), errors.NewPermissionValidationError(err), w) + return + } + if !allowed { + util.WriteError(r.Context(), errors.NewPermissionDeniedError(), w) + } + + vars := mux.Vars(r) + userId := vars["userId"] + + user, err := h.manager.GetUserByID(r.Context(), nil, db.LockingStrengthShare, userId) + if err != nil { + http.Error(w, "Not Found", http.StatusNotFound) + return + } + _ = json.NewEncoder(w).Encode(user) +} diff --git a/internal/modules/groups/manager.go b/internal/modules/groups/manager.go new file mode 100644 index 0000000..f50710f --- /dev/null +++ b/internal/modules/groups/manager.go @@ -0,0 +1,144 @@ +package groups + +import ( + "context" + "fmt" + + "github.com/gorilla/mux" + "github.com/netbirdio/netbird/management/server/http/api" + "github.com/netbirdio/netbird/management/server/store" + "github.com/netbirdio/netbird/management/server/types" + + "management/internal/shared/db" + "management/internal/shared/permissions" +) + +type Manager struct { + repo Repository +} + +func NewManager(store *db.Store, router *mux.Router, permissionsManager permissions.Manager) *Manager { + repo := newRepository(store) + m := &Manager{repo: repo} + api := newHandler(m, permissionsManager) + api.RegisterEndpoints(router) + return m +} + +func (m *Manager) GetAllGroups(ctx context.Context, accountID, userID string) ([]*types.Group, error) { + groups, err := m.repo.GetAccountGroups(ctx, store.LockingStrengthShare, accountID) + if err != nil { + return nil, fmt.Errorf("error getting account groups: %w", err) + } + + return groups, nil +} + +func (m *Manager) GetAllGroupsMap(ctx context.Context, accountID, userID string) (map[string]*types.Group, error) { + groups, err := m.GetAllGroups(ctx, accountID, userID) + if err != nil { + return nil, err + } + + groupsMap := make(map[string]*types.Group) + for _, group := range groups { + groupsMap[group.ID] = group + } + + return groupsMap, nil +} + +func (m *Manager) AddResourceToGroup(ctx context.Context, accountID, userID, groupID string, resource *types.Resource) error { + event, err := m.AddResourceToGroupInTransaction(ctx, m.store, accountID, userID, groupID, resource) + if err != nil { + return fmt.Errorf("error adding resource to group: %w", err) + } + + event() + + return nil +} + +func (m *Manager) AddResourceToGroupInTransaction(ctx context.Context, tx db.Transaction, accountID, userID, groupID string, resource *types.Resource) (func(), error) { + err := transaction.AddResourceToGroup(ctx, accountID, groupID, resource) + if err != nil { + return nil, fmt.Errorf("error adding resource to group: %w", err) + } + + group, err := transaction.GetGroupByID(ctx, store.LockingStrengthShare, accountID, groupID) + if err != nil { + return nil, fmt.Errorf("error getting group: %w", err) + } + + // TODO: at some point, this will need to become a switch statement + networkResource, err := transaction.GetNetworkResourceByID(ctx, store.LockingStrengthShare, accountID, resource.ID) + if err != nil { + return nil, fmt.Errorf("error getting network resource: %w", err) + } + + event := func() { + m.accountManager.StoreEvent(ctx, userID, groupID, accountID, activity.ResourceAddedToGroup, group.EventMetaResource(networkResource)) + } + + return event, nil +} + +func (m *Manager) RemoveResourceFromGroupInTransaction(ctx context.Context, transaction store.Store, accountID, userID, groupID, resourceID string) (func(), error) { + err := transaction.RemoveResourceFromGroup(ctx, accountID, groupID, resourceID) + if err != nil { + return nil, fmt.Errorf("error removing resource from group: %w", err) + } + + group, err := transaction.GetGroupByID(ctx, store.LockingStrengthShare, accountID, groupID) + if err != nil { + return nil, fmt.Errorf("error getting group: %w", err) + } + + // TODO: at some point, this will need to become a switch statement + networkResource, err := transaction.GetNetworkResourceByID(ctx, store.LockingStrengthShare, accountID, resourceID) + if err != nil { + return nil, fmt.Errorf("error getting network resource: %w", err) + } + + event := func() { + m.accountManager.StoreEvent(ctx, userID, groupID, accountID, activity.ResourceRemovedFromGroup, group.EventMetaResource(networkResource)) + } + + return event, nil +} + +func (m *Manager) GetResourceGroupsInTransaction(ctx context.Context, transaction store.Store, lockingStrength store.LockingStrength, accountID, resourceID string) ([]*types.Group, error) { + return transaction.GetResourceGroups(ctx, lockingStrength, accountID, resourceID) +} + +func ToGroupsInfoMap(groups []*types.Group, idCount int) map[string][]api.GroupMinimum { + groupsInfoMap := make(map[string][]api.GroupMinimum, idCount) + groupsChecked := make(map[string]struct{}, len(groups)) // not sure why this is needed (left over from old implementation) + for _, group := range groups { + _, ok := groupsChecked[group.ID] + if ok { + continue + } + + groupsChecked[group.ID] = struct{}{} + for _, pk := range group.Peers { + info := api.GroupMinimum{ + Id: group.ID, + Name: group.Name, + PeersCount: len(group.Peers), + ResourcesCount: len(group.Resources), + } + groupsInfoMap[pk] = append(groupsInfoMap[pk], info) + } + for _, rk := range group.Resources { + info := api.GroupMinimum{ + Id: group.ID, + Name: group.Name, + PeersCount: len(group.Peers), + ResourcesCount: len(group.Resources), + } + groupsInfoMap[rk.ID] = append(groupsInfoMap[rk.ID], info) + } + } + return groupsInfoMap +} diff --git a/internal/modules/groups/repository.go b/internal/modules/groups/repository.go new file mode 100644 index 0000000..c4dc24f --- /dev/null +++ b/internal/modules/groups/repository.go @@ -0,0 +1,21 @@ +package groups + +import ( + "management/internal/shared/db" +) + +type Repository interface { + RunInTx(fn func(tx db.Transaction) error) error +} + +type repository struct { + store *db.Store +} + +func newRepository(s *db.Store) Repository { + return &repository{store: s} +} + +func (r *repository) RunInTx(fn func(tx db.Transaction) error) error { + return r.store.RunInTx(fn) +} diff --git a/internal/modules/peers/api.go b/internal/modules/peers/api.go new file mode 100644 index 0000000..9d054a5 --- /dev/null +++ b/internal/modules/peers/api.go @@ -0,0 +1,94 @@ +package peers + +import ( + "fmt" + "net/http" + + "github.com/gorilla/mux" + nbcontext "github.com/netbirdio/netbird/management/server/context" + "github.com/netbirdio/netbird/management/server/groups" + "github.com/netbirdio/netbird/management/server/http/api" + "github.com/netbirdio/netbird/management/server/http/util" + + "management/internal/shared/db" + "management/internal/shared/permissions" +) + +type handler struct { + manager *Manager + permissionsManager permissions.Manager +} + +func newHandler(manager *Manager, permissionsManager permissions.Manager) *handler { + return &handler{ + manager: manager, + permissionsManager: permissionsManager, + } +} + +func (h *handler) RegisterEndpoints(router *mux.Router) { + router.HandleFunc("/peers", h.getAllPeers).Methods("GET", "OPTIONS") + router.HandleFunc("/peers/{peerId}", h.getPeer).Methods("GET", "OPTIONS") + router.HandleFunc("/peers/{peerId}", h.updatePeer).Methods("PUT", "OPTIONS") + router.HandleFunc("/peers/{peerId}", h.deletePeer).Methods("DELETE", "OPTIONS") + router.HandleFunc("/peers/{peerId}/accessible-peers", h.getAccessiblePeers).Methods("GET", "OPTIONS") +} + +func (h *handler) getAllPeers(w http.ResponseWriter, r *http.Request) { + userAuth, err := nbcontext.GetUserAuthFromContext(r.Context()) + if err != nil { + util.WriteError(r.Context(), err, w) + return + } + + nameFilter := r.URL.Query().Get("name") + ipFilter := r.URL.Query().Get("ip") + + peers, err := h.manager.GetFilteredPeers(r.Context(), nil, db.LockingStrengthShare, userAuth.AccountId, nameFilter, ipFilter) + if err != nil { + util.WriteError(r.Context(), err, w) + return + } + + dnsDomain := h.accountManager.GetDNSDomain() + + grps, _ := h.accountManager.GetAllGroups(r.Context(), accountID, userID) + + grpsInfoMap := groups.ToGroupsInfoMap(grps, len(peers)) + respBody := make([]*api.PeerBatch, 0, len(peers)) + for _, peer := range peers { + peerToReturn, err := h.checkPeerStatus(peer) + if err != nil { + util.WriteError(r.Context(), err, w) + return + } + + respBody = append(respBody, toPeerListItemResponse(peerToReturn, grpsInfoMap[peer.ID], dnsDomain, 0)) + } + + validPeersMap, err := h.accountManager.GetValidatedPeers(r.Context(), accountID) + if err != nil { + log.WithContext(r.Context()).Errorf("failed to list appreoved peers: %v", err) + util.WriteError(r.Context(), fmt.Errorf("internal error"), w) + return + } + h.setApprovalRequiredFlag(respBody, validPeersMap) + + util.WriteJSONObject(r.Context(), w, respBody) +} + +func (h *handler) getPeer(w http.ResponseWriter, r *http.Request) { + +} + +func (h *handler) updatePeer(w http.ResponseWriter, r *http.Request) { + +} + +func (h *handler) deletePeer(w http.ResponseWriter, r *http.Request) { + +} + +func (h *handler) getAccessiblePeers(w http.ResponseWriter, r *http.Request) { + +} diff --git a/internal/modules/peers/manager.go b/internal/modules/peers/manager.go index 7dd5fa6..be5af1d 100644 --- a/internal/modules/peers/manager.go +++ b/internal/modules/peers/manager.go @@ -1 +1,112 @@ package peers + +import ( + "context" + + "github.com/gorilla/mux" + + "management/internal/modules/peers/types" + "management/internal/shared/db" + "management/internal/shared/permissions" + "management/pkg/logging" +) + +var log = logging.LoggerForThisPackage() + +type Manager struct { + repo Repository +} + +func NewManager(store *db.Store, router *mux.Router, permissionsManager permissions.Manager) *Manager { + repo := newRepository(store) + m := &Manager{repo: repo} + api := newHandler(m, permissionsManager) + api.RegisterEndpoints(router) + return m +} + +func (m *Manager) GetPeer(ctx context.Context, tx db.Transaction, strength db.LockingStrength, accountID, peerID string) (*types.Peer, error) { + return m.repo.GetPeerByID(tx, strength, accountID, peerID) +} + +func (m *Manager) GetPeers(ctx context.Context, tx db.Transaction, strength db.LockingStrength, accountID string) ([]*types.Peer, error) { + return m.repo.GetPeers(tx, strength, accountID) +} + +func (m *Manager) GetFilteredPeers(ctx context.Context, tx db.Transaction, strength db.LockingStrength, accountID, nameFilter, ipFilter string) ([]*types.Peer, error) { + return m.repo.GetFilteredPeers(tx, strength, accountID, nameFilter, ipFilter) +} + +import ( + "context" + + "github.com/gorilla/mux" + + "management/internal/modules/peers/types" + "management/internal/shared/db" + "management/internal/shared/permissions" + "management/pkg/logging" +) + +var log = logging.LoggerForThisPackage() + +type Manager struct { + repo Repository +} + +func NewManager(store *db.Store, router *mux.Router, permissionsManager permissions.Manager) *Manager { + repo := newRepository(store) + m := &Manager{repo: repo} + api := newHandler(m, permissionsManager) + api.RegisterEndpoints(router) + return m +} + +func (m *Manager) GetPeer(ctx context.Context, tx db.Transaction, strength db.LockingStrength, accountID, peerID string) (*types.Peer, error) { + return m.repo.GetPeerByID(tx, strength, accountID, peerID) +} + +func (m *Manager) GetPeers(ctx context.Context, tx db.Transaction, strength db.LockingStrength, accountID string) ([]*types.Peer, error) { + return m.repo.GetPeers(tx, strength, accountID) +} + +func (m *Manager) GetFilteredPeers(ctx context.Context, tx db.Transaction, strength db.LockingStrength, accountID, nameFilter, ipFilter string) ([]*types.Peer, error) { + return m.repo.GetFilteredPeers(tx, strength, accountID, nameFilter, ipFilter) +} + +import ( + "context" + + "github.com/gorilla/mux" + + "management/internal/modules/peers/types" + "management/internal/shared/db" + "management/internal/shared/permissions" + "management/pkg/logging" +) + +var log = logging.LoggerForThisPackage() + +type Manager struct { + repo Repository +} + +func NewManager(store *db.Store, router *mux.Router, permissionsManager permissions.Manager) *Manager { + repo := newRepository(store) + m := &Manager{repo: repo} + api := newHandler(m, permissionsManager) + api.RegisterEndpoints(router) + return m +} + +func (m *Manager) GetPeer(ctx context.Context, tx db.Transaction, strength db.LockingStrength, accountID, peerID string) (*types.Peer, error) { + return m.repo.GetPeerByID(tx, strength, accountID, peerID) +} + +func (m *Manager) GetPeers(ctx context.Context, tx db.Transaction, strength db.LockingStrength, accountID string) ([]*types.Peer, error) { + return m.repo.GetPeers(tx, strength, accountID) +} + +func (m *Manager) GetFilteredPeers(ctx context.Context, tx db.Transaction, strength db.LockingStrength, accountID, nameFilter, ipFilter string) ([]*types.Peer, error) { + return m.repo.GetFilteredPeers(tx, strength, accountID, nameFilter, ipFilter) +} diff --git a/internal/modules/peers/repository.go b/internal/modules/peers/repository.go new file mode 100644 index 0000000..0c263b4 --- /dev/null +++ b/internal/modules/peers/repository.go @@ -0,0 +1,65 @@ +package peers + +import ( + "management/internal/modules/peers/types" + "management/internal/shared/db" +) + +type Repository interface { + RunInTx(fn func(tx db.Transaction) error) error + GetPeerByID(tx db.Transaction, strength db.LockingStrength, accountID, peerId string) (*types.Peer, error) + GetPeers(tx db.Transaction, strength db.LockingStrength, accountID string) ([]*types.Peer, error) + GetFilteredPeers(tx db.Transaction, strength db.LockingStrength, accountID string, nameFilter, ipFilter string) ([]*types.Peer, error) +} + +type repository struct { + store *db.Store +} + +func newRepository(s *db.Store) Repository { + return &repository{store: s} +} + +func (r *repository) RunInTx(fn func(tx db.Transaction) error) error { + return r.store.RunInTx(fn) +} + +func (r *repository) GetPeerByID(tx db.Transaction, strength db.LockingStrength, accountID, peerId string) (*types.Peer, error) { + var peer types.Peer + err := r.store.GetOne(tx, strength, &peer, "account_id = ? AND id = ?", accountID, peerId) + if err != nil { + return nil, err + } + return &peer, nil +} + +func (r *repository) GetPeers(tx db.Transaction, strength db.LockingStrength, accountID string) ([]*types.Peer, error) { + var peers []*types.Peer + err := r.store.GetMany(tx, strength, &peers, "account_id = ?", accountID) + if err != nil { + return nil, err + } + return peers, nil +} + +func (r *repository) GetFilteredPeers(tx db.Transaction, strength db.LockingStrength, accountID string, nameFilter, ipFilter string) ([]*types.Peer, error) { + query := "account_id = ?" + args := []interface{}{accountID} + + if nameFilter != "" { + query += " AND name LIKE ?" + args = append(args, nameFilter) + } + + if ipFilter != "" { + query += " AND ip LIKE ?" + args = append(args, ipFilter) + } + + var peers []*types.Peer + err := r.store.GetMany(tx, strength, &peers, query, args) + if err != nil { + return nil, err + } + return peers, nil +} diff --git a/internal/modules/peers/types/peer.go b/internal/modules/peers/types/peer.go new file mode 100644 index 0000000..d66850e --- /dev/null +++ b/internal/modules/peers/types/peer.go @@ -0,0 +1,317 @@ +package types + +import ( + "net" + "net/netip" + "slices" + "sort" + "time" + + "github.com/netbirdio/netbird/management/server/util" +) + +// Peer represents a machine connected to the network. +// The Peer is a WireGuard peer identified by a public key +type Peer struct { + // ID is an internal ID of the peer + ID string `gorm:"primaryKey"` + // AccountID is a reference to Account that this object belongs + AccountID string `json:"-" gorm:"index"` + // WireGuard public key + Key string `gorm:"index"` + // IP address of the Peer + IP net.IP `gorm:"serializer:json"` + // Meta is a Peer system meta data + Meta PeerSystemMeta `gorm:"embedded;embeddedPrefix:meta_"` + // Name is peer's name (machine name) + Name string + // DNSLabel is the parsed peer name for domain resolution. It is used to form an FQDN by appending the account's + // domain to the peer label. e.g. peer-dns-label.netbird.cloud + DNSLabel string + // Status peer's management connection status + Status *PeerStatus `gorm:"embedded;embeddedPrefix:peer_status_"` + // The user ID that registered the peer + UserID string + // SSHKey is a public SSH key of the peer + SSHKey string + // SSHEnabled indicates whether SSH server is enabled on the peer + SSHEnabled bool + // LoginExpirationEnabled indicates whether peer's login expiration is enabled and once expired the peer has to re-login. + // Works with LastLogin + LoginExpirationEnabled bool + + InactivityExpirationEnabled bool + // LastLogin the time when peer performed last login operation + LastLogin *time.Time + // CreatedAt records the time the peer was created + CreatedAt time.Time + // Indicate ephemeral peer attribute + Ephemeral bool `gorm:"index"` + // Geo location based on connection IP + Location Location `gorm:"embedded;embeddedPrefix:location_"` + + // ExtraDNSLabels is a list of additional DNS labels that can be used to resolve the peer + ExtraDNSLabels []string `gorm:"serializer:json"` + // AllowExtraDNSLabels indicates whether the peer allows extra DNS labels to be used for resolving the peer + AllowExtraDNSLabels bool +} + +type PeerStatus struct { //nolint:revive + // LastSeen is the last time peer was connected to the management service + LastSeen time.Time + // Connected indicates whether peer is connected to the management service or not + Connected bool + // LoginExpired + LoginExpired bool + // RequiresApproval indicates whether peer requires approval or not + RequiresApproval bool +} + +// Location is a geo location information of a Peer based on public connection IP +type Location struct { + ConnectionIP net.IP `gorm:"serializer:json"` // from grpc peer or reverse proxy headers depends on setup + CountryCode string + CityName string + GeoNameID uint // city level geoname id +} + +// NetworkAddress is the IP address with network and MAC address of a network interface +type NetworkAddress struct { + NetIP netip.Prefix `gorm:"serializer:json"` + Mac string +} + +// Environment is a system environment information +type Environment struct { + Cloud string + Platform string +} + +// File is a file on the system. +type File struct { + Path string + Exist bool + ProcessIsRunning bool +} + +// PeerSystemMeta is a metadata of a Peer machine system +type PeerSystemMeta struct { //nolint:revive + Hostname string + GoOS string + Kernel string + Core string + Platform string + OS string + OSVersion string + WtVersion string + UIVersion string + KernelVersion string + NetworkAddresses []NetworkAddress `gorm:"serializer:json"` + SystemSerialNumber string + SystemProductName string + SystemManufacturer string + Environment Environment `gorm:"serializer:json"` + Files []File `gorm:"serializer:json"` +} + +func (p PeerSystemMeta) isEqual(other PeerSystemMeta) bool { + sort.Slice(p.NetworkAddresses, func(i, j int) bool { + return p.NetworkAddresses[i].Mac < p.NetworkAddresses[j].Mac + }) + sort.Slice(other.NetworkAddresses, func(i, j int) bool { + return other.NetworkAddresses[i].Mac < other.NetworkAddresses[j].Mac + }) + equalNetworkAddresses := slices.EqualFunc(p.NetworkAddresses, other.NetworkAddresses, func(addr NetworkAddress, oAddr NetworkAddress) bool { + return addr.Mac == oAddr.Mac && addr.NetIP == oAddr.NetIP + }) + if !equalNetworkAddresses { + return false + } + + sort.Slice(p.Files, func(i, j int) bool { + return p.Files[i].Path < p.Files[j].Path + }) + sort.Slice(other.Files, func(i, j int) bool { + return other.Files[i].Path < other.Files[j].Path + }) + equalFiles := slices.EqualFunc(p.Files, other.Files, func(file File, oFile File) bool { + return file.Path == oFile.Path && file.Exist == oFile.Exist && file.ProcessIsRunning == oFile.ProcessIsRunning + }) + if !equalFiles { + return false + } + + return p.Hostname == other.Hostname && + p.GoOS == other.GoOS && + p.Kernel == other.Kernel && + p.KernelVersion == other.KernelVersion && + p.Core == other.Core && + p.Platform == other.Platform && + p.OS == other.OS && + p.OSVersion == other.OSVersion && + p.WtVersion == other.WtVersion && + p.UIVersion == other.UIVersion && + p.SystemSerialNumber == other.SystemSerialNumber && + p.SystemProductName == other.SystemProductName && + p.SystemManufacturer == other.SystemManufacturer && + p.Environment.Cloud == other.Environment.Cloud && + p.Environment.Platform == other.Environment.Platform +} + +func (p PeerSystemMeta) isEmpty() bool { + return p.Hostname == "" && + p.GoOS == "" && + p.Kernel == "" && + p.Core == "" && + p.Platform == "" && + p.OS == "" && + p.OSVersion == "" && + p.WtVersion == "" && + p.UIVersion == "" && + p.KernelVersion == "" && + len(p.NetworkAddresses) == 0 && + p.SystemSerialNumber == "" && + p.SystemProductName == "" && + p.SystemManufacturer == "" && + p.Environment.Cloud == "" && + p.Environment.Platform == "" && + len(p.Files) == 0 +} + +// AddedWithSSOLogin indicates whether this peer has been added with an SSO login by a user. +func (p *Peer) AddedWithSSOLogin() bool { + return p.UserID != "" +} + +// Copy copies Peer object +func (p *Peer) Copy() *Peer { + peerStatus := p.Status + if peerStatus != nil { + peerStatus = p.Status.Copy() + } + return &Peer{ + ID: p.ID, + AccountID: p.AccountID, + Key: p.Key, + IP: p.IP, + Meta: p.Meta, + Name: p.Name, + DNSLabel: p.DNSLabel, + Status: peerStatus, + UserID: p.UserID, + SSHKey: p.SSHKey, + SSHEnabled: p.SSHEnabled, + LoginExpirationEnabled: p.LoginExpirationEnabled, + LastLogin: p.LastLogin, + CreatedAt: p.CreatedAt, + Ephemeral: p.Ephemeral, + Location: p.Location, + InactivityExpirationEnabled: p.InactivityExpirationEnabled, + ExtraDNSLabels: slices.Clone(p.ExtraDNSLabels), + AllowExtraDNSLabels: p.AllowExtraDNSLabels, + } +} + +// UpdateMetaIfNew updates peer's system metadata if new information is provided +// returns true if meta was updated, false otherwise +func (p *Peer) UpdateMetaIfNew(meta PeerSystemMeta) bool { + if meta.isEmpty() { + return false + } + + // Avoid overwriting UIVersion if the update was triggered sole by the CLI client + if meta.UIVersion == "" { + meta.UIVersion = p.Meta.UIVersion + } + + if p.Meta.isEqual(meta) { + return false + } + p.Meta = meta + return true +} + +// GetLastLogin returns the last login time of the peer. +func (p *Peer) GetLastLogin() time.Time { + if p.LastLogin != nil { + return *p.LastLogin + } + return time.Time{} + +} + +// MarkLoginExpired marks peer's status expired or not +func (p *Peer) MarkLoginExpired(expired bool) { + newStatus := p.Status.Copy() + newStatus.LoginExpired = expired + if expired { + newStatus.Connected = false + } + p.Status = newStatus +} + +// SessionExpired indicates whether the peer's session has expired or not. +// If Peer.LastLogin plus the expiresIn duration has happened already; then session has expired. +// Return true if a session has expired, false otherwise, and time left to expiration (negative when expired). +// Session expiration can be disabled/enabled on a Peer level via Peer.LoginExpirationEnabled property. +// Session expiration can also be disabled/enabled globally on the Account level via Settings.PeerLoginExpirationEnabled. +// Only peers added by interactive SSO login can be expired. +func (p *Peer) SessionExpired(expiresIn time.Duration) (bool, time.Duration) { + if !p.AddedWithSSOLogin() || !p.InactivityExpirationEnabled || p.Status.Connected { + return false, 0 + } + expiresAt := p.Status.LastSeen.Add(expiresIn) + now := time.Now() + timeLeft := expiresAt.Sub(now) + return timeLeft <= 0, timeLeft +} + +// LoginExpired indicates whether the peer's login has expired or not. +// If Peer.LastLogin plus the expiresIn duration has happened already; then login has expired. +// Return true if a login has expired, false otherwise, and time left to expiration (negative when expired). +// Login expiration can be disabled/enabled on a Peer level via Peer.LoginExpirationEnabled property. +// Login expiration can also be disabled/enabled globally on the Account level via Settings.PeerLoginExpirationEnabled. +// Only peers added by interactive SSO login can be expired. +func (p *Peer) LoginExpired(expiresIn time.Duration) (bool, time.Duration) { + if !p.AddedWithSSOLogin() || !p.LoginExpirationEnabled { + return false, 0 + } + expiresAt := p.GetLastLogin().Add(expiresIn) + now := time.Now() + timeLeft := expiresAt.Sub(now) + return timeLeft <= 0, timeLeft +} + +// FQDN returns peers FQDN combined of the peer's DNS label and the system's DNS domain +func (p *Peer) FQDN(dnsDomain string) string { + if dnsDomain == "" { + return "" + } + return p.DNSLabel + "." + dnsDomain +} + +// EventMeta returns activity event meta related to the peer +func (p *Peer) EventMeta(dnsDomain string) map[string]any { + return map[string]any{"name": p.Name, "fqdn": p.FQDN(dnsDomain), "ip": p.IP, "created_at": p.CreatedAt, + "location_city_name": p.Location.CityName, "location_country_code": p.Location.CountryCode, + "location_geo_name_id": p.Location.GeoNameID, "location_connection_ip": p.Location.ConnectionIP} +} + +// Copy PeerStatus +func (p *PeerStatus) Copy() *PeerStatus { + return &PeerStatus{ + LastSeen: p.LastSeen, + Connected: p.Connected, + LoginExpired: p.LoginExpired, + RequiresApproval: p.RequiresApproval, + } +} + +// UpdateLastLogin and set login expired false +func (p *Peer) UpdateLastLogin() *Peer { + p.LastLogin = util.ToPtr(time.Now().UTC()) + newStatus := p.Status.Copy() + newStatus.LoginExpired = false + p.Status = newStatus + return p +} diff --git a/internal/modules/peers/types/peer_test.go b/internal/modules/peers/types/peer_test.go new file mode 100644 index 0000000..f06c8f0 --- /dev/null +++ b/internal/modules/peers/types/peer_test.go @@ -0,0 +1,85 @@ +package types + +import ( + "fmt" + "net/netip" + "testing" +) + +// FQDNOld is the original implementation for benchmarking purposes +func (p *Peer) FQDNOld(dnsDomain string) string { + if dnsDomain == "" { + return "" + } + return fmt.Sprintf("%s.%s", p.DNSLabel, dnsDomain) +} + +func BenchmarkFQDN(b *testing.B) { + p := &Peer{DNSLabel: "test-peer"} + dnsDomain := "example.com" + + b.Run("Old", func(b *testing.B) { + for i := 0; i < b.N; i++ { + p.FQDNOld(dnsDomain) + } + }) + + b.Run("New", func(b *testing.B) { + for i := 0; i < b.N; i++ { + p.FQDN(dnsDomain) + } + }) +} + +func TestIsEqual(t *testing.T) { + meta1 := PeerSystemMeta{ + NetworkAddresses: []NetworkAddress{{ + NetIP: netip.MustParsePrefix("192.168.1.2/24"), + Mac: "2", + }, + { + NetIP: netip.MustParsePrefix("192.168.1.0/24"), + Mac: "1", + }, + }, + Files: []File{ + { + Path: "/etc/hosts1", + Exist: true, + ProcessIsRunning: true, + }, + { + Path: "/etc/hosts2", + Exist: false, + ProcessIsRunning: false, + }, + }, + } + meta2 := PeerSystemMeta{ + NetworkAddresses: []NetworkAddress{ + { + NetIP: netip.MustParsePrefix("192.168.1.0/24"), + Mac: "1", + }, + { + NetIP: netip.MustParsePrefix("192.168.1.2/24"), + Mac: "2", + }, + }, + Files: []File{ + { + Path: "/etc/hosts2", + Exist: false, + ProcessIsRunning: false, + }, + { + Path: "/etc/hosts1", + Exist: true, + ProcessIsRunning: true, + }, + }, + } + if !meta1.isEqual(meta2) { + t.Error("meta1 should be equal to meta2") + } +} diff --git a/internal/modules/users/api.go b/internal/modules/users/api.go index 6e11f71..a0c42bb 100644 --- a/internal/modules/users/api.go +++ b/internal/modules/users/api.go @@ -27,7 +27,7 @@ func newHandler(manager *Manager, permissionsManager permissions.Manager) *handl } } -func (h *handler) RegisterAPI(router *mux.Router) { +func (h *handler) RegisterEndpoints(router *mux.Router) { router.HandleFunc("/users", h.getAllUsers).Methods("GET", "OPTIONS") router.HandleFunc("/users/{userId}", h.getUser).Methods("GET", "OPTIONS") } diff --git a/internal/modules/users/manager.go b/internal/modules/users/manager.go index 7b18599..44c6d7e 100644 --- a/internal/modules/users/manager.go +++ b/internal/modules/users/manager.go @@ -3,6 +3,8 @@ package users import ( "context" + "github.com/gorilla/mux" + "management/internal/modules/users/types" "management/internal/shared/db" "management/internal/shared/permissions" @@ -12,14 +14,14 @@ import ( var log = logging.LoggerForThisPackage() type Manager struct { - repo Repository - handler *handler + repo Repository } -func NewManager(store *db.Store, permissionsManager permissions.Manager) *Manager { +func NewManager(store *db.Store, router *mux.Router, permissionsManager permissions.Manager) *Manager { repo := newRepository(store) m := &Manager{repo: repo} - m.handler = newHandler(m, permissionsManager) + api := newHandler(m, permissionsManager) + api.RegisterEndpoints(router) return m } diff --git a/internal/server/config.go b/internal/server/config.go new file mode 100644 index 0000000..6bb0e7b --- /dev/null +++ b/internal/server/config.go @@ -0,0 +1,187 @@ +package server + +import ( + "net/netip" + + "github.com/netbirdio/netbird/management/server/idp" + "github.com/netbirdio/netbird/util" + + "management/internal/shared/db" +) + +type ( + // Protocol type + Protocol string + + // Provider authorization flow type + Provider string +) + +const ( + UDP Protocol = "udp" + DTLS Protocol = "dtls" + TCP Protocol = "tcp" + HTTP Protocol = "http" + HTTPS Protocol = "https" + NONE Provider = "none" +) + +const ( + // DefaultDeviceAuthFlowScope defines the bare minimum scope to request in the device authorization flow + DefaultDeviceAuthFlowScope string = "openid" +) + +var MgmtConfigPath string + +// Config of the Management service +type Config struct { + Stuns []*Host + TURNConfig *TURNConfig + Relay *Relay + Signal *Host + + Datadir string + DataStoreEncryptionKey string + + HttpConfig *HttpServerConfig + + IdpManagerConfig *idp.Config + + DeviceAuthorizationFlow *DeviceAuthorizationFlow + + PKCEAuthorizationFlow *PKCEAuthorizationFlow + + StoreConfig StoreConfig + + ReverseProxy ReverseProxy +} + +// GetAuthAudiences returns the audience from the http config and device authorization flow config +func (c Config) GetAuthAudiences() []string { + audiences := []string{c.HttpConfig.AuthAudience} + + if c.HttpConfig.ExtraAuthAudience != "" { + audiences = append(audiences, c.HttpConfig.ExtraAuthAudience) + } + + if c.DeviceAuthorizationFlow != nil && c.DeviceAuthorizationFlow.ProviderConfig.Audience != "" { + audiences = append(audiences, c.DeviceAuthorizationFlow.ProviderConfig.Audience) + } + + return audiences +} + +// TURNConfig is a config of the TURNCredentialsManager +type TURNConfig struct { + TimeBasedCredentials bool + CredentialsTTL util.Duration + Secret string + Turns []*Host +} + +// Relay configuration type +type Relay struct { + Addresses []string + CredentialsTTL util.Duration + Secret string +} + +// HttpServerConfig is a config of the HTTP Management service server +type HttpServerConfig struct { + LetsEncryptDomain string + // CertFile is the location of the certificate + CertFile string + // CertKey is the location of the certificate private key + CertKey string + // AuthAudience identifies the recipients that the JWT is intended for (aud in JWT) + AuthAudience string + // AuthIssuer identifies principal that issued the JWT + AuthIssuer string + // AuthUserIDClaim is the name of the claim that used as user ID + AuthUserIDClaim string + // AuthKeysLocation is a location of JWT key set containing the public keys used to verify JWT + AuthKeysLocation string + // OIDCConfigEndpoint is the endpoint of an IDP manager to get OIDC configuration + OIDCConfigEndpoint string + // IdpSignKeyRefreshEnabled identifies the signing key is currently being rotated or not + IdpSignKeyRefreshEnabled bool + // Extra audience + ExtraAuthAudience string +} + +// Host represents a Netbird host (e.g. STUN, TURN, Signal) +type Host struct { + Proto Protocol + // URI e.g. turns://stun.netbird.io:4430 or signal.netbird.io:10000 + URI string + Username string + Password string +} + +// DeviceAuthorizationFlow represents Device Authorization Flow information +// that can be used by the client to login initiate a Oauth 2.0 device authorization grant flow +// see https://datatracker.ietf.org/doc/html/rfc8628 +type DeviceAuthorizationFlow struct { + Provider string + ProviderConfig ProviderConfig +} + +// PKCEAuthorizationFlow represents Authorization Code Flow information +// that can be used by the client to login initiate a Oauth 2.0 authorization code grant flow +// with Proof Key for Code Exchange (PKCE). See https://datatracker.ietf.org/doc/html/rfc7636 +type PKCEAuthorizationFlow struct { + ProviderConfig ProviderConfig +} + +// ProviderConfig has all attributes needed to initiate a device/pkce authorization flow +type ProviderConfig struct { + // ClientID An IDP application client id + ClientID string + // ClientSecret An IDP application client secret + ClientSecret string + // Domain An IDP API domain + // Deprecated. Use TokenEndpoint and DeviceAuthEndpoint + Domain string + // Audience An Audience for to authorization validation + Audience string + // TokenEndpoint is the endpoint of an IDP manager where clients can obtain access token + TokenEndpoint string + // DeviceAuthEndpoint is the endpoint of an IDP manager where clients can obtain device authorization code + DeviceAuthEndpoint string + // AuthorizationEndpoint is the endpoint of an IDP manager where clients can obtain authorization code + AuthorizationEndpoint string + // Scopes provides the scopes to be included in the token request + Scope string + // UseIDToken indicates if the id token should be used for authentication + UseIDToken bool + // RedirectURL handles authorization code from IDP manager + RedirectURLs []string + // DisablePromptLogin makes the PKCE flow to not prompt the user for login + DisablePromptLogin bool +} + +// StoreConfig contains Store configuration +type StoreConfig struct { + Engine db.Engine +} + +// ReverseProxy contains reverse proxy configuration in front of management. +type ReverseProxy struct { + // TrustedHTTPProxies represents a list of trusted HTTP proxies by their IP prefixes. + // When extracting the real IP address from request headers, the middleware will verify + // if the peer's address falls within one of these trusted IP prefixes. + TrustedHTTPProxies []netip.Prefix + + // TrustedHTTPProxiesCount specifies the count of trusted HTTP proxies between the internet + // and the server. When using the trusted proxy count method to extract the real IP address, + // the middleware will search the X-Forwarded-For IP list from the rightmost by this count + // minus one. + TrustedHTTPProxiesCount uint + + // TrustedPeers represents a list of trusted peers by their IP prefixes. + // These peers are considered trustworthy by the gRPC server operator, + // and the middleware will attempt to extract the real IP address from + // request headers if the peer's address falls within one of these + // trusted IP prefixes. + TrustedPeers []netip.Prefix +} diff --git a/internal/shared/api/grpc/grpcserver.go b/internal/shared/api/grpc/grpcserver.go index 4ef7a22..7d856a7 100644 --- a/internal/shared/api/grpc/grpcserver.go +++ b/internal/shared/api/grpc/grpcserver.go @@ -40,13 +40,13 @@ type GRPCServer struct { settingsManager settings.Manager wgKey wgtypes.Key proto.UnimplementedManagementServiceServer - peersUpdateManager *PeersUpdateManager - config *types.Config - secretsManager SecretsManager - appMetrics telemetry.AppMetrics - ephemeralManager *EphemeralManager - peerLocks sync.Map - authManager auth.Manager + updateChannel *UpdateChannel + config *types.Config + secretsManager SecretsManager + appMetrics telemetry.AppMetrics + ephemeralManager *EphemeralManager + peerLocks sync.Map + authManager auth.Manager } // NewServer creates a new Management server @@ -55,7 +55,7 @@ func NewServer( config *types.Config, accountManager account.Manager, settingsManager settings.Manager, - peersUpdateManager *PeersUpdateManager, + updateChannel *UpdateChannel, secretsManager SecretsManager, appMetrics telemetry.AppMetrics, ephemeralManager *EphemeralManager, @@ -69,7 +69,7 @@ func NewServer( if appMetrics != nil { // update gauge based on number of connected peers which is equal to open gRPC streams err = appMetrics.GRPCMetrics().RegisterConnectedStreams(func() int64 { - return int64(len(peersUpdateManager.peerChannels)) + return int64(len(updateChannel.peerChannels)) }) if err != nil { return nil, err @@ -79,14 +79,14 @@ func NewServer( return &GRPCServer{ wgKey: key, // peerKey -> event channel - peersUpdateManager: peersUpdateManager, - accountManager: accountManager, - settingsManager: settingsManager, - config: config, - secretsManager: secretsManager, - authManager: authManager, - appMetrics: appMetrics, - ephemeralManager: ephemeralManager, + updateChannel: updateChannel, + accountManager: accountManager, + settingsManager: settingsManager, + config: config, + secretsManager: secretsManager, + authManager: authManager, + appMetrics: appMetrics, + ephemeralManager: ephemeralManager, }, nil } @@ -184,7 +184,7 @@ func (s *GRPCServer) Sync(req *proto.EncryptedMessage, srv proto.ManagementServi return err } - updates := s.peersUpdateManager.CreateChannel(ctx, peer.ID) + updates := s.updateChannel.CreateChannel(ctx, peer.ID) s.ephemeralManager.OnPeerConnected(ctx, peer) @@ -262,7 +262,7 @@ func (s *GRPCServer) cancelPeerRoutines(ctx context.Context, accountID string, p if err != nil { log.WithContext(ctx).Errorf("failed to disconnect peer %s properly: %v", peer.Key, err) } - s.peersUpdateManager.CloseChannel(ctx, peer.ID) + s.updateChannel.CloseChannel(ctx, peer.ID) s.secretsManager.CancelRefresh(peer.ID) s.ephemeralManager.OnPeerDisconnected(ctx, peer) diff --git a/internal/shared/api/grpc/updatechannel.go b/internal/shared/api/grpc/updatechannel.go new file mode 100644 index 0000000..1b40b3e --- /dev/null +++ b/internal/shared/api/grpc/updatechannel.go @@ -0,0 +1,178 @@ +package server + +import ( + "context" + "sync" + "time" + + log "github.com/sirupsen/logrus" + + "github.com/netbirdio/netbird/management/proto" + "github.com/netbirdio/netbird/management/server/telemetry" + "github.com/netbirdio/netbird/management/server/types" +) + +const channelBufferSize = 100 + +type UpdateMessage struct { + Update *proto.SyncResponse + NetworkMap *types.NetworkMap +} + +type UpdateChannel struct { + // peerChannels is an update channel indexed by Peer.ID + peerChannels map[string]chan *UpdateMessage + // channelsMux keeps the mutex to access peerChannels + channelsMux *sync.RWMutex + // metrics provides method to collect application metrics + metrics telemetry.AppMetrics +} + +// NewUpdateChannel returns a new instance of UpdateChannel +func NewUpdateChannel(metrics telemetry.AppMetrics) *UpdateChannel { + return &UpdateChannel{ + peerChannels: make(map[string]chan *UpdateMessage), + channelsMux: &sync.RWMutex{}, + metrics: metrics, + } +} + +// SendUpdate sends update message to the peer's channel +func (p *UpdateChannel) SendUpdate(ctx context.Context, peerID string, update *UpdateMessage) { + start := time.Now() + var found, dropped bool + + p.channelsMux.RLock() + + defer func() { + p.channelsMux.RUnlock() + if p.metrics != nil { + p.metrics.UpdateChannelMetrics().CountSendUpdateDuration(time.Since(start), found, dropped) + } + }() + + if channel, ok := p.peerChannels[peerID]; ok { + found = true + select { + case channel <- update: + log.WithContext(ctx).Debugf("update was sent to channel for peer %s", peerID) + default: + dropped = true + log.WithContext(ctx).Warnf("channel for peer %s is %d full or closed", peerID, len(channel)) + } + } else { + log.WithContext(ctx).Debugf("peer %s has no channel", peerID) + } +} + +// CreateChannel creates a go channel for a given peer used to deliver updates relevant to the peer. +func (p *UpdateChannel) CreateChannel(ctx context.Context, peerID string) chan *UpdateMessage { + start := time.Now() + + closed := false + + p.channelsMux.Lock() + defer func() { + p.channelsMux.Unlock() + if p.metrics != nil { + p.metrics.UpdateChannelMetrics().CountCreateChannelDuration(time.Since(start), closed) + } + }() + + if channel, ok := p.peerChannels[peerID]; ok { + closed = true + delete(p.peerChannels, peerID) + close(channel) + } + // mbragin: todo shouldn't it be more? or configurable? + channel := make(chan *UpdateMessage, channelBufferSize) + p.peerChannels[peerID] = channel + + log.WithContext(ctx).Debugf("opened updates channel for a peer %s", peerID) + + return channel +} + +func (p *UpdateChannel) closeChannel(ctx context.Context, peerID string) { + if channel, ok := p.peerChannels[peerID]; ok { + delete(p.peerChannels, peerID) + close(channel) + + log.WithContext(ctx).Debugf("closed updates channel of a peer %s", peerID) + return + } + + log.WithContext(ctx).Debugf("closing updates channel: peer %s has no channel", peerID) +} + +// CloseChannels closes updates channel for each given peer +func (p *UpdateChannel) CloseChannels(ctx context.Context, peerIDs []string) { + start := time.Now() + + p.channelsMux.Lock() + defer func() { + p.channelsMux.Unlock() + if p.metrics != nil { + p.metrics.UpdateChannelMetrics().CountCloseChannelsDuration(time.Since(start), len(peerIDs)) + } + }() + + for _, id := range peerIDs { + p.closeChannel(ctx, id) + } +} + +// CloseChannel closes updates channel of a given peer +func (p *UpdateChannel) CloseChannel(ctx context.Context, peerID string) { + start := time.Now() + + p.channelsMux.Lock() + defer func() { + p.channelsMux.Unlock() + if p.metrics != nil { + p.metrics.UpdateChannelMetrics().CountCloseChannelDuration(time.Since(start)) + } + }() + + p.closeChannel(ctx, peerID) +} + +// GetAllConnectedPeers returns a copy of the connected peers map +func (p *UpdateChannel) GetAllConnectedPeers() map[string]struct{} { + start := time.Now() + + p.channelsMux.RLock() + + m := make(map[string]struct{}) + + defer func() { + p.channelsMux.RUnlock() + if p.metrics != nil { + p.metrics.UpdateChannelMetrics().CountGetAllConnectedPeersDuration(time.Since(start), len(m)) + } + }() + + for ID := range p.peerChannels { + m[ID] = struct{}{} + } + + return m +} + +// HasChannel returns true if peers has channel in update manager, otherwise false +func (p *UpdateChannel) HasChannel(peerID string) bool { + start := time.Now() + + p.channelsMux.RLock() + + defer func() { + p.channelsMux.RUnlock() + if p.metrics != nil { + p.metrics.UpdateChannelMetrics().CountHasChannelDuration(time.Since(start)) + } + }() + + _, ok := p.peerChannels[peerID] + + return ok +} diff --git a/internal/controllers/network_map/updatechannel_test.go b/internal/shared/api/grpc/updatechannel_test.go similarity index 100% rename from internal/controllers/network_map/updatechannel_test.go rename to internal/shared/api/grpc/updatechannel_test.go diff --git a/internal/shared/api/rest/router.go b/internal/shared/api/rest/router.go index 7c9d882..931328b 100644 --- a/internal/shared/api/rest/router.go +++ b/internal/shared/api/rest/router.go @@ -8,7 +8,7 @@ import ( // NewRouter creates and returns a mux.Router configured with default middleware // and placeholder endpoints. You can add your own handlers here or in other files. -func NewRouter() http.Handler { +func NewRouter() *mux.Router { r := mux.NewRouter() // Attach middlewares diff --git a/internal/shared/db/store.go b/internal/shared/db/store.go index cb2d4ad..3f88d1e 100644 --- a/internal/shared/db/store.go +++ b/internal/shared/db/store.go @@ -84,3 +84,15 @@ func (s *Store) GetMany(tx Transaction, strength LockingStrength, dest interface func (s *Store) Delete(value interface{}) error { return s.db.Delete(value).Error } + +func (s *Store) Update(tx Transaction, value interface{}) error { + return s.Using(tx).Save(value).Error +} + +func (s *Store) Update(tx Transaction, value interface{}) error { + return s.Using(tx).Save(value).Error +} + +func (s *Store) Update(tx Transaction, value interface{}) error { + return s.Using(tx).Save(value).Error +} diff --git a/internal/shared/permissions/manager.go b/internal/shared/permissions/manager.go index c758a93..198fcd3 100644 --- a/internal/shared/permissions/manager.go +++ b/internal/shared/permissions/manager.go @@ -6,10 +6,10 @@ import ( "context" "github.com/netbirdio/netbird/management/server/status" - "github.com/netbirdio/netbird/management/server/store" "management/internal/modules/users/types" "management/internal/shared/activity" + "management/internal/shared/db" "management/internal/shared/permissions/modules" "management/internal/shared/permissions/operations" "management/internal/shared/permissions/roles" @@ -22,20 +22,23 @@ type Manager interface { ValidateUserPermissions(ctx context.Context, accountID, userID string, module modules.Module, operation operations.Operation) (bool, error) ValidateRoleModuleAccess(ctx context.Context, accountID string, role roles.RolePermissions, module modules.Module, operation operations.Operation) bool ValidateAccountAccess(ctx context.Context, accountID string, user *types.User, allowOwnerAndAdmin bool) error + Init(userManager userManager) } type userManager interface { - GetUserByUserID(ctx context.Context, lockingStrength store.LockingStrength, userID string) (*types.User, error) + GetUserByID(ctx context.Context, tx db.Transaction, strength db.LockingStrength, id string) (*types.User, error) } type managerImpl struct { userManager userManager } -func NewManager(userManager userManager) Manager { - return &managerImpl{ - userManager: userManager, - } +func NewManager() Manager { + return &managerImpl{} +} + +func (m *managerImpl) Init(userManager userManager) { + m.userManager = userManager } func (m *managerImpl) ValidateUserPermissions( @@ -49,7 +52,7 @@ func (m *managerImpl) ValidateUserPermissions( return true, nil } - user, err := m.userManager.GetUserByUserID(ctx, store.LockingStrengthShare, userID) + user, err := m.userManager.GetUserByID(ctx, nil, db.LockingStrengthShare, userID) if err != nil { return false, err }