sendnrw/netbird
2
0
Fork 0
mirror of https://github.com/netbirdio/netbird.git synced 2026-07-18 04:29:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

use native grpc + unify token refresh

Browse Source
This commit is contained in:
pascal
2026-07-16 01:22:30 +02:00
parent e1a24376ab
commit 5aa2a748d7
6 changed files with 198 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
go.mod
View File

@@ -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

3
go.sum
View File

@@ -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=

9
management/internals/server/boot.go
View File

@@ -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)

123
management/internals/server/server.go
View File

@@ -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.

99
management/internals/shared/grpc/token_mgr.go
View File

@@ -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 {

31
management/internals/shared/grpc/token_mgr_test.go
View File

@@ -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)
}
}