mirror of
https://github.com/netbirdio/netbird.git
synced 2026-04-22 10:16:38 +00:00
initial implementation
This commit is contained in:
Ashley Mensah
152
management/server/idp/migration/migration.go
Normal file
152
management/server/idp/migration/migration.go
Normal file
@@ -0,0 +1,152 @@
|
||||
// Package migration provides utility functions for migrating from an external IdP
|
||||
// to NetBird's embedded DEX-based IdP. It re-keys user IDs in the main store and
|
||||
// activity store so that they match DEX's encoded format.
|
||||
package migration
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
|
||||
"github.com/netbirdio/netbird/idp/dex"
|
||||
"github.com/netbirdio/netbird/management/server/types"
|
||||
)
|
||||
|
||||
// MainStoreUpdater is the subset of the main store needed for migration.
|
||||
type MainStoreUpdater interface {
|
||||
ListUsers(ctx context.Context) ([]*types.User, error)
|
||||
UpdateUserID(ctx context.Context, accountID, oldUserID, newUserID string) error
|
||||
}
|
||||
|
||||
// ActivityStoreUpdater is the subset of the activity store needed for migration.
|
||||
type ActivityStoreUpdater interface {
|
||||
UpdateUserID(ctx context.Context, oldUserID, newUserID string) error
|
||||
}
|
||||
|
||||
// Config holds migration parameters.
|
||||
type Config struct {
|
||||
ConnectorID string
|
||||
DryRun bool
|
||||
MainStore MainStoreUpdater
|
||||
ActivityStore ActivityStoreUpdater // nil if activity store is unavailable
|
||||
}
|
||||
|
||||
// Result holds migration outcome counts.
|
||||
type Result struct {
|
||||
Migrated int
|
||||
Skipped int
|
||||
}
|
||||
|
||||
// progressInterval controls how often progress is logged for large user counts.
|
||||
const progressInterval = 100
|
||||
|
||||
// Migrate re-keys every user ID in both stores so that it encodes the given
|
||||
// connector ID. Already-migrated users (detectable via DecodeDexUserID) are
|
||||
// skipped, making the operation idempotent.
|
||||
func Migrate(ctx context.Context, cfg *Config) (*Result, error) {
|
||||
if cfg.ConnectorID == "" {
|
||||
return nil, fmt.Errorf("connector ID must not be empty")
|
||||
}
|
||||
|
||||
users, err := cfg.MainStore.ListUsers(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("list users: %w", err)
|
||||
}
|
||||
|
||||
if len(users) == 0 {
|
||||
log.Info("no users found, nothing to migrate")
|
||||
return &Result{}, nil
|
||||
}
|
||||
|
||||
log.Infof("found %d users to process", len(users))
|
||||
|
||||
// Reconciliation pass: fix activity store for users already migrated in
|
||||
// the main DB but whose activity references may still use old IDs (from
|
||||
// a previous partial failure).
|
||||
if cfg.ActivityStore != nil && !cfg.DryRun {
|
||||
if err := reconcileActivityStore(ctx, cfg.ActivityStore, users); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
res := &Result{}
|
||||
|
||||
for i, user := range users {
|
||||
if user.Id == "" {
|
||||
log.Warnf("skipping user with empty ID in account %s", user.AccountID)
|
||||
res.Skipped++
|
||||
continue
|
||||
}
|
||||
|
||||
_, _, decErr := dex.DecodeDexUserID(user.Id)
|
||||
if decErr == nil {
|
||||
// Already encoded in DEX format — skip.
|
||||
res.Skipped++
|
||||
continue
|
||||
}
|
||||
|
||||
newUserID := dex.EncodeDexUserID(user.Id, cfg.ConnectorID)
|
||||
|
||||
if cfg.DryRun {
|
||||
log.Infof("[DRY RUN] would migrate user %s -> %s (account: %s)",
|
||||
user.Id, newUserID, user.AccountID)
|
||||
res.Migrated++
|
||||
continue
|
||||
}
|
||||
|
||||
if err := migrateUser(ctx, cfg, user.Id, user.AccountID, newUserID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
res.Migrated++
|
||||
|
||||
if (i+1)%progressInterval == 0 {
|
||||
log.Infof("progress: %d/%d users processed", i+1, len(users))
|
||||
}
|
||||
}
|
||||
|
||||
if cfg.DryRun {
|
||||
log.Infof("[DRY RUN] migration summary: %d users would be migrated, %d already migrated",
|
||||
res.Migrated, res.Skipped)
|
||||
} else {
|
||||
log.Infof("migration complete: %d users migrated, %d already migrated",
|
||||
res.Migrated, res.Skipped)
|
||||
}
|
||||
|
||||
return res, nil
|
||||
}
|
||||
|
||||
// reconcileActivityStore updates activity store references for users already
|
||||
// migrated in the main DB whose activity entries may still use old IDs from a
|
||||
// previous partial failure.
|
||||
func reconcileActivityStore(ctx context.Context, activityStore ActivityStoreUpdater, users []*types.User) error {
|
||||
for _, user := range users {
|
||||
originalID, _, err := dex.DecodeDexUserID(user.Id)
|
||||
if err != nil {
|
||||
// Not yet migrated — will be handled in the main loop.
|
||||
continue
|
||||
}
|
||||
if err := activityStore.UpdateUserID(ctx, originalID, user.Id); err != nil {
|
||||
return fmt.Errorf("reconcile activity store for user %s: %w", user.Id, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// migrateUser updates a single user's ID in both the main store and the activity store.
|
||||
func migrateUser(ctx context.Context, cfg *Config, oldID, accountID, newID string) error {
|
||||
if err := cfg.MainStore.UpdateUserID(ctx, accountID, oldID, newID); err != nil {
|
||||
return fmt.Errorf("update user ID for user %s: %w", oldID, err)
|
||||
}
|
||||
|
||||
if cfg.ActivityStore == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
if err := cfg.ActivityStore.UpdateUserID(ctx, oldID, newID); err != nil {
|
||||
return fmt.Errorf("update activity store user ID for user %s: %w", oldID, err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
287
management/server/idp/migration/migration_test.go
Normal file
287
management/server/idp/migration/migration_test.go
Normal file
@@ -0,0 +1,287 @@
|
||||
package migration
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/netbirdio/netbird/idp/dex"
|
||||
"github.com/netbirdio/netbird/management/server/types"
|
||||
)
|
||||
|
||||
const testConnectorID = "oidc"
|
||||
|
||||
// mockMainStore implements MainStoreUpdater for testing.
|
||||
type mockMainStore struct {
|
||||
users []*types.User
|
||||
listErr error
|
||||
updateErr error
|
||||
updateCalls []updateCall
|
||||
}
|
||||
|
||||
type updateCall struct {
|
||||
AccountID string
|
||||
OldID string
|
||||
NewID string
|
||||
}
|
||||
|
||||
func (m *mockMainStore) ListUsers(_ context.Context) ([]*types.User, error) {
|
||||
return m.users, m.listErr
|
||||
}
|
||||
|
||||
func (m *mockMainStore) UpdateUserID(_ context.Context, accountID, oldUserID, newUserID string) error {
|
||||
m.updateCalls = append(m.updateCalls, updateCall{accountID, oldUserID, newUserID})
|
||||
return m.updateErr
|
||||
}
|
||||
|
||||
// mockActivityStore implements ActivityStoreUpdater for testing.
|
||||
type mockActivityStore struct {
|
||||
updateErr error
|
||||
updateCalls []activityUpdateCall
|
||||
}
|
||||
|
||||
type activityUpdateCall struct {
|
||||
OldID string
|
||||
NewID string
|
||||
}
|
||||
|
||||
func (m *mockActivityStore) UpdateUserID(_ context.Context, oldUserID, newUserID string) error {
|
||||
m.updateCalls = append(m.updateCalls, activityUpdateCall{oldUserID, newUserID})
|
||||
return m.updateErr
|
||||
}
|
||||
|
||||
func TestMigrate_NormalMigration(t *testing.T) {
|
||||
mainStore := &mockMainStore{
|
||||
users: []*types.User{
|
||||
{Id: "user-1", AccountID: "acc-1"},
|
||||
{Id: "user-2", AccountID: "acc-1"},
|
||||
},
|
||||
}
|
||||
actStore := &mockActivityStore{}
|
||||
|
||||
res, err := Migrate(context.Background(), &Config{
|
||||
ConnectorID: testConnectorID,
|
||||
MainStore: mainStore,
|
||||
ActivityStore: actStore,
|
||||
})
|
||||
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, 2, res.Migrated)
|
||||
assert.Equal(t, 0, res.Skipped)
|
||||
assert.Len(t, mainStore.updateCalls, 2)
|
||||
assert.Len(t, actStore.updateCalls, 2)
|
||||
|
||||
// Verify the new IDs are DEX-encoded
|
||||
for _, call := range mainStore.updateCalls {
|
||||
userID, connID, decErr := dex.DecodeDexUserID(call.NewID)
|
||||
require.NoError(t, decErr)
|
||||
assert.Equal(t, testConnectorID, connID)
|
||||
assert.Equal(t, call.OldID, userID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMigrate_SkipAlreadyMigrated(t *testing.T) {
|
||||
alreadyMigrated := dex.EncodeDexUserID("original-user", testConnectorID)
|
||||
mainStore := &mockMainStore{
|
||||
users: []*types.User{
|
||||
{Id: alreadyMigrated, AccountID: "acc-1"},
|
||||
{Id: "not-migrated", AccountID: "acc-1"},
|
||||
},
|
||||
}
|
||||
actStore := &mockActivityStore{}
|
||||
|
||||
res, err := Migrate(context.Background(), &Config{
|
||||
ConnectorID: testConnectorID,
|
||||
MainStore: mainStore,
|
||||
ActivityStore: actStore,
|
||||
})
|
||||
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, 1, res.Migrated)
|
||||
assert.Equal(t, 1, res.Skipped)
|
||||
assert.Len(t, mainStore.updateCalls, 1)
|
||||
assert.Equal(t, "not-migrated", mainStore.updateCalls[0].OldID)
|
||||
}
|
||||
|
||||
func TestMigrate_DryRun(t *testing.T) {
|
||||
mainStore := &mockMainStore{
|
||||
users: []*types.User{
|
||||
{Id: "user-1", AccountID: "acc-1"},
|
||||
},
|
||||
}
|
||||
actStore := &mockActivityStore{}
|
||||
|
||||
res, err := Migrate(context.Background(), &Config{
|
||||
ConnectorID: testConnectorID,
|
||||
DryRun: true,
|
||||
MainStore: mainStore,
|
||||
ActivityStore: actStore,
|
||||
})
|
||||
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, 1, res.Migrated)
|
||||
// No actual updates should have been made
|
||||
assert.Empty(t, mainStore.updateCalls)
|
||||
assert.Empty(t, actStore.updateCalls)
|
||||
}
|
||||
|
||||
func TestMigrate_EmptyUserList(t *testing.T) {
|
||||
mainStore := &mockMainStore{users: []*types.User{}}
|
||||
actStore := &mockActivityStore{}
|
||||
|
||||
res, err := Migrate(context.Background(), &Config{
|
||||
ConnectorID: testConnectorID,
|
||||
MainStore: mainStore,
|
||||
ActivityStore: actStore,
|
||||
})
|
||||
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, 0, res.Migrated)
|
||||
assert.Equal(t, 0, res.Skipped)
|
||||
}
|
||||
|
||||
func TestMigrate_EmptyUserID(t *testing.T) {
|
||||
mainStore := &mockMainStore{
|
||||
users: []*types.User{
|
||||
{Id: "", AccountID: "acc-1"},
|
||||
{Id: "user-1", AccountID: "acc-1"},
|
||||
},
|
||||
}
|
||||
actStore := &mockActivityStore{}
|
||||
|
||||
res, err := Migrate(context.Background(), &Config{
|
||||
ConnectorID: testConnectorID,
|
||||
MainStore: mainStore,
|
||||
ActivityStore: actStore,
|
||||
})
|
||||
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, 1, res.Migrated)
|
||||
assert.Equal(t, 1, res.Skipped)
|
||||
}
|
||||
|
||||
func TestMigrate_NilActivityStore(t *testing.T) {
|
||||
mainStore := &mockMainStore{
|
||||
users: []*types.User{
|
||||
{Id: "user-1", AccountID: "acc-1"},
|
||||
},
|
||||
}
|
||||
|
||||
res, err := Migrate(context.Background(), &Config{
|
||||
ConnectorID: testConnectorID,
|
||||
MainStore: mainStore,
|
||||
// ActivityStore is nil
|
||||
})
|
||||
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, 1, res.Migrated)
|
||||
assert.Len(t, mainStore.updateCalls, 1)
|
||||
}
|
||||
|
||||
func TestMigrate_EmptyConnectorID(t *testing.T) {
|
||||
mainStore := &mockMainStore{}
|
||||
|
||||
_, err := Migrate(context.Background(), &Config{
|
||||
ConnectorID: "",
|
||||
MainStore: mainStore,
|
||||
})
|
||||
|
||||
require.Error(t, err)
|
||||
assert.Contains(t, err.Error(), "connector ID must not be empty")
|
||||
}
|
||||
|
||||
func TestMigrate_ListUsersError(t *testing.T) {
|
||||
mainStore := &mockMainStore{listErr: errors.New("db error")}
|
||||
|
||||
_, err := Migrate(context.Background(), &Config{
|
||||
ConnectorID: testConnectorID,
|
||||
MainStore: mainStore,
|
||||
})
|
||||
|
||||
require.Error(t, err)
|
||||
assert.Contains(t, err.Error(), "list users")
|
||||
}
|
||||
|
||||
func TestMigrate_UpdateError(t *testing.T) {
|
||||
mainStore := &mockMainStore{
|
||||
users: []*types.User{{Id: "user-1", AccountID: "acc-1"}},
|
||||
updateErr: errors.New("tx error"),
|
||||
}
|
||||
|
||||
_, err := Migrate(context.Background(), &Config{
|
||||
ConnectorID: testConnectorID,
|
||||
MainStore: mainStore,
|
||||
})
|
||||
|
||||
require.Error(t, err)
|
||||
assert.Contains(t, err.Error(), "update user ID")
|
||||
}
|
||||
|
||||
func TestMigrate_Reconciliation(t *testing.T) {
|
||||
// Simulate a previously migrated user whose activity store wasn't updated
|
||||
alreadyMigrated := dex.EncodeDexUserID("original-user", testConnectorID)
|
||||
mainStore := &mockMainStore{
|
||||
users: []*types.User{
|
||||
{Id: alreadyMigrated, AccountID: "acc-1"},
|
||||
},
|
||||
}
|
||||
actStore := &mockActivityStore{}
|
||||
|
||||
res, err := Migrate(context.Background(), &Config{
|
||||
ConnectorID: testConnectorID,
|
||||
MainStore: mainStore,
|
||||
ActivityStore: actStore,
|
||||
})
|
||||
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, 0, res.Migrated)
|
||||
assert.Equal(t, 1, res.Skipped)
|
||||
// Reconciliation should have called activity store with the original -> new mapping
|
||||
require.Len(t, actStore.updateCalls, 1)
|
||||
assert.Equal(t, "original-user", actStore.updateCalls[0].OldID)
|
||||
assert.Equal(t, alreadyMigrated, actStore.updateCalls[0].NewID)
|
||||
}
|
||||
|
||||
func TestMigrate_Idempotent(t *testing.T) {
|
||||
mainStore := &mockMainStore{
|
||||
users: []*types.User{
|
||||
{Id: "user-1", AccountID: "acc-1"},
|
||||
{Id: "user-2", AccountID: "acc-1"},
|
||||
},
|
||||
}
|
||||
actStore := &mockActivityStore{}
|
||||
|
||||
// First run
|
||||
res1, err := Migrate(context.Background(), &Config{
|
||||
ConnectorID: testConnectorID,
|
||||
MainStore: mainStore,
|
||||
ActivityStore: actStore,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, 2, res1.Migrated)
|
||||
|
||||
// Simulate that the store now has the migrated IDs
|
||||
for _, call := range mainStore.updateCalls {
|
||||
for i, u := range mainStore.users {
|
||||
if u.Id == call.OldID {
|
||||
mainStore.users[i].Id = call.NewID
|
||||
}
|
||||
}
|
||||
}
|
||||
mainStore.updateCalls = nil
|
||||
actStore.updateCalls = nil
|
||||
|
||||
// Second run should skip all
|
||||
res2, err := Migrate(context.Background(), &Config{
|
||||
ConnectorID: testConnectorID,
|
||||
MainStore: mainStore,
|
||||
ActivityStore: actStore,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, 0, res2.Migrated)
|
||||
assert.Equal(t, 2, res2.Skipped)
|
||||
assert.Empty(t, mainStore.updateCalls)
|
||||
}
|
||||
Reference in New Issue
Block a user