Add signal flag

client/grpc/dialer.go

@@ -29,8 +29,7 @@ func Backoff(ctx context.Context) backoff.BackOff {
 // The component parameter specifies the WebSocket proxy component path (e.g., "/management", "/signal").
 func CreateConnection(ctx context.Context, addr string, tlsEnabled bool, component string) (*grpc.ClientConn, error) {
 	transportOption := grpc.WithTransportCredentials(insecure.NewCredentials())
-	// for js, the outer websocket layer takes care of tls
-	if tlsEnabled && runtime.GOOS != "js" {
+	if tlsEnabled {
 		certPool, err := x509.SystemCertPool()
 		if err != nil || certPool == nil {
 			log.Debugf("System cert pool not available; falling back to embedded cert, error: %v", err)
@@ -38,7 +37,9 @@ func CreateConnection(ctx context.Context, addr string, tlsEnabled bool, compone
 		}
 
 		transportOption = grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{
-			RootCAs: certPool,
+			// for js, outer websocket layer takes care of tls verification via WithCustomDialer
+			InsecureSkipVerify: runtime.GOOS == "js",
+			RootCAs:            certPool,
 		}))
 	}
 

infrastructure_files/docker-compose.yml.tmpl.traefik

@@ -47,7 +47,7 @@ services:
     - traefik.enable=true
     - traefik.http.routers.netbird-wsproxy-signal.rule=Host(`$NETBIRD_DOMAIN`) && PathPrefix(`/ws-proxy/signal`)
     - traefik.http.routers.netbird-wsproxy-signal.service=netbird-wsproxy-signal
-    - traefik.http.services.netbird-wsproxy-signal.loadbalancer.server.port=80
+    - traefik.http.services.netbird-wsproxy-signal.loadbalancer.server.port=10000
     - traefik.http.routers.netbird-signal.rule=Host(`$NETBIRD_DOMAIN`) && PathPrefix(`/signalexchange.SignalExchange/`)
     - traefik.http.services.netbird-signal.loadbalancer.server.port=10000
     - traefik.http.services.netbird-signal.loadbalancer.server.scheme=h2c

infrastructure_files/getting-started-with-zitadel.sh

@@ -621,7 +621,7 @@ renderCaddyfile() {
     # relay
     reverse_proxy /relay* relay:80
     # Signal
-    reverse_proxy /ws-proxy/signal* signal:80
+    reverse_proxy /ws-proxy/signal* signal:10000
     reverse_proxy /signalexchange.SignalExchange/* h2c://signal:10000
     # Management
     reverse_proxy /api/* management:80

management/cmd/defaults.go

@@ -1,5 +1,11 @@
 package cmd
 
+import (
+	"fmt"
+
+	"github.com/netbirdio/netbird/management/internals/server"
+)
+
 const (
 	defaultMgmtDataDir   = "/var/lib/netbird/"
 	defaultMgmtConfigDir = "/etc/netbird"
@@ -16,3 +22,5 @@ const (
 
 	defaultSingleAccModeDomain = "netbird.selfhosted"
 )
+
+var defaultWSProxyAddr = fmt.Sprintf("127.0.0.1:%d", server.ManagementLegacyPort)

management/cmd/management.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"io"
 	"io/fs"
+	"net"
 	"net/http"
 	"net/url"
 	"os"
@@ -26,11 +27,11 @@ import (
 	"github.com/netbirdio/netbird/util"
 )
 
-var newServer = func(config *nbconfig.Config, dnsDomain, mgmtSingleAccModeDomain string, mgmtPort int, mgmtMetricsPort int, disableMetrics, disableGeoliteUpdate, userDeleteFromIDPEnabled bool) server.Server {
-	return server.NewServer(config, dnsDomain, mgmtSingleAccModeDomain, mgmtPort, mgmtMetricsPort, disableMetrics, disableGeoliteUpdate, userDeleteFromIDPEnabled)
+var newServer = func(config *nbconfig.Config, dnsDomain, mgmtSingleAccModeDomain string, mgmtPort int, mgmtMetricsPort int, disableMetrics, disableGeoliteUpdate, userDeleteFromIDPEnabled bool, wsProxyAddr string) server.Server {
+	return server.NewServer(config, dnsDomain, mgmtSingleAccModeDomain, mgmtPort, mgmtMetricsPort, disableMetrics, disableGeoliteUpdate, userDeleteFromIDPEnabled, wsProxyAddr)
 }
 
-func SetNewServer(fn func(config *nbconfig.Config, dnsDomain, mgmtSingleAccModeDomain string, mgmtPort int, mgmtMetricsPort int, disableMetrics, disableGeoliteUpdate, userDeleteFromIDPEnabled bool) server.Server) {
+func SetNewServer(fn func(config *nbconfig.Config, dnsDomain, mgmtSingleAccModeDomain string, mgmtPort int, mgmtMetricsPort int, disableMetrics, disableGeoliteUpdate, userDeleteFromIDPEnabled bool, wsProxyAddr string) server.Server) {
 	newServer = fn
 }
 
@@ -82,6 +83,10 @@ var (
 				return fmt.Errorf("failed parsing the provided dns-domain. Valid status: %t, Length: %d", valid, len(dnsDomain))
 			}
 
+			if _, _, err := net.SplitHostPort(wsProxyAddr); err != nil {
+				return fmt.Errorf("invalid ws-proxy-backend-addr format: %w", err)
+			}
+
 			return nil
 		},
 		RunE: func(cmd *cobra.Command, args []string) error {
@@ -108,7 +113,7 @@ var (
 				mgmtSingleAccModeDomain = ""
 			}
 
-			srv := newServer(config, dnsDomain, mgmtSingleAccModeDomain, mgmtPort, mgmtMetricsPort, disableMetrics, disableGeoliteUpdate, userDeleteFromIDPEnabled)
+			srv := newServer(config, dnsDomain, mgmtSingleAccModeDomain, mgmtPort, mgmtMetricsPort, disableMetrics, disableGeoliteUpdate, userDeleteFromIDPEnabled, wsProxyAddr)
 			go func() {
 				if err := srv.Start(cmd.Context()); err != nil {
 					log.Fatalf("Server error: %v", err)

management/cmd/root.go

@@ -31,6 +31,7 @@ var (
 	mgmtSingleAccModeDomain  string
 	certFile                 string
 	certKey                  string
+	wsProxyAddr              string
 
 	rootCmd = &cobra.Command{
 		Use:          "netbird-mgmt",
@@ -57,6 +58,7 @@ func init() {
 	mgmtCmd.Flags().IntVar(&mgmtPort, "port", 80, "server port to listen on (defaults to 443 if TLS is enabled, 80 otherwise")
 	mgmtCmd.Flags().IntVar(&mgmtMetricsPort, "metrics-port", 9090, "metrics endpoint http port. Metrics are accessible under host:metrics-port/metrics")
 	mgmtCmd.Flags().StringVar(&mgmtDataDir, "datadir", defaultMgmtDataDir, "server data directory location")
+	mgmtCmd.Flags().StringVar(&wsProxyAddr, "ws-proxy-backend-addr", defaultWSProxyAddr, "WebSocket proxy backend address in host:port format that the proxy will dial")
 	mgmtCmd.Flags().StringVar(&nbconfig.MgmtConfigPath, "config", defaultMgmtConfig, "Netbird config file location. Config params specified via command line (e.g. datadir) have a precedence over configuration from this file")
 	mgmtCmd.Flags().StringVar(&mgmtLetsencryptDomain, "letsencrypt-domain", "", "a domain to issue Let's Encrypt certificate for. Enables TLS using Let's Encrypt. Will fetch and renew certificate, and run the server with TLS")
 	mgmtCmd.Flags().StringVar(&mgmtSingleAccModeDomain, "single-account-mode-domain", defaultSingleAccModeDomain, "Enables single account mode. This means that all the users will be under the same account grouped by the specified domain. If the installation has more than one account, the property is ineffective. Enabled by default with the default domain "+defaultSingleAccModeDomain)

management/internals/server/server.go

@@ -57,6 +57,7 @@ type BaseServer struct {
 	mgmtSingleAccModeDomain  string
 	mgmtMetricsPort          int
 	mgmtPort                 int
+	wsProxyAddr              string
 
 	listener    net.Listener
 	certManager *autocert.Manager
@@ -68,7 +69,7 @@ type BaseServer struct {
 }
 
 // NewServer initializes and configures a new Server instance
-func NewServer(config *nbconfig.Config, dnsDomain, mgmtSingleAccModeDomain string, mgmtPort, mgmtMetricsPort int, disableMetrics, disableGeoliteUpdate, userDeleteFromIDPEnabled bool) *BaseServer {
+func NewServer(config *nbconfig.Config, dnsDomain, mgmtSingleAccModeDomain string, mgmtPort, mgmtMetricsPort int, disableMetrics, disableGeoliteUpdate, userDeleteFromIDPEnabled bool, wsProxyAddr string) *BaseServer {
 	return &BaseServer{
 		config:                   config,
 		container:                make(map[string]any),
@@ -79,6 +80,7 @@ func NewServer(config *nbconfig.Config, dnsDomain, mgmtSingleAccModeDomain strin
 		userDeleteFromIDPEnabled: userDeleteFromIDPEnabled,
 		mgmtPort:                 mgmtPort,
 		mgmtMetricsPort:          mgmtMetricsPort,
+		wsProxyAddr:              wsProxyAddr,
 	}
 }
 
@@ -181,6 +183,7 @@ 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())
+	log.WithContext(ctx).Infof("WebSocket proxy configured to forward to: %s", s.wsProxyAddr)
 	s.serveGRPCWithHTTP(ctx, s.listener, rootHandler, tlsEnabled)
 
 	s.update = version.NewUpdate("nb/management")
@@ -251,7 +254,7 @@ func updateMgmtConfig(ctx context.Context, path string, config *nbconfig.Config)
 }
 
 func (s *BaseServer) handlerFunc(gRPCHandler *grpc.Server, httpHandler http.Handler, meter metric.Meter) http.Handler {
-	wsProxy := wsproxyserver.New(gRPCHandler, wsproxyserver.WithOTelMeter(meter))
+	wsProxy := wsproxyserver.New(s.wsProxyAddr, wsproxyserver.WithOTelMeter(meter))
 
 	return http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
 		switch {

signal/cmd/run.go

@@ -45,6 +45,7 @@ var (
 	defaultSignalSSLDir     string
 	signalCertFile          string
 	signalCertKey           string
+	wsProxyBackendAddr      string
 
 	signalKaep = grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{
 		MinTime:             5 * time.Second,
@@ -78,6 +79,10 @@ var (
 				tlsEnabled = true
 			}
 
+			if _, _, err := net.SplitHostPort(wsProxyBackendAddr); err != nil {
+				return fmt.Errorf("invalid ws-proxy-backend-addr format: %w", err)
+			}
+
 			if !userPort {
 				// different defaults for signalPort
 				if tlsEnabled {
@@ -161,6 +166,7 @@ var (
 			}
 
 			log.Infof("signal server version %s", version.NetbirdVersion())
+			log.Infof("WebSocket proxy configured to forward to: %s", wsProxyBackendAddr)
 			log.Infof("started Signal Service")
 
 			SetupCloseHandler()
@@ -255,7 +261,7 @@ func startServerWithCertManager(certManager *autocert.Manager, grpcRootHandler h
 }
 
 func grpcHandlerFunc(grpcServer *grpc.Server, meter metric.Meter) http.Handler {
-	wsProxy := wsproxyserver.New(grpcServer, wsproxyserver.WithOTelMeter(meter))
+	wsProxy := wsproxyserver.New(wsProxyBackendAddr, wsproxyserver.WithOTelMeter(meter))
 
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		switch {
@@ -329,5 +335,6 @@ func init() {
 	runCmd.Flags().StringVar(&signalLetsencryptDomain, "letsencrypt-domain", "", "a domain to issue Let's Encrypt certificate for. Enables TLS using Let's Encrypt. Will fetch and renew certificate, and run the server with TLS")
 	runCmd.Flags().StringVar(&signalCertFile, "cert-file", "", "Location of your SSL certificate. Can be used when you have an existing certificate and don't want a new certificate be generated automatically. If letsencrypt-domain is specified this property has no effect")
 	runCmd.Flags().StringVar(&signalCertKey, "cert-key", "", "Location of your SSL certificate private key. Can be used when you have an existing certificate and don't want a new certificate be generated automatically. If letsencrypt-domain is specified this property has no effect")
+	runCmd.Flags().StringVar(&wsProxyBackendAddr, "ws-proxy-backend-addr", fmt.Sprintf("127.0.0.1:%d", legacyGRPCPort), "WebSocket proxy backend address in host:port format that the proxy will dial")
 	setFlagsFromEnvVars(runCmd)
 }

util/wsproxy/server/proxy.go

@@ -2,6 +2,7 @@ package server
 
 import (
 	"context"
+	"errors"
 	"io"
 	"net"
 	"net/http"
@@ -10,33 +11,32 @@ import (
 
 	"github.com/coder/websocket"
 	log "github.com/sirupsen/logrus"
-	"golang.org/x/net/http2"
 
 	"github.com/netbirdio/netbird/util/wsproxy"
 )
 
 const (
-	bufferSize = 32 * 1024
-	ioTimeout  = 5 * time.Second
+	dialTimeout = 10 * time.Second
+	bufferSize  = 32 * 1024
 )
 
 // Config contains the configuration for the WebSocket proxy.
 type Config struct {
-	Handler         http.Handler
+	LocalGRPCAddr   string
 	Path            string
 	MetricsRecorder MetricsRecorder
 }
 
-// Proxy handles WebSocket to gRPC handler proxying.
+// Proxy handles WebSocket to TCP proxying for gRPC connections.
 type Proxy struct {
 	config  Config
 	metrics MetricsRecorder
 }
 
 // New creates a new WebSocket proxy instance with optional configuration
-func New(handler http.Handler, opts ...Option) *Proxy {
+func New(localGRPCAddr string, opts ...Option) *Proxy {
 	config := Config{
-		Handler:         handler,
+		LocalGRPCAddr:   localGRPCAddr,
 		Path:            wsproxy.ProxyPath,
 		MetricsRecorder: NoOpMetricsRecorder{}, // Default to no-op
 	}
@@ -62,7 +62,7 @@ func (p *Proxy) handleWebSocket(w http.ResponseWriter, r *http.Request) {
 	p.metrics.RecordConnection(ctx)
 	defer p.metrics.RecordDisconnection(ctx)
 
-	log.Debugf("WebSocket proxy handling connection from %s, forwarding to internal gRPC handler", r.RemoteAddr)
+	log.Debugf("WebSocket proxy handling connection from %s, forwarding to %s", r.RemoteAddr, p.config.LocalGRPCAddr)
 	acceptOptions := &websocket.AcceptOptions{
 		OriginPatterns: []string{"*"},
 	}
@@ -74,41 +74,71 @@ func (p *Proxy) handleWebSocket(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	defer func() {
-		_ = wsConn.Close(websocket.StatusNormalClosure, "")
+		if err := wsConn.Close(websocket.StatusNormalClosure, ""); err != nil {
+			log.Debugf("Failed to close WebSocket: %v", err)
+		}
 	}()
 
-	clientConn, serverConn := net.Pipe()
+	log.Debugf("WebSocket proxy attempting to connect to local gRPC at %s", p.config.LocalGRPCAddr)
+	tcpConn, err := net.DialTimeout("tcp", p.config.LocalGRPCAddr, dialTimeout)
+	if err != nil {
+		p.metrics.RecordError(ctx, "tcp_dial_failed")
+		log.Warnf("Failed to connect to local gRPC server at %s: %v", p.config.LocalGRPCAddr, err)
+		if err := wsConn.Close(websocket.StatusInternalError, "Backend unavailable"); err != nil {
+			log.Debugf("Failed to close WebSocket after connection failure: %v", err)
+		}
+		return
+	}
 	defer func() {
-		_ = clientConn.Close()
-		_ = serverConn.Close()
+		if err := tcpConn.Close(); err != nil {
+			log.Debugf("Failed to close TCP connection: %v", err)
+		}
 	}()
 
-	log.Debugf("WebSocket proxy established: %s -> gRPC handler", r.RemoteAddr)
+	log.Debugf("WebSocket proxy established: client %s -> local gRPC %s", r.RemoteAddr, p.config.LocalGRPCAddr)
 
-	go func() {
-		(&http2.Server{}).ServeConn(serverConn, &http2.ServeConnOpts{
-			Context: ctx,
-			Handler: p.config.Handler,
-		})
-	}()
-
-	p.proxyData(ctx, wsConn, clientConn, r.RemoteAddr)
+	p.proxyData(ctx, wsConn, tcpConn)
 }
 
-func (p *Proxy) proxyData(ctx context.Context, wsConn *websocket.Conn, pipeConn net.Conn, clientAddr string) {
+func (p *Proxy) proxyData(ctx context.Context, wsConn *websocket.Conn, tcpConn net.Conn) {
 	proxyCtx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
 	var wg sync.WaitGroup
 	wg.Add(2)
 
-	go p.wsToPipe(proxyCtx, cancel, &wg, wsConn, pipeConn, clientAddr)
-	go p.pipeToWS(proxyCtx, cancel, &wg, wsConn, pipeConn, clientAddr)
+	go p.wsToTCP(proxyCtx, cancel, &wg, wsConn, tcpConn)
+	go p.tcpToWS(proxyCtx, cancel, &wg, wsConn, tcpConn)
 
-	wg.Wait()
+	done := make(chan struct{})
+	go func() {
+		wg.Wait()
+		close(done)
+	}()
+
+	select {
+	case <-done:
+		log.Tracef("Proxy data transfer completed, both goroutines terminated")
+	case <-proxyCtx.Done():
+		log.Tracef("Proxy data transfer cancelled, forcing connection closure")
+
+		if err := wsConn.Close(websocket.StatusGoingAway, "proxy cancelled"); err != nil {
+			log.Tracef("Error closing WebSocket during cancellation: %v", err)
+		}
+		if err := tcpConn.Close(); err != nil {
+			log.Tracef("Error closing TCP connection during cancellation: %v", err)
+		}
+
+		select {
+		case <-done:
+			log.Tracef("Goroutines terminated after forced connection closure")
+		case <-time.After(2 * time.Second):
+			log.Tracef("Goroutines did not terminate within timeout after connection closure")
+		}
+	}
 }
 
-func (p *Proxy) wsToPipe(ctx context.Context, cancel context.CancelFunc, wg *sync.WaitGroup, wsConn *websocket.Conn, pipeConn net.Conn, clientAddr string) {
+func (p *Proxy) wsToTCP(ctx context.Context, cancel context.CancelFunc, wg *sync.WaitGroup, wsConn *websocket.Conn, tcpConn net.Conn) {
 	defer wg.Done()
 	defer cancel()
 
@@ -117,73 +147,80 @@ func (p *Proxy) wsToPipe(ctx context.Context, cancel context.CancelFunc, wg *syn
 		if err != nil {
 			switch {
 			case ctx.Err() != nil:
-				log.Debugf("WebSocket from %s terminating due to context cancellation", clientAddr)
-			case websocket.CloseStatus(err) != -1:
-				log.Debugf("WebSocket from %s disconnected", clientAddr)
+				log.Debugf("wsToTCP goroutine terminating due to context cancellation")
+			case websocket.CloseStatus(err) == websocket.StatusNormalClosure:
+				log.Debugf("WebSocket closed normally")
 			default:
 				p.metrics.RecordError(ctx, "websocket_read_error")
-				log.Debugf("WebSocket read error from %s: %v", clientAddr, err)
+				log.Errorf("WebSocket read error: %v", err)
 			}
 			return
 		}
 
 		if msgType != websocket.MessageBinary {
-			log.Warnf("Unexpected WebSocket message type from %s: %v", clientAddr, msgType)
+			log.Warnf("Unexpected WebSocket message type: %v", msgType)
 			continue
 		}
 
 		if ctx.Err() != nil {
-			log.Tracef("wsToPipe goroutine terminating due to context cancellation before pipe write")
+			log.Tracef("wsToTCP goroutine terminating due to context cancellation before TCP write")
 			return
 		}
 
-		if err := pipeConn.SetWriteDeadline(time.Now().Add(ioTimeout)); err != nil {
-			log.Debugf("Failed to set pipe write deadline: %v", err)
+		if err := tcpConn.SetWriteDeadline(time.Now().Add(5 * time.Second)); err != nil {
+			log.Debugf("Failed to set TCP write deadline: %v", err)
 		}
 
-		n, err := pipeConn.Write(data)
+		n, err := tcpConn.Write(data)
 		if err != nil {
-			p.metrics.RecordError(ctx, "pipe_write_error")
-			log.Warnf("Pipe write error for %s: %v", clientAddr, err)
+			p.metrics.RecordError(ctx, "tcp_write_error")
+			log.Errorf("TCP write error: %v", err)
 			return
 		}
 
-		p.metrics.RecordBytesTransferred(ctx, "ws_to_grpc", int64(n))
+		p.metrics.RecordBytesTransferred(ctx, "ws_to_tcp", int64(n))
 	}
 }
 
-func (p *Proxy) pipeToWS(ctx context.Context, cancel context.CancelFunc, wg *sync.WaitGroup, wsConn *websocket.Conn, pipeConn net.Conn, clientAddr string) {
+func (p *Proxy) tcpToWS(ctx context.Context, cancel context.CancelFunc, wg *sync.WaitGroup, wsConn *websocket.Conn, tcpConn net.Conn) {
 	defer wg.Done()
 	defer cancel()
 
 	buf := make([]byte, bufferSize)
 	for {
-		n, err := pipeConn.Read(buf)
+		if err := tcpConn.SetReadDeadline(time.Now().Add(5 * time.Second)); err != nil {
+			log.Debugf("Failed to set TCP read deadline: %v", err)
+		}
+		n, err := tcpConn.Read(buf)
+
 		if err != nil {
 			if ctx.Err() != nil {
-				log.Tracef("pipeToWS goroutine terminating due to context cancellation")
+				log.Tracef("tcpToWS goroutine terminating due to context cancellation")
 				return
 			}
 
+			var netErr net.Error
+			if errors.As(err, &netErr) && netErr.Timeout() {
+				continue
+			}
+
 			if err != io.EOF {
-				log.Debugf("Pipe read error for %s: %v", clientAddr, err)
+				log.Errorf("TCP read error: %v", err)
 			}
 			return
 		}
 
 		if ctx.Err() != nil {
-			log.Tracef("pipeToWS goroutine terminating due to context cancellation before WebSocket write")
+			log.Tracef("tcpToWS goroutine terminating due to context cancellation before WebSocket write")
 			return
 		}
 
-		if n > 0 {
-			if err := wsConn.Write(ctx, websocket.MessageBinary, buf[:n]); err != nil {
-				p.metrics.RecordError(ctx, "websocket_write_error")
-				log.Warnf("WebSocket write error for %s: %v", clientAddr, err)
-				return
-			}
-
-			p.metrics.RecordBytesTransferred(ctx, "grpc_to_ws", int64(n))
+		if err := wsConn.Write(ctx, websocket.MessageBinary, buf[:n]); err != nil {
+			p.metrics.RecordError(ctx, "websocket_write_error")
+			log.Errorf("WebSocket write error: %v", err)
+			return
 		}
+
+		p.metrics.RecordBytesTransferred(ctx, "tcp_to_ws", int64(n))
 	}
 }