[proxy] feature: bring your own proxy

management/internals/modules/reverseproxy/domain/manager/manager.go

@@ -31,6 +31,7 @@ type store interface {
 
 type proxyManager interface {
 	GetActiveClusterAddresses(ctx context.Context) ([]string, error)
+	GetActiveClusterAddressesForAccount(ctx context.Context, accountID string) ([]string, error)
 	ClusterSupportsCustomPorts(ctx context.Context, clusterAddr string) *bool
 	ClusterRequireSubdomain(ctx context.Context, clusterAddr string) *bool
 }
@@ -70,8 +71,8 @@ func (m Manager) GetDomains(ctx context.Context, accountID, userID string) ([]*d
 	var ret []*domain.Domain
 
 	// Add connected proxy clusters as free domains.
-	// The cluster address itself is the free domain base (e.g., "eu.proxy.netbird.io").
-	allowList, err := m.proxyManager.GetActiveClusterAddresses(ctx)
+	// For BYOP accounts, only their own cluster is returned; otherwise shared clusters.
+	allowList, err := m.getClusterAllowList(ctx, accountID)
 	if err != nil {
 		log.WithContext(ctx).Errorf("failed to get active proxy cluster addresses: %v", err)
 		return nil, err
@@ -123,8 +124,8 @@ func (m Manager) CreateDomain(ctx context.Context, accountID, userID, domainName
 		return nil, status.NewPermissionDeniedError()
 	}
 
-	// Verify the target cluster is in the available clusters
-	allowList, err := m.proxyManager.GetActiveClusterAddresses(ctx)
+	// Verify the target cluster is in the available clusters for this account
+	allowList, err := m.getClusterAllowList(ctx, accountID)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get active proxy cluster addresses: %w", err)
 	}
@@ -270,7 +271,7 @@ func (m Manager) GetClusterDomains() []string {
 // For free domains (those ending with a known cluster suffix), the cluster is extracted from the domain.
 // For custom domains, the cluster is determined by checking the registered custom domain's target cluster.
 func (m Manager) DeriveClusterFromDomain(ctx context.Context, accountID, domain string) (string, error) {
-	allowList, err := m.proxyManager.GetActiveClusterAddresses(ctx)
+	allowList, err := m.getClusterAllowList(ctx, accountID)
 	if err != nil {
 		return "", fmt.Errorf("failed to get active proxy cluster addresses: %w", err)
 	}
@@ -295,6 +296,17 @@ func (m Manager) DeriveClusterFromDomain(ctx context.Context, accountID, domain
 	return "", fmt.Errorf("domain %s does not match any available proxy cluster", domain)
 }
 
+func (m Manager) getClusterAllowList(ctx context.Context, accountID string) ([]string, error) {
+	byopAddresses, err := m.proxyManager.GetActiveClusterAddressesForAccount(ctx, accountID)
+	if err != nil {
+		return nil, fmt.Errorf("get BYOP cluster addresses: %w", err)
+	}
+	if len(byopAddresses) > 0 {
+		return byopAddresses, nil
+	}
+	return m.proxyManager.GetActiveClusterAddresses(ctx)
+}
+
 func extractClusterFromCustomDomains(serviceDomain string, customDomains []*domain.Domain) (string, bool) {
 	bestCluster := ""
 	bestLen := -1

management/internals/modules/reverseproxy/domain/manager/manager_test.go

@@ -0,0 +1,106 @@
+package manager
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+type mockProxyManager struct {
+	getActiveClusterAddressesFunc          func(ctx context.Context) ([]string, error)
+	getActiveClusterAddressesForAccountFunc func(ctx context.Context, accountID string) ([]string, error)
+}
+
+func (m *mockProxyManager) GetActiveClusterAddresses(ctx context.Context) ([]string, error) {
+	if m.getActiveClusterAddressesFunc != nil {
+		return m.getActiveClusterAddressesFunc(ctx)
+	}
+	return nil, nil
+}
+
+func (m *mockProxyManager) GetActiveClusterAddressesForAccount(ctx context.Context, accountID string) ([]string, error) {
+	if m.getActiveClusterAddressesForAccountFunc != nil {
+		return m.getActiveClusterAddressesForAccountFunc(ctx, accountID)
+	}
+	return nil, nil
+}
+
+func (m *mockProxyManager) ClusterSupportsCustomPorts(_ context.Context, _ string) *bool {
+	return nil
+}
+
+func (m *mockProxyManager) ClusterRequireSubdomain(_ context.Context, _ string) *bool {
+	return nil
+}
+
+func TestGetClusterAllowList_BYOPProxy(t *testing.T) {
+	pm := &mockProxyManager{
+		getActiveClusterAddressesForAccountFunc: func(_ context.Context, accID string) ([]string, error) {
+			assert.Equal(t, "acc-123", accID)
+			return []string{"byop.example.com"}, nil
+		},
+		getActiveClusterAddressesFunc: func(_ context.Context) ([]string, error) {
+			t.Fatal("should not call GetActiveClusterAddresses when BYOP addresses exist")
+			return nil, nil
+		},
+	}
+
+	mgr := Manager{proxyManager: pm}
+	result, err := mgr.getClusterAllowList(context.Background(), "acc-123")
+	require.NoError(t, err)
+	assert.Equal(t, []string{"byop.example.com"}, result)
+}
+
+func TestGetClusterAllowList_NoBYOP_FallbackToShared(t *testing.T) {
+	pm := &mockProxyManager{
+		getActiveClusterAddressesForAccountFunc: func(_ context.Context, _ string) ([]string, error) {
+			return nil, nil
+		},
+		getActiveClusterAddressesFunc: func(_ context.Context) ([]string, error) {
+			return []string{"eu.proxy.netbird.io", "us.proxy.netbird.io"}, nil
+		},
+	}
+
+	mgr := Manager{proxyManager: pm}
+	result, err := mgr.getClusterAllowList(context.Background(), "acc-123")
+	require.NoError(t, err)
+	assert.Equal(t, []string{"eu.proxy.netbird.io", "us.proxy.netbird.io"}, result)
+}
+
+func TestGetClusterAllowList_BYOPError_ReturnsError(t *testing.T) {
+	pm := &mockProxyManager{
+		getActiveClusterAddressesForAccountFunc: func(_ context.Context, _ string) ([]string, error) {
+			return nil, errors.New("db error")
+		},
+		getActiveClusterAddressesFunc: func(_ context.Context) ([]string, error) {
+			t.Fatal("should not call GetActiveClusterAddresses when BYOP lookup fails")
+			return nil, nil
+		},
+	}
+
+	mgr := Manager{proxyManager: pm}
+	result, err := mgr.getClusterAllowList(context.Background(), "acc-123")
+	require.Error(t, err)
+	assert.Nil(t, result)
+	assert.Contains(t, err.Error(), "BYOP cluster addresses")
+}
+
+func TestGetClusterAllowList_BYOPEmptySlice_FallbackToShared(t *testing.T) {
+	pm := &mockProxyManager{
+		getActiveClusterAddressesForAccountFunc: func(_ context.Context, _ string) ([]string, error) {
+			return []string{}, nil
+		},
+		getActiveClusterAddressesFunc: func(_ context.Context) ([]string, error) {
+			return []string{"eu.proxy.netbird.io"}, nil
+		},
+	}
+
+	mgr := Manager{proxyManager: pm}
+	result, err := mgr.getClusterAllowList(context.Background(), "acc-123")
+	require.NoError(t, err)
+	assert.Equal(t, []string{"eu.proxy.netbird.io"}, result)
+}
+

management/internals/modules/reverseproxy/proxy/manager.go

@@ -11,14 +11,19 @@ import (
 
 // Manager defines the interface for proxy operations
 type Manager interface {
-	Connect(ctx context.Context, proxyID, clusterAddress, ipAddress string, capabilities *Capabilities) error
+	Connect(ctx context.Context, proxyID, clusterAddress, ipAddress string, accountID *string, capabilities *Capabilities) error
 	Disconnect(ctx context.Context, proxyID string) error
 	Heartbeat(ctx context.Context, proxyID, clusterAddress, ipAddress string) error
 	GetActiveClusterAddresses(ctx context.Context) ([]string, error)
+	GetActiveClusterAddressesForAccount(ctx context.Context, accountID string) ([]string, error)
 	GetActiveClusters(ctx context.Context) ([]Cluster, error)
 	ClusterSupportsCustomPorts(ctx context.Context, clusterAddr string) *bool
 	ClusterRequireSubdomain(ctx context.Context, clusterAddr string) *bool
 	CleanupStale(ctx context.Context, inactivityDuration time.Duration) error
+	GetAccountProxy(ctx context.Context, accountID string) (*Proxy, error)
+	CountAccountProxies(ctx context.Context, accountID string) (int64, error)
+	IsClusterAddressAvailable(ctx context.Context, clusterAddress, accountID string) (bool, error)
+	DeleteProxy(ctx context.Context, proxyID string) error
 }
 
 // OIDCValidationConfig contains the OIDC configuration needed for token validation.

management/internals/modules/reverseproxy/proxy/manager/manager.go

@@ -13,12 +13,18 @@ import (
 // store defines the interface for proxy persistence operations
 type store interface {
 	SaveProxy(ctx context.Context, p *proxy.Proxy) error
+	DisconnectProxy(ctx context.Context, proxyID string) error
 	UpdateProxyHeartbeat(ctx context.Context, proxyID, clusterAddress, ipAddress string) error
 	GetActiveProxyClusterAddresses(ctx context.Context) ([]string, error)
+	GetActiveProxyClusterAddressesForAccount(ctx context.Context, accountID string) ([]string, error)
 	GetActiveProxyClusters(ctx context.Context) ([]proxy.Cluster, error)
 	GetClusterSupportsCustomPorts(ctx context.Context, clusterAddr string) *bool
 	GetClusterRequireSubdomain(ctx context.Context, clusterAddr string) *bool
 	CleanupStaleProxies(ctx context.Context, inactivityDuration time.Duration) error
+	GetProxyByAccountID(ctx context.Context, accountID string) (*proxy.Proxy, error)
+	CountProxiesByAccountID(ctx context.Context, accountID string) (int64, error)
+	IsClusterAddressConflicting(ctx context.Context, clusterAddress, accountID string) (bool, error)
+	DeleteProxy(ctx context.Context, proxyID string) error
 }
 
 // Manager handles all proxy operations
@@ -42,7 +48,7 @@ func NewManager(store store, meter metric.Meter) (*Manager, error) {
 
 // Connect registers a new proxy connection in the database.
 // capabilities may be nil for old proxies that do not report them.
-func (m Manager) Connect(ctx context.Context, proxyID, clusterAddress, ipAddress string, capabilities *proxy.Capabilities) error {
+func (m *Manager) Connect(ctx context.Context, proxyID, clusterAddress, ipAddress string, accountID *string, capabilities *proxy.Capabilities) error {
 	now := time.Now()
 	var caps proxy.Capabilities
 	if capabilities != nil {
@@ -52,9 +58,10 @@ func (m Manager) Connect(ctx context.Context, proxyID, clusterAddress, ipAddress
 		ID:             proxyID,
 		ClusterAddress: clusterAddress,
 		IPAddress:      ipAddress,
+		AccountID:      accountID,
 		LastSeen:       now,
 		ConnectedAt:    &now,
-		Status:         "connected",
+		Status:         proxy.StatusConnected,
 		Capabilities:   caps,
 	}
 
@@ -73,16 +80,8 @@ func (m Manager) Connect(ctx context.Context, proxyID, clusterAddress, ipAddress
 }
 
 // Disconnect marks a proxy as disconnected in the database
-func (m Manager) Disconnect(ctx context.Context, proxyID string) error {
-	now := time.Now()
-	p := &proxy.Proxy{
-		ID:             proxyID,
-		Status:         "disconnected",
-		DisconnectedAt: &now,
-		LastSeen:       now,
-	}
-
-	if err := m.store.SaveProxy(ctx, p); err != nil {
+func (m *Manager) Disconnect(ctx context.Context, proxyID string) error {
+	if err := m.store.DisconnectProxy(ctx, proxyID); err != nil {
 		log.WithContext(ctx).Errorf("failed to disconnect proxy %s: %v", proxyID, err)
 		return err
 	}
@@ -95,7 +94,7 @@ func (m Manager) Disconnect(ctx context.Context, proxyID string) error {
 }
 
 // Heartbeat updates the proxy's last seen timestamp
-func (m Manager) Heartbeat(ctx context.Context, proxyID, clusterAddress, ipAddress string) error {
+func (m *Manager) Heartbeat(ctx context.Context, proxyID, clusterAddress, ipAddress string) error {
 	if err := m.store.UpdateProxyHeartbeat(ctx, proxyID, clusterAddress, ipAddress); err != nil {
 		log.WithContext(ctx).Debugf("failed to update proxy %s heartbeat: %v", proxyID, err)
 		return err
@@ -107,7 +106,7 @@ func (m Manager) Heartbeat(ctx context.Context, proxyID, clusterAddress, ipAddre
 }
 
 // GetActiveClusterAddresses returns all unique cluster addresses for active proxies
-func (m Manager) GetActiveClusterAddresses(ctx context.Context) ([]string, error) {
+func (m *Manager) GetActiveClusterAddresses(ctx context.Context) ([]string, error) {
 	addresses, err := m.store.GetActiveProxyClusterAddresses(ctx)
 	if err != nil {
 		log.WithContext(ctx).Errorf("failed to get active proxy cluster addresses: %v", err)
@@ -139,10 +138,44 @@ func (m Manager) ClusterRequireSubdomain(ctx context.Context, clusterAddr string
 }
 
 // CleanupStale removes proxies that haven't sent heartbeat in the specified duration
-func (m Manager) CleanupStale(ctx context.Context, inactivityDuration time.Duration) error {
+func (m *Manager) CleanupStale(ctx context.Context, inactivityDuration time.Duration) error {
 	if err := m.store.CleanupStaleProxies(ctx, inactivityDuration); err != nil {
 		log.WithContext(ctx).Errorf("failed to cleanup stale proxies: %v", err)
 		return err
 	}
 	return nil
 }
+
+func (m *Manager) GetActiveClusterAddressesForAccount(ctx context.Context, accountID string) ([]string, error) {
+	addresses, err := m.store.GetActiveProxyClusterAddressesForAccount(ctx, accountID)
+	if err != nil {
+		log.WithContext(ctx).Errorf("failed to get active proxy cluster addresses for account %s: %v", accountID, err)
+		return nil, err
+	}
+	return addresses, nil
+}
+
+func (m *Manager) GetAccountProxy(ctx context.Context, accountID string) (*proxy.Proxy, error) {
+	return m.store.GetProxyByAccountID(ctx, accountID)
+}
+
+func (m *Manager) CountAccountProxies(ctx context.Context, accountID string) (int64, error) {
+	return m.store.CountProxiesByAccountID(ctx, accountID)
+}
+
+func (m *Manager) IsClusterAddressAvailable(ctx context.Context, clusterAddress, accountID string) (bool, error) {
+	conflicting, err := m.store.IsClusterAddressConflicting(ctx, clusterAddress, accountID)
+	if err != nil {
+		return false, err
+	}
+	return !conflicting, nil
+}
+
+func (m *Manager) DeleteProxy(ctx context.Context, proxyID string) error {
+	if err := m.store.DeleteProxy(ctx, proxyID); err != nil {
+		log.WithContext(ctx).Errorf("failed to delete proxy %s: %v", proxyID, err)
+		return err
+	}
+	return nil
+}
+