From 771d81b72ab8be9da5857d0e2e1f734dc4a70b11 Mon Sep 17 00:00:00 2001 From: Pascal Fischer <32096965+pascal-fischer@users.noreply.github.com> Date: Mon, 21 Sep 2026 15:55:56 +0200 Subject: [PATCH] [management] Add proxy credentials limiter on management (#7569) --- management/internals/shared/grpc/proxy.go | 15 +- .../shared/grpc/proxy_credential_limiter.go | 101 ++++++++++++++ .../grpc/proxy_credential_limiter_test.go | 79 +++++++++++ .../shared/grpc/proxy_credentials.md | 18 +++ .../shared/grpc/proxy_credentials_test.go | 131 ++++++++++++++++++ 5 files changed, 342 insertions(+), 2 deletions(-) create mode 100644 management/internals/shared/grpc/proxy_credential_limiter.go create mode 100644 management/internals/shared/grpc/proxy_credential_limiter_test.go create mode 100644 management/internals/shared/grpc/proxy_credentials.md create mode 100644 management/internals/shared/grpc/proxy_credentials_test.go diff --git a/management/internals/shared/grpc/proxy.go b/management/internals/shared/grpc/proxy.go index 2fc969ad0..b1527f0ab 100644 --- a/management/internals/shared/grpc/proxy.go +++ b/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 diff --git a/management/internals/shared/grpc/proxy_credential_limiter.go b/management/internals/shared/grpc/proxy_credential_limiter.go new file mode 100644 index 000000000..2a3a02347 --- /dev/null +++ b/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() +} diff --git a/management/internals/shared/grpc/proxy_credential_limiter_test.go b/management/internals/shared/grpc/proxy_credential_limiter_test.go new file mode 100644 index 000000000..565b14889 --- /dev/null +++ b/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") +} diff --git a/management/internals/shared/grpc/proxy_credentials.md b/management/internals/shared/grpc/proxy_credentials.md new file mode 100644 index 000000000..1caf432e5 --- /dev/null +++ b/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. diff --git a/management/internals/shared/grpc/proxy_credentials_test.go b/management/internals/shared/grpc/proxy_credentials_test.go new file mode 100644 index 000000000..2e144bd69 --- /dev/null +++ b/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") +}