From 5aa2a748d7e2dd889733d7fd0ed5e519dbf582c1 Mon Sep 17 00:00:00 2001 From: pascal Date: Thu, 16 Jul 2026 01:22:30 +0200 Subject: [PATCH] use native grpc + unify token refresh --- go.mod | 1 + go.sum | 3 + management/internals/server/boot.go | 9 +- management/internals/server/server.go | 123 ++++++++++++++++-- management/internals/shared/grpc/token_mgr.go | 99 +++++++------- .../internals/shared/grpc/token_mgr_test.go | 31 +++-- 6 files changed, 198 insertions(+), 68 deletions(-) diff --git a/go.mod b/go.mod index b90b68446..a0cdec43f 100644 --- a/go.mod +++ b/go.mod @@ -103,6 +103,7 @@ require ( github.com/rs/xid v1.3.0 github.com/shirou/gopsutil/v4 v4.25.8 github.com/skratchdot/open-golang v0.0.0-20200116055534-eef842397966 + github.com/soheilhy/cmux v0.1.5 github.com/songgao/water v0.0.0-20200317203138-2b4b6d7c09d8 github.com/stretchr/testify v1.11.1 github.com/testcontainers/testcontainers-go v0.37.0 diff --git a/go.sum b/go.sum index ad43dc109..b1779ca88 100644 --- a/go.sum +++ b/go.sum @@ -603,6 +603,8 @@ github.com/sirupsen/logrus v1.9.4 h1:TsZE7l11zFCLZnZ+teH4Umoq5BhEIfIzfRDZ1Uzql2w github.com/sirupsen/logrus v1.9.4/go.mod h1:ftWc9WdOfJ0a92nsE2jF5u5ZwH8Bv2zdeOC42RjbV2g= github.com/skratchdot/open-golang v0.0.0-20200116055534-eef842397966 h1:JIAuq3EEf9cgbU6AtGPK4CTG3Zf6CKMNqf0MHTggAUA= github.com/skratchdot/open-golang v0.0.0-20200116055534-eef842397966/go.mod h1:sUM3LWHvSMaG192sy56D9F7CNvL7jUJVXoqM1QKLnog= +github.com/soheilhy/cmux v0.1.5 h1:jjzc5WVemNEDTLwv9tlmemhC73tI08BNOIGwBOo10Js= +github.com/soheilhy/cmux v0.1.5/go.mod h1:T7TcVDs9LWfQgPlPsdngu6I6QIoyIFZDDC6sNE1GqG0= github.com/songgao/water v0.0.0-20200317203138-2b4b6d7c09d8 h1:TG/diQgUe0pntT/2D9tmUCz4VNwm9MfrtPr0SU2qSX8= github.com/songgao/water v0.0.0-20200317203138-2b4b6d7c09d8/go.mod h1:P5HUIBuIWKbyjl083/loAegFkfbFNx5i2qEP4CNbm7E= github.com/spf13/cast v1.10.0 h1:h2x0u2shc1QuLHfxi+cTJvs30+ZAHOGRic8uyGTDWxY= @@ -757,6 +759,7 @@ golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLL golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20200520004742-59133d7f0dd7/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A= golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= +golang.org/x/net v0.0.0-20201202161906-c7110b5ffcbb/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM= golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= diff --git a/management/internals/server/boot.go b/management/internals/server/boot.go index 1c78af9d0..e5be2a0f1 100644 --- a/management/internals/server/boot.go +++ b/management/internals/server/boot.go @@ -24,13 +24,13 @@ import ( "github.com/netbirdio/netbird/encryption" "github.com/netbirdio/netbird/formatter/hook" + "github.com/netbirdio/netbird/management/internals/modules/agentnetwork" "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/accesslogs" accesslogsmanager "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/accesslogs/manager" rpservice "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service" nbgrpc "github.com/netbirdio/netbird/management/internals/shared/grpc" "github.com/netbirdio/netbird/management/server/activity" activitystore "github.com/netbirdio/netbird/management/server/activity/store" - "github.com/netbirdio/netbird/management/internals/modules/agentnetwork" nbcache "github.com/netbirdio/netbird/management/server/cache" nbContext "github.com/netbirdio/netbird/management/server/context" nbhttp "github.com/netbirdio/netbird/management/server/http" @@ -184,7 +184,12 @@ func (s *BaseServer) GRPCServer() *grpc.Server { grpc.ChainStreamInterceptor(realip.StreamServerInterceptorOpts(realipOpts...), streamInterceptor, proxyStream), } - if s.Config.HttpConfig.LetsEncryptDomain != "" { + // With the native transport enabled, TLS is terminated at the listeners + // (cmux-split shared listener and legacy port), so transport credentials + // must not be set or the server would attempt a second handshake. + if nativeGRPCEnabled() { + log.Info("native gRPC transport enabled, TLS is terminated at the listeners") + } else if s.Config.HttpConfig.LetsEncryptDomain != "" { certManager, err := encryption.CreateCertManager(s.Config.Datadir, s.Config.HttpConfig.LetsEncryptDomain) if err != nil { log.Fatalf("failed to create certificate service: %v", err) diff --git a/management/internals/server/server.go b/management/internals/server/server.go index 7fd06d947..c0f0db012 100644 --- a/management/internals/server/server.go +++ b/management/internals/server/server.go @@ -6,12 +6,16 @@ import ( "fmt" "net" "net/http" + "os" + "slices" + "strconv" "strings" "sync" "time" "github.com/google/uuid" log "github.com/sirupsen/logrus" + "github.com/soheilhy/cmux" "go.opentelemetry.io/otel/metric" "golang.org/x/crypto/acme/autocert" "golang.org/x/net/http2" @@ -36,8 +40,18 @@ const ( DefaultSelfHostedDomain = "netbird.selfhosted" ContainerKeyBaseServer = "baseServer" + + // NativeGRPCEnvVar enables serving gRPC on the native gRPC transport, + // multiplexed with HTTP on the shared listener, instead of through the + // net/http ServeHTTP path which costs two extra goroutines per stream. + NativeGRPCEnvVar = "NB_MGMT_NATIVE_GRPC" ) +func nativeGRPCEnabled() bool { + enabled, _ := strconv.ParseBool(os.Getenv(NativeGRPCEnvVar)) + return enabled +} + type Server interface { Start(ctx context.Context) error Stop() error @@ -182,11 +196,22 @@ func (s *BaseServer) Start(ctx context.Context) error { } } + // With the native transport enabled the gRPC server carries no transport + // credentials, so TLS must be terminated at each of its listeners. + var grpcTLSConfig *tls.Config + if nativeGRPCEnabled() { + if s.certManager != nil { + grpcTLSConfig = s.certManager.TLSConfig() + } else { + grpcTLSConfig = tlsConfig + } + } + var compatListener net.Listener if s.mgmtPort != ManagementLegacyPort && !s.disableLegacyManagementPort { // The Management gRPC server was running on port 33073 previously. Old agents that are already connected to it // are using port 33073. For compatibility purposes we keep running a 2nd gRPC server on port 33073. - compatListener, err = s.serveGRPC(srvCtx, s.GRPCServer(), ManagementLegacyPort) + compatListener, err = s.serveGRPC(srvCtx, s.GRPCServer(), ManagementLegacyPort, grpcTLSConfig) if err != nil { return err } @@ -196,22 +221,38 @@ func (s *BaseServer) Start(ctx context.Context) error { rootHandler := s.handlerFunc(srvCtx, s.GRPCServer(), s.APIHandler(), s.IDPHandler(), s.Metrics().GetMeter()) switch { case s.certManager != nil: - // a call to certManager.Listener() always creates a new listener so we do it once - cml := s.certManager.Listener() if s.mgmtPort == 443 { // CertManager, HTTP and gRPC API all on the same port rootHandler = s.certManager.HTTPHandler(rootHandler) - s.listener = cml + if nativeGRPCEnabled() { + var tcpListener net.Listener + tcpListener, err = net.Listen("tcp", fmt.Sprintf(":%d", s.mgmtPort)) + if err != nil { + return fmt.Errorf("failed creating TCP listener on port %d: %v", s.mgmtPort, err) + } + s.listener = tls.NewListener(tcpListener, preferHTTP1ForDualProtoClients(s.certManager.TLSConfig())) + } else { + s.listener = s.certManager.Listener() + } } else { - s.listener, err = tls.Listen("tcp", fmt.Sprintf(":%d", s.mgmtPort), s.certManager.TLSConfig()) + mgmtTLSConfig := s.certManager.TLSConfig() + if nativeGRPCEnabled() { + mgmtTLSConfig = preferHTTP1ForDualProtoClients(mgmtTLSConfig) + } + s.listener, err = tls.Listen("tcp", fmt.Sprintf(":%d", s.mgmtPort), mgmtTLSConfig) if err != nil { return fmt.Errorf("failed creating TLS listener on port %d: %v", s.mgmtPort, err) } + cml := s.certManager.Listener() log.WithContext(ctx).Infof("running HTTP server (LetsEncrypt challenge handler): %s", cml.Addr().String()) s.serveHTTP(ctx, cml, s.certManager.HTTPHandler(nil)) } case tlsConfig != nil: - s.listener, err = tls.Listen("tcp", fmt.Sprintf(":%d", s.mgmtPort), tlsConfig) + mgmtTLSConfig := tlsConfig + if nativeGRPCEnabled() { + mgmtTLSConfig = preferHTTP1ForDualProtoClients(mgmtTLSConfig) + } + s.listener, err = tls.Listen("tcp", fmt.Sprintf(":%d", s.mgmtPort), mgmtTLSConfig) if err != nil { return fmt.Errorf("failed creating TLS listener on port %d: %v", s.mgmtPort, err) } @@ -224,7 +265,12 @@ func (s *BaseServer) Start(ctx context.Context) error { log.WithContext(ctx).Infof("management server version %s", version.NetbirdVersion()) log.WithContext(ctx).Infof("running HTTP server and gRPC server on the same port: %s", s.listener.Addr().String()) - s.serveGRPCWithHTTP(ctx, s.listener, rootHandler, tlsEnabled) + if nativeGRPCEnabled() { + log.WithContext(ctx).Infof("serving gRPC on the native transport (multiplexed with HTTP)") + s.serveMultiplexed(ctx, s.listener, s.GRPCServer(), rootHandler, tlsEnabled) + } else { + s.serveGRPCWithHTTP(ctx, s.listener, rootHandler, tlsEnabled) + } s.update = version.NewUpdateAndStart("nb/management") s.update.SetDaemonVersion(version.NetbirdVersion()) @@ -331,11 +377,14 @@ func (s *BaseServer) handlerFunc(_ context.Context, gRPCHandler *grpc.Server, ht }) } -func (s *BaseServer) serveGRPC(ctx context.Context, grpcServer *grpc.Server, port int) (net.Listener, error) { +func (s *BaseServer) serveGRPC(ctx context.Context, grpcServer *grpc.Server, port int, tlsConf *tls.Config) (net.Listener, error) { listener, err := net.Listen("tcp", fmt.Sprintf(":%d", port)) if err != nil { return nil, err } + if tlsConf != nil { + listener = tls.NewListener(listener, tlsConf) + } s.wg.Add(1) go func() { @@ -399,6 +448,64 @@ func (s *BaseServer) serveGRPCWithHTTP(ctx context.Context, listener net.Listene }() } +// preferHTTP1ForDualProtoClients steers TLS clients that offer both "h2" and +// "http/1.1" in ALPN (browsers, REST clients) to HTTP/1.1. gRPC clients offer +// only "h2", so with this steering every HTTP/2 connection on the shared +// listener carries gRPC and can be routed to the native transport without +// inspecting frames. ACME "acme-tls/1" and single-protocol clients keep the +// base configuration. +func preferHTTP1ForDualProtoClients(base *tls.Config) *tls.Config { + h1Config := base.Clone() + h1Config.NextProtos = []string{"http/1.1"} + steered := base.Clone() + steered.GetConfigForClient = func(hello *tls.ClientHelloInfo) (*tls.Config, error) { + if slices.Contains(hello.SupportedProtos, "http/1.1") && slices.Contains(hello.SupportedProtos, "h2") { + return h1Config, nil + } + return nil, nil + } + return steered +} + +// serveMultiplexed splits the shared listener by protocol: HTTP/2 connections +// go to the native gRPC transport (see preferHTTP1ForDualProtoClients for why +// they are all gRPC), everything else is served by net/http. +func (s *BaseServer) serveMultiplexed(ctx context.Context, listener net.Listener, grpcServer *grpc.Server, handler http.Handler, tlsEnabled bool) { + mux := cmux.New(listener) + grpcListener := mux.Match(cmux.HTTP2()) + httpListener := mux.Match(cmux.Any()) + + httpHandler := handler + if !tlsEnabled { + //nolint:staticcheck // h2c also handles the HTTP/1 Upgrade mechanism, which http.Server's UnencryptedHTTP2 does not + httpHandler = h2c.NewHandler(handler, &http2.Server{}) + } + + s.wg.Add(3) + go func() { + defer s.wg.Done() + s.reportServeError(ctx, grpcServer.Serve(grpcListener)) + }() + go func() { + defer s.wg.Done() + s.reportServeError(ctx, http.Serve(httpListener, httpHandler)) + }() + go func() { + defer s.wg.Done() + s.reportServeError(ctx, mux.Serve()) + }() +} + +func (s *BaseServer) reportServeError(ctx context.Context, err error) { + if ctx.Err() != nil || err == nil { + return + } + select { + case s.errCh <- err: + default: + } +} + // ResolveDomains determines dnsDomain and mgmtSingleAccModeDomain based on store state. // Fresh installs use the default self-hosted domain, while existing installs reuse the // persisted account domain to keep addressing stable across config changes. diff --git a/management/internals/shared/grpc/token_mgr.go b/management/internals/shared/grpc/token_mgr.go index fb2d83a9a..14d9088e5 100644 --- a/management/internals/shared/grpc/token_mgr.go +++ b/management/internals/shared/grpc/token_mgr.go @@ -43,8 +43,9 @@ type TimeBasedAuthSecretsManager struct { updateManager network_map.PeersUpdateManager settingsManager settings.Manager groupsManager groups.Manager - turnCancelMap map[string]chan struct{} - relayCancelMap map[string]chan struct{} + scheduler *refreshScheduler + turnJobs map[string]*refreshJob + relayJobs map[string]*refreshJob wgKey wgtypes.Key } @@ -60,8 +61,6 @@ func NewTimeBasedAuthSecretsManager(updateManager network_map.PeersUpdateManager updateManager: updateManager, turnCfg: turnCfg, relayCfg: relayCfg, - turnCancelMap: make(map[string]chan struct{}), - relayCancelMap: make(map[string]chan struct{}), settingsManager: settingsManager, groupsManager: groupsManager, wgKey: key, @@ -127,16 +126,16 @@ func (m *TimeBasedAuthSecretsManager) GenerateRelayToken() (*Token, error) { } func (m *TimeBasedAuthSecretsManager) cancelTURN(peerID string) { - if channel, ok := m.turnCancelMap[peerID]; ok { - close(channel) - delete(m.turnCancelMap, peerID) + if job, ok := m.turnJobs[peerID]; ok { + m.scheduler.cancel(job) + delete(m.turnJobs, peerID) } } func (m *TimeBasedAuthSecretsManager) cancelRelay(peerID string) { - if channel, ok := m.relayCancelMap[peerID]; ok { - close(channel) - delete(m.relayCancelMap, peerID) + if job, ok := m.relayJobs[peerID]; ok { + m.scheduler.cancel(job) + delete(m.relayJobs, peerID) } } @@ -148,6 +147,30 @@ func (m *TimeBasedAuthSecretsManager) CancelRefresh(peerID string) { m.cancelRelay(peerID) } +func (m *TimeBasedAuthSecretsManager) ensureScheduler() { + if m.scheduler == nil { + m.scheduler = newRefreshScheduler(m.runRefreshJob) + m.turnJobs = make(map[string]*refreshJob) + m.relayJobs = make(map[string]*refreshJob) + } +} + +func (m *TimeBasedAuthSecretsManager) runRefreshJob(job *refreshJob) { + switch job.kind { + case refreshKindTURN: + m.pushNewTURNAndRelayTokens(job.ctx, job.accountID, job.peerID) + case refreshKindRelay: + m.pushNewRelayTokens(job.ctx, job.accountID, job.peerID) + } +} + +func refreshInterval(ttl time.Duration) time.Duration { + if ttl <= 0 { + ttl = defaultDuration + } + return ttl / 4 * 3 +} + // SetupRefresh starts peer credentials refresh func (m *TimeBasedAuthSecretsManager) SetupRefresh(ctx context.Context, accountID, peerID string) { m.mux.Lock() @@ -157,54 +180,38 @@ func (m *TimeBasedAuthSecretsManager) SetupRefresh(ctx context.Context, accountI m.cancelRelay(peerID) if m.turnCfg != nil && m.turnCfg.TimeBasedCredentials { - turnCancel := make(chan struct{}, 1) - m.turnCancelMap[peerID] = turnCancel - go m.refreshTURNTokens(ctx, accountID, peerID, turnCancel) + m.ensureScheduler() + job := &refreshJob{ + ctx: ctx, + accountID: accountID, + peerID: peerID, + kind: refreshKindTURN, + interval: refreshInterval(m.turnCfg.CredentialsTTL.Duration), + } + m.turnJobs[peerID] = job + m.scheduler.schedule(job) log.WithContext(ctx).Debugf("starting TURN refresh for %s", peerID) } else { log.WithContext(ctx).Debugf("no TURN configuration, skipping TURN refresh for %s", peerID) } if m.relayCfg != nil { - relayCancel := make(chan struct{}, 1) - m.relayCancelMap[peerID] = relayCancel - go m.refreshRelayTokens(ctx, accountID, peerID, relayCancel) + m.ensureScheduler() + job := &refreshJob{ + ctx: ctx, + accountID: accountID, + peerID: peerID, + kind: refreshKindRelay, + interval: refreshInterval(m.relayCfg.CredentialsTTL.Duration), + } + m.relayJobs[peerID] = job + m.scheduler.schedule(job) log.WithContext(ctx).Tracef("starting relay refresh for %s", peerID) } else { log.WithContext(ctx).Tracef("no relay configuration, skipping relay refresh for %s", peerID) } } -func (m *TimeBasedAuthSecretsManager) refreshTURNTokens(ctx context.Context, accountID, peerID string, cancel chan struct{}) { - ticker := time.NewTicker(m.turnCfg.CredentialsTTL.Duration / 4 * 3) - defer ticker.Stop() - - for { - select { - case <-cancel: - log.WithContext(ctx).Tracef("stopping TURN refresh for %s", peerID) - return - case <-ticker.C: - m.pushNewTURNAndRelayTokens(ctx, accountID, peerID) - } - } -} - -func (m *TimeBasedAuthSecretsManager) refreshRelayTokens(ctx context.Context, accountID, peerID string, cancel chan struct{}) { - ticker := time.NewTicker(m.relayCfg.CredentialsTTL.Duration / 4 * 3) - defer ticker.Stop() - - for { - select { - case <-cancel: - log.WithContext(ctx).Tracef("stopping relay refresh for %s", peerID) - return - case <-ticker.C: - m.pushNewRelayTokens(ctx, accountID, peerID) - } - } -} - func (m *TimeBasedAuthSecretsManager) pushNewTURNAndRelayTokens(ctx context.Context, accountID, peerID string) { turnToken, err := m.turnHmacToken.GenerateToken(sha1.New) if err != nil { diff --git a/management/internals/shared/grpc/token_mgr_test.go b/management/internals/shared/grpc/token_mgr_test.go index 98eb66fb5..1c1ccd523 100644 --- a/management/internals/shared/grpc/token_mgr_test.go +++ b/management/internals/shared/grpc/token_mgr_test.go @@ -112,12 +112,12 @@ func TestTimeBasedAuthSecretsManager_SetupRefresh(t *testing.T) { tested.SetupRefresh(ctx, "someAccountID", peer) - if _, ok := tested.turnCancelMap[peer]; !ok { - t.Errorf("expecting peer to be present in the turn cancel map, got not present") + if _, ok := tested.turnJobs[peer]; !ok { + t.Errorf("expecting peer to be present in the turn jobs map, got not present") } - if _, ok := tested.relayCancelMap[peer]; !ok { - t.Errorf("expecting peer to be present in the relay cancel map, got not present") + if _, ok := tested.relayJobs[peer]; !ok { + t.Errorf("expecting peer to be present in the relay jobs map, got not present") } var updates []*network_map.UpdateMessage @@ -212,19 +212,26 @@ func TestTimeBasedAuthSecretsManager_CancelRefresh(t *testing.T) { require.NoError(t, err) tested.SetupRefresh(context.Background(), "someAccountID", peer) - if _, ok := tested.turnCancelMap[peer]; !ok { - t.Errorf("expecting peer to be present in turn cancel map, got not present") + if _, ok := tested.turnJobs[peer]; !ok { + t.Errorf("expecting peer to be present in turn jobs map, got not present") } - if _, ok := tested.relayCancelMap[peer]; !ok { - t.Errorf("expecting peer to be present in relay cancel map, got not present") + if _, ok := tested.relayJobs[peer]; !ok { + t.Errorf("expecting peer to be present in relay jobs map, got not present") } tested.CancelRefresh(peer) - if _, ok := tested.turnCancelMap[peer]; ok { - t.Errorf("expecting peer to be not present in turn cancel map, got present") + if _, ok := tested.turnJobs[peer]; ok { + t.Errorf("expecting peer to be not present in turn jobs map, got present") } - if _, ok := tested.relayCancelMap[peer]; ok { - t.Errorf("expecting peer to be not present in relay cancel map, got present") + if _, ok := tested.relayJobs[peer]; ok { + t.Errorf("expecting peer to be not present in relay jobs map, got present") + } + + tested.scheduler.mu.Lock() + heapLen := len(tested.scheduler.jobs) + tested.scheduler.mu.Unlock() + if heapLen != 0 { + t.Errorf("expecting scheduler heap to be empty after cancel, got %d entries", heapLen) } }