[management] Add proxy credentials limiter on management (#7569)

management/internals/shared/grpc/proxy.go

@@ -102,7 +102,8 @@ type ProxyServiceServer struct {
 
 	mu sync.RWMutex
 	// Manager for reverse proxy operations
-	serviceManager rpservice.Manager
+	serviceManager   rpservice.Manager
+	credentialLimits credentialVerificationLimiter
 	// agentNetworkSynth produces synthesised reverse-proxy services from
 	// Agent Network state. Optional — when nil the snapshot path only ships
 	// persisted services.
@@ -242,9 +243,10 @@ func (s *ProxyServiceServer) cleanupStaleProxies(ctx context.Context) {
 	}
 }
 
-// Close stops background goroutines.
+// Close stops background goroutines and releases credential verification state.
 func (s *ProxyServiceServer) Close() {
 	s.cancel()
+	s.credentialLimits.close()
 }
 
 // SetServiceManager sets the service manager. Must be called before serving.
@@ -1223,6 +1225,7 @@ func shallowCloneMapping(m *proto.ProxyMapping) *proto.ProxyMapping {
 	}
 }
 
+// Authenticate verifies service credentials and issues a session token.
 func (s *ProxyServiceServer) Authenticate(ctx context.Context, req *proto.AuthenticateRequest) (*proto.AuthenticateResponse, error) {
 	if err := enforceAccountScope(ctx, req.GetAccountId()); err != nil {
 		return nil, err
@@ -1234,6 +1237,14 @@ func (s *ProxyServiceServer) Authenticate(ctx context.Context, req *proto.Authen
 		return nil, status.Errorf(codes.FailedPrecondition, "get service from store: %v", err)
 	}
 
+	switch req.GetRequest().(type) {
+	case *proto.AuthenticateRequest_Pin, *proto.AuthenticateRequest_Password:
+		key := credentialVerificationKey{accountID: credentialAccountID(service.AccountID), serviceID: credentialServiceID(service.ID)}
+		if err := s.credentialLimits.allow(key); err != nil {
+			return nil, err
+		}
+	}
+
 	authenticated, userId, method := s.authenticateRequest(ctx, req, service)
 
 	// Non-OIDC schemes (PIN/Password/Header) authenticate against per-service

management/internals/shared/grpc/proxy_credential_limiter.go

@@ -0,0 +1,101 @@
+package grpc
+
+import (
+	"sync"
+	"time"
+
+	"golang.org/x/time/rate"
+	"google.golang.org/genproto/googleapis/rpc/errdetails"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/types/known/durationpb"
+)
+
+const (
+	credentialVerificationInterval        = 6 * time.Second
+	credentialVerificationBurst           = 5
+	credentialVerificationMaxServices     = 4096
+	credentialVerificationIdleTimeout     = 15 * time.Minute
+	credentialVerificationCleanupInterval = time.Minute
+)
+
+type credentialAccountID string
+type credentialServiceID string
+
+type credentialVerificationKey struct {
+	accountID credentialAccountID
+	serviceID credentialServiceID
+}
+
+type credentialVerificationBudget struct {
+	limiter  *rate.Limiter
+	lastUsed time.Time
+}
+
+// The zero value is ready to use. Budgets are local to this Management process;
+// proxy replicas reaching this process share a service's verification budget.
+type credentialVerificationLimiter struct {
+	mu          sync.Mutex
+	now         func() time.Time
+	services    map[credentialVerificationKey]*credentialVerificationBudget
+	nextCleanup time.Time
+	closed      bool
+}
+
+func (l *credentialVerificationLimiter) allow(key credentialVerificationKey) error {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	if l.closed {
+		return status.Error(codes.Unavailable, "credential verification is closed")
+	}
+	now := time.Now()
+	if l.now != nil {
+		now = l.now()
+	}
+	l.cleanup(now)
+	budget := l.services[key]
+	if budget == nil {
+		if len(l.services) >= credentialVerificationMaxServices {
+			return credentialVerificationThrottled(credentialVerificationCleanupInterval)
+		}
+		if l.services == nil {
+			l.services = make(map[credentialVerificationKey]*credentialVerificationBudget)
+		}
+		budget = &credentialVerificationBudget{limiter: rate.NewLimiter(rate.Every(credentialVerificationInterval), credentialVerificationBurst)}
+		l.services[key] = budget
+	}
+	budget.lastUsed = now
+	if budget.limiter.AllowN(now, 1) {
+		return nil
+	}
+	delay := max(time.Nanosecond, time.Duration((1-budget.limiter.TokensAt(now))*float64(credentialVerificationInterval)))
+	return credentialVerificationThrottled(delay)
+}
+
+func (l *credentialVerificationLimiter) cleanup(now time.Time) {
+	if now.Before(l.nextCleanup) {
+		return
+	}
+	l.nextCleanup = now.Add(credentialVerificationCleanupInterval)
+	for key, budget := range l.services {
+		if now.Sub(budget.lastUsed) >= credentialVerificationIdleTimeout {
+			delete(l.services, key)
+		}
+	}
+}
+
+func (l *credentialVerificationLimiter) close() {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	l.closed = true
+	l.services = nil
+}
+
+func credentialVerificationThrottled(delay time.Duration) error {
+	s := status.New(codes.ResourceExhausted, "too many credential verification attempts")
+	withRetry, err := s.WithDetails(&errdetails.RetryInfo{RetryDelay: durationpb.New(delay)})
+	if err != nil {
+		return s.Err()
+	}
+	return withRetry.Err()
+}

management/internals/shared/grpc/proxy_credential_limiter_test.go

@@ -0,0 +1,79 @@
+package grpc
+
+import (
+	"strconv"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/genproto/googleapis/rpc/errdetails"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+func TestCredentialVerificationRefillAndIsolation(t *testing.T) {
+	now := time.Now()
+	l := credentialVerificationLimiter{now: func() time.Time { return now }}
+	key := credentialVerificationKey{accountID: "account", serviceID: "service"}
+	for range credentialVerificationBurst {
+		require.NoError(t, l.allow(key))
+	}
+	err := l.allow(key)
+	require.Equal(t, codes.ResourceExhausted, status.Code(err), "the burst must be bounded")
+	now = now.Add(3 * time.Second)
+	err = l.allow(key)
+	require.Equal(t, codes.ResourceExhausted, status.Code(err), "a partially refilled token must not permit a check")
+	details := status.Convert(err).Details()
+	require.Len(t, details, 1, "throttling must provide RetryInfo")
+	retry, ok := details[0].(*errdetails.RetryInfo)
+	require.True(t, ok, "retry details must use the standard message")
+	assert.Equal(t, 3*time.Second, retry.RetryDelay.AsDuration(), "retry hint must reflect time until the next check")
+	now = now.Add(3 * time.Second)
+	require.NoError(t, l.allow(key))
+	assert.Equal(t, codes.ResourceExhausted, status.Code(l.allow(key)), "only one check must refill every six seconds")
+	require.NoError(t, l.allow(credentialVerificationKey{accountID: "other-account", serviceID: key.serviceID}))
+	require.NoError(t, l.allow(credentialVerificationKey{accountID: key.accountID, serviceID: "other-service"}))
+}
+
+func TestCredentialVerificationCapacityAndExpiry(t *testing.T) {
+	now := time.Now()
+	l := credentialVerificationLimiter{now: func() time.Time { return now }}
+	for i := range credentialVerificationMaxServices {
+		require.NoError(t, l.allow(credentialVerificationKey{accountID: "account", serviceID: credentialServiceID(strconv.Itoa(i))}))
+	}
+	key := credentialVerificationKey{accountID: "account", serviceID: "new-service"}
+	assert.Equal(t, codes.ResourceExhausted, status.Code(l.allow(key)), "capacity exhaustion must deny new checks")
+	now = now.Add(credentialVerificationIdleTimeout)
+	for range credentialVerificationBurst {
+		require.NoError(t, l.allow(key))
+	}
+	assert.Equal(t, codes.ResourceExhausted, status.Code(l.allow(key)), "expiry must retain the normal burst bound")
+}
+
+func TestCredentialVerificationConcurrentChecksAndClose(t *testing.T) {
+	var l credentialVerificationLimiter
+	key := credentialVerificationKey{accountID: "account", serviceID: "service"}
+	var admitted atomic.Int32
+	var wg sync.WaitGroup
+	for range 100 {
+		wg.Go(func() {
+			if err := l.allow(key); err == nil {
+				admitted.Add(1)
+			} else {
+				assert.Equal(t, codes.ResourceExhausted, status.Code(err), "excess checks must be throttled")
+			}
+		})
+	}
+	wg.Wait()
+	assert.EqualValues(t, credentialVerificationBurst, admitted.Load(), "concurrent checks must share the burst")
+	for range 10 {
+		wg.Go(l.close)
+		wg.Go(func() { assert.Error(t, l.allow(key)) })
+	}
+	wg.Wait()
+	assert.Empty(t, l.services, "closing must release retained budgets")
+	assert.Equal(t, codes.Unavailable, status.Code(l.allow(key)), "checks after close must fail closed")
+}

management/internals/shared/grpc/proxy_credentials.md

@@ -0,0 +1,18 @@
+# Reverse proxy credential verification
+
+The `ProxyService.Authenticate` RPC limits PIN and password checks before
+verifying their Argon2 hashes. Both methods share one budget per account and
+service: a burst of five checks, replenishing one check every six seconds
+(ten per minute). Successful and failed checks consume the budget. Account
+scope and service lookup run before the limiter.
+
+Excess checks receive gRPC `ResourceExhausted` with a standard `RetryInfo` delay.
+Updated proxies translate it to HTTP 429 and `Retry-After`. Older proxies show
+an authentication-service error but cannot bypass the Management limit.
+
+Budgets are held in memory per Management process and reset on restart. Proxy
+replicas reaching the same Management process share its budgets. Multiple
+Management processes have independent budgets; this is not a cluster-wide
+limit. At most 4,096 service budgets are retained, with idle entries expiring
+after fifteen minutes. Capacity exhaustion denies new checks until entries
+expire. Closing the server releases the retained state.

management/internals/shared/grpc/proxy_credentials_test.go

@@ -0,0 +1,131 @@
+package grpc_test
+
+import (
+	"context"
+	"net"
+	"net/netip"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/genproto/googleapis/rpc/errdetails"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/peer"
+	"google.golang.org/grpc/status"
+
+	"github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service"
+	servicemanager "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service/manager"
+	"github.com/netbirdio/netbird/management/internals/modules/reverseproxy/sessionkey"
+	nbgrpc "github.com/netbirdio/netbird/management/internals/shared/grpc"
+	"github.com/netbirdio/netbird/management/server/store"
+	"github.com/netbirdio/netbird/management/server/types"
+	"github.com/netbirdio/netbird/shared/management/proto"
+)
+
+func credentialServer(t *testing.T) (*nbgrpc.ProxyServiceServer, context.Context, grpc.UnaryServerInterceptor) {
+	t.Helper()
+	ctx := context.Background()
+	s, err := store.NewStore(ctx, types.SqliteStoreEngine, t.TempDir(), nil, false)
+	require.NoError(t, err)
+	t.Cleanup(func() { assert.NoError(t, s.Close(ctx)) })
+	require.NoError(t, s.SaveAccount(ctx, &types.Account{Id: "account"}))
+	keys, err := sessionkey.GenerateKeyPair()
+	require.NoError(t, err)
+	for _, id := range []string{"service", "other-service"} {
+		svc := &service.Service{
+			ID: id, AccountID: "account", Name: id, Domain: id + ".example.com",
+			Enabled: true, SessionPrivateKey: keys.PrivateKey, SessionPublicKey: keys.PublicKey,
+			Auth: service.AuthConfig{
+				PinAuth:      &service.PINAuthConfig{Enabled: true, Pin: "842716"},
+				PasswordAuth: &service.PasswordAuthConfig{Enabled: true, Password: "test-password"},
+			},
+		}
+		require.NoError(t, svc.Auth.HashSecrets())
+		require.NoError(t, s.CreateService(ctx, svc))
+	}
+	account := "account"
+	token, err := types.CreateNewProxyAccessToken("test proxy", time.Hour, &account, "admin")
+	require.NoError(t, err)
+	require.NoError(t, s.SaveProxyAccessToken(ctx, &token.ProxyAccessToken))
+	ctx = metadata.NewIncomingContext(ctx, metadata.Pairs("authorization", "Bearer "+string(token.PlainToken)))
+	ctx = peer.NewContext(ctx, &peer.Peer{Addr: net.TCPAddrFromAddrPort(netip.MustParseAddrPort("192.0.2.1:443"))})
+	server := nbgrpc.NewProxyServiceServer(nil, nil, nil, nbgrpc.ProxyOIDCConfig{}, nil, nil, nil, nil, nil)
+	t.Cleanup(server.Close)
+	server.SetServiceManager(servicemanager.NewManager(s, nil, nil, nil, nil, nil))
+	interceptor, _, closeInterceptor := nbgrpc.NewProxyAuthInterceptors(s)
+	t.Cleanup(closeInterceptor)
+	return server, ctx, interceptor
+}
+
+func TestAuthenticateCredentialRateLimit(t *testing.T) {
+	server, ctx, interceptor := credentialServer(t)
+	authenticate := func(req *proto.AuthenticateRequest) (*proto.AuthenticateResponse, error) {
+		response, err := interceptor(ctx, req, &grpc.UnaryServerInfo{FullMethod: "/management.ProxyService/Authenticate"}, func(ctx context.Context, req any) (any, error) {
+			return server.Authenticate(ctx, req.(*proto.AuthenticateRequest))
+		})
+		if err != nil {
+			return nil, err
+		}
+		return response.(*proto.AuthenticateResponse), nil
+	}
+	for i := range 5 {
+		req := &proto.AuthenticateRequest{AccountId: "account", Id: "service"}
+		if i%2 == 0 {
+			req.Request = &proto.AuthenticateRequest_Pin{Pin: &proto.PinRequest{Pin: "000000"}}
+		} else {
+			req.Request = &proto.AuthenticateRequest_Password{Password: &proto.PasswordRequest{Password: "wrong-password"}}
+		}
+		resp, err := authenticate(req)
+		require.NoError(t, err)
+		assert.False(t, resp.GetSuccess(), "incorrect PINs and passwords must be denied")
+		assert.Empty(t, resp.GetSessionToken(), "incorrect credentials must not issue a token")
+	}
+	req := &proto.AuthenticateRequest{AccountId: "account", Id: "service", Request: &proto.AuthenticateRequest_Pin{Pin: &proto.PinRequest{Pin: "842716"}}}
+	resp, err := authenticate(req)
+	assert.Nil(t, resp, "a throttled verification must not return a session")
+	require.Equal(t, codes.ResourceExhausted, status.Code(err), "PIN and password checks must share a service budget even with a valid proxy token")
+	details := status.Convert(err).Details()
+	require.Len(t, details, 1, "throttled responses must include a retry hint")
+	retry, ok := details[0].(*errdetails.RetryInfo)
+	require.True(t, ok, "the hint must use the standard RetryInfo message")
+	assert.Positive(t, retry.RetryDelay.AsDuration(), "the retry delay must be positive")
+	assert.LessOrEqual(t, retry.RetryDelay.AsDuration(), 6*time.Second, "the service must replenish one verification every six seconds")
+	req.AccountId = "another-account"
+	_, err = authenticate(req)
+	assert.Equal(t, codes.PermissionDenied, status.Code(err), "account scope must still be enforced before throttling")
+	req.AccountId = "account"
+	req.Id = "other-service"
+	resp, err = authenticate(req)
+	require.NoError(t, err)
+	assert.True(t, resp.GetSuccess(), "one service's throttle must not block another service")
+	assert.NotEmpty(t, resp.GetSessionToken(), "valid credentials on another service must issue a session")
+}
+
+func TestAuthenticateCredentialConcurrentLimit(t *testing.T) {
+	server, _, _ := credentialServer(t)
+	req := &proto.AuthenticateRequest{AccountId: "account", Id: "service", Request: &proto.AuthenticateRequest_Pin{Pin: &proto.PinRequest{Pin: "000000"}}}
+	var checked, throttled atomic.Int32
+	var wg sync.WaitGroup
+	for range 20 {
+		wg.Go(func() {
+			resp, err := server.Authenticate(context.Background(), req)
+			switch status.Code(err) {
+			case codes.OK:
+				checked.Add(1)
+				assert.False(t, resp.GetSuccess(), "incorrect credentials must be denied")
+			case codes.ResourceExhausted:
+				throttled.Add(1)
+			default:
+				assert.NoError(t, err)
+			}
+		})
+	}
+	wg.Wait()
+	assert.EqualValues(t, 5, checked.Load(), "only the burst budget may reach concurrent credential verification")
+	assert.EqualValues(t, 15, throttled.Load(), "excess concurrent checks must be throttled")
+}