Add dialWebSocket method to WASM client

client/firewall/create_linux.go

@@ -56,13 +56,6 @@ func NewFirewall(iface IFaceMapper, stateManager *statemanager.Manager, flowLogg
 		return createUserspaceFirewall(iface, nil, disableServerRoutes, flowLogger, mtu)
 	}
 
-	// Native firewall handles packet filtering, but the userspace WireGuard bind
-	// needs a device filter for DNS interception hooks. Install a minimal
-	// hooks-only filter that passes all traffic through to the kernel firewall.
-	if err := iface.SetFilter(&uspfilter.HooksFilter{}); err != nil {
-		log.Warnf("failed to set hooks filter, DNS via memory hooks will not work: %v", err)
-	}
-
 	return fm, nil
 }
 

client/firewall/uspfilter/common/hooks.go

@@ -1,37 +0,0 @@
-package common
-
-import (
-	"net/netip"
-	"sync/atomic"
-)
-
-// PacketHook stores a registered hook for a specific IP:port.
-type PacketHook struct {
-	IP   netip.Addr
-	Port uint16
-	Fn   func([]byte) bool
-}
-
-// HookMatches checks if a packet's destination matches the hook and invokes it.
-func HookMatches(h *PacketHook, dstIP netip.Addr, dport uint16, packetData []byte) bool {
-	if h == nil {
-		return false
-	}
-	if h.IP == dstIP && h.Port == dport {
-		return h.Fn(packetData)
-	}
-	return false
-}
-
-// SetHook atomically stores a hook, handling nil removal.
-func SetHook(ptr *atomic.Pointer[PacketHook], ip netip.Addr, dPort uint16, hook func([]byte) bool) {
-	if hook == nil {
-		ptr.Store(nil)
-		return
-	}
-	ptr.Store(&PacketHook{
-		IP:   ip,
-		Port: dPort,
-		Fn:   hook,
-	})
-}

client/firewall/uspfilter/filter.go

@@ -142,8 +142,15 @@ type Manager struct {
 	mssClampEnabled bool
 
 	// Only one hook per protocol is supported. Outbound direction only.
-	udpHookOut atomic.Pointer[common.PacketHook]
-	tcpHookOut atomic.Pointer[common.PacketHook]
+	udpHookOut atomic.Pointer[packetHook]
+	tcpHookOut atomic.Pointer[packetHook]
+}
+
+// packetHook stores a registered hook for a specific IP:port.
+type packetHook struct {
+	ip   netip.Addr
+	port uint16
+	fn   func([]byte) bool
 }
 
 // decoder for packages
@@ -905,11 +912,21 @@ func (m *Manager) trackInbound(d *decoder, srcIP, dstIP netip.Addr, ruleID []byt
 }
 
 func (m *Manager) udpHooksDrop(dport uint16, dstIP netip.Addr, packetData []byte) bool {
-	return common.HookMatches(m.udpHookOut.Load(), dstIP, dport, packetData)
+	return hookMatches(m.udpHookOut.Load(), dstIP, dport, packetData)
 }
 
 func (m *Manager) tcpHooksDrop(dport uint16, dstIP netip.Addr, packetData []byte) bool {
-	return common.HookMatches(m.tcpHookOut.Load(), dstIP, dport, packetData)
+	return hookMatches(m.tcpHookOut.Load(), dstIP, dport, packetData)
+}
+
+func hookMatches(h *packetHook, dstIP netip.Addr, dport uint16, packetData []byte) bool {
+	if h == nil {
+		return false
+	}
+	if h.ip == dstIP && h.port == dport {
+		return h.fn(packetData)
+	}
+	return false
 }
 
 // filterInbound implements filtering logic for incoming packets.
@@ -1320,12 +1337,28 @@ func (m *Manager) ruleMatches(rule *RouteRule, srcAddr, dstAddr netip.Addr, prot
 
 // SetUDPPacketHook sets the outbound UDP packet hook. Pass nil hook to remove.
 func (m *Manager) SetUDPPacketHook(ip netip.Addr, dPort uint16, hook func(packet []byte) bool) {
-	common.SetHook(&m.udpHookOut, ip, dPort, hook)
+	if hook == nil {
+		m.udpHookOut.Store(nil)
+		return
+	}
+	m.udpHookOut.Store(&packetHook{
+		ip:   ip,
+		port: dPort,
+		fn:   hook,
+	})
 }
 
 // SetTCPPacketHook sets the outbound TCP packet hook. Pass nil hook to remove.
 func (m *Manager) SetTCPPacketHook(ip netip.Addr, dPort uint16, hook func(packet []byte) bool) {
-	common.SetHook(&m.tcpHookOut, ip, dPort, hook)
+	if hook == nil {
+		m.tcpHookOut.Store(nil)
+		return
+	}
+	m.tcpHookOut.Store(&packetHook{
+		ip:   ip,
+		port: dPort,
+		fn:   hook,
+	})
 }
 
 // SetLogLevel sets the log level for the firewall manager

client/firewall/uspfilter/filter_test.go

@@ -202,9 +202,9 @@ func TestSetUDPPacketHook(t *testing.T) {
 
 	h := manager.udpHookOut.Load()
 	require.NotNil(t, h)
-	assert.Equal(t, netip.MustParseAddr("10.168.0.1"), h.IP)
-	assert.Equal(t, uint16(8000), h.Port)
-	assert.True(t, h.Fn(nil))
+	assert.Equal(t, netip.MustParseAddr("10.168.0.1"), h.ip)
+	assert.Equal(t, uint16(8000), h.port)
+	assert.True(t, h.fn(nil))
 	assert.True(t, called)
 
 	manager.SetUDPPacketHook(netip.MustParseAddr("10.168.0.1"), 8000, nil)
@@ -226,9 +226,9 @@ func TestSetTCPPacketHook(t *testing.T) {
 
 	h := manager.tcpHookOut.Load()
 	require.NotNil(t, h)
-	assert.Equal(t, netip.MustParseAddr("10.168.0.1"), h.IP)
-	assert.Equal(t, uint16(53), h.Port)
-	assert.True(t, h.Fn(nil))
+	assert.Equal(t, netip.MustParseAddr("10.168.0.1"), h.ip)
+	assert.Equal(t, uint16(53), h.port)
+	assert.True(t, h.fn(nil))
 	assert.True(t, called)
 
 	manager.SetTCPPacketHook(netip.MustParseAddr("10.168.0.1"), 53, nil)

client/firewall/uspfilter/hooks_filter.go

@@ -1,90 +0,0 @@
-package uspfilter
-
-import (
-	"encoding/binary"
-	"net/netip"
-	"sync/atomic"
-
-	"github.com/netbirdio/netbird/client/firewall/uspfilter/common"
-	"github.com/netbirdio/netbird/client/iface/device"
-)
-
-const (
-	ipv4HeaderMinLen = 20
-	ipv4ProtoOffset  = 9
-	ipv4FlagsOffset  = 6
-	ipv4DstOffset    = 16
-	ipProtoUDP       = 17
-	ipProtoTCP       = 6
-	ipv4FragOffMask  = 0x1fff
-	// dstPortOffset is the offset of the destination port within a UDP or TCP header.
-	dstPortOffset = 2
-)
-
-// HooksFilter is a minimal packet filter that only handles outbound DNS hooks.
-// It is installed on the WireGuard interface when the userspace bind is active
-// but a full firewall filter (Manager) is not needed because a native kernel
-// firewall (nftables/iptables) handles packet filtering.
-type HooksFilter struct {
-	udpHook atomic.Pointer[common.PacketHook]
-	tcpHook atomic.Pointer[common.PacketHook]
-}
-
-var _ device.PacketFilter = (*HooksFilter)(nil)
-
-// FilterOutbound checks outbound packets for DNS hook matches.
-// Only IPv4 packets matching the registered hook IP:port are intercepted.
-// IPv6 and non-IP packets pass through unconditionally.
-func (f *HooksFilter) FilterOutbound(packetData []byte, _ int) bool {
-	if len(packetData) < ipv4HeaderMinLen {
-		return false
-	}
-
-	// Only process IPv4 packets, let everything else pass through.
-	if packetData[0]>>4 != 4 {
-		return false
-	}
-
-	ihl := int(packetData[0]&0x0f) * 4
-	if ihl < ipv4HeaderMinLen || len(packetData) < ihl+4 {
-		return false
-	}
-
-	// Skip non-first fragments: they don't carry L4 headers.
-	flagsAndOffset := binary.BigEndian.Uint16(packetData[ipv4FlagsOffset : ipv4FlagsOffset+2])
-	if flagsAndOffset&ipv4FragOffMask != 0 {
-		return false
-	}
-
-	dstIP, ok := netip.AddrFromSlice(packetData[ipv4DstOffset : ipv4DstOffset+4])
-	if !ok {
-		return false
-	}
-
-	proto := packetData[ipv4ProtoOffset]
-	dstPort := binary.BigEndian.Uint16(packetData[ihl+dstPortOffset : ihl+dstPortOffset+2])
-
-	switch proto {
-	case ipProtoUDP:
-		return common.HookMatches(f.udpHook.Load(), dstIP, dstPort, packetData)
-	case ipProtoTCP:
-		return common.HookMatches(f.tcpHook.Load(), dstIP, dstPort, packetData)
-	default:
-		return false
-	}
-}
-
-// FilterInbound allows all inbound packets (native firewall handles filtering).
-func (f *HooksFilter) FilterInbound([]byte, int) bool {
-	return false
-}
-
-// SetUDPPacketHook registers the UDP packet hook.
-func (f *HooksFilter) SetUDPPacketHook(ip netip.Addr, dPort uint16, hook func([]byte) bool) {
-	common.SetHook(&f.udpHook, ip, dPort, hook)
-}
-
-// SetTCPPacketHook registers the TCP packet hook.
-func (f *HooksFilter) SetTCPPacketHook(ip netip.Addr, dPort uint16, hook func([]byte) bool) {
-	common.SetHook(&f.tcpHook, ip, dPort, hook)
-}

client/internal/routemanager/manager.go

@@ -168,7 +168,6 @@ func (m *DefaultManager) setupAndroidRoutes(config ManagerConfig) {
 			NetworkType: route.IPv4Network,
 		}
 		cr = append(cr, fakeIPRoute)
-		m.notifier.SetFakeIPRoute(fakeIPRoute)
 	}
 
 	m.notifier.SetInitialClientRoutes(cr, routesForComparison)

client/internal/routemanager/notifier/notifier_android.go

@@ -16,7 +16,6 @@ import (
 type Notifier struct {
 	initialRoutes []*route.Route
 	currentRoutes []*route.Route
-	fakeIPRoute   *route.Route
 
 	listener    listener.NetworkChangeListener
 	listenerMux sync.Mutex
@@ -32,17 +31,13 @@ func (n *Notifier) SetListener(listener listener.NetworkChangeListener) {
 	n.listener = listener
 }
 
-// SetInitialClientRoutes stores the initial route sets for TUN configuration.
+// SetInitialClientRoutes stores the full initial route set (including fake IP blocks)
+// and a separate comparison set (without fake IP blocks) for diff detection.
 func (n *Notifier) SetInitialClientRoutes(initialRoutes []*route.Route, routesForComparison []*route.Route) {
 	n.initialRoutes = filterStatic(initialRoutes)
 	n.currentRoutes = filterStatic(routesForComparison)
 }
 
-// SetFakeIPRoute stores the fake IP route to be included in every TUN rebuild.
-func (n *Notifier) SetFakeIPRoute(r *route.Route) {
-	n.fakeIPRoute = r
-}
-
 func (n *Notifier) OnNewRoutes(idMap route.HAMap) {
 	var newRoutes []*route.Route
 	for _, routes := range idMap {
@@ -74,9 +69,7 @@ func (n *Notifier) notify() {
 	}
 
 	allRoutes := slices.Clone(n.currentRoutes)
-	if n.fakeIPRoute != nil {
-		allRoutes = append(allRoutes, n.fakeIPRoute)
-	}
+	allRoutes = append(allRoutes, n.extraInitialRoutes()...)
 
 	routeStrings := n.routesToStrings(allRoutes)
 	sort.Strings(routeStrings)
@@ -85,6 +78,23 @@ func (n *Notifier) notify() {
 	}(n.listener)
 }
 
+// extraInitialRoutes returns initialRoutes whose network prefix is absent
+// from currentRoutes (e.g. the fake IP block added at setup time).
+func (n *Notifier) extraInitialRoutes() []*route.Route {
+	currentNets := make(map[netip.Prefix]struct{}, len(n.currentRoutes))
+	for _, r := range n.currentRoutes {
+		currentNets[r.Network] = struct{}{}
+	}
+
+	var extra []*route.Route
+	for _, r := range n.initialRoutes {
+		if _, ok := currentNets[r.Network]; !ok {
+			extra = append(extra, r)
+		}
+	}
+	return extra
+}
+
 func filterStatic(routes []*route.Route) []*route.Route {
 	out := make([]*route.Route, 0, len(routes))
 	for _, r := range routes {

client/internal/routemanager/notifier/notifier_ios.go

@@ -34,10 +34,6 @@ func (n *Notifier) SetInitialClientRoutes([]*route.Route, []*route.Route) {
 	// iOS doesn't care about initial routes
 }
 
-func (n *Notifier) SetFakeIPRoute(*route.Route) {
-	// Not used on iOS
-}
-
 func (n *Notifier) OnNewRoutes(route.HAMap) {
 	// Not used on iOS
 }

client/internal/routemanager/notifier/notifier_other.go

@@ -23,10 +23,6 @@ func (n *Notifier) SetInitialClientRoutes([]*route.Route, []*route.Route) {
 	// Not used on non-mobile platforms
 }
 
-func (n *Notifier) SetFakeIPRoute(*route.Route) {
-	// Not used on non-mobile platforms
-}
-
 func (n *Notifier) OnNewRoutes(idMap route.HAMap) {
 	// Not used on non-mobile platforms
 }

client/wasm/cmd/main.go

@@ -17,6 +17,7 @@ import (
 	"github.com/netbirdio/netbird/client/wasm/internal/http"
 	"github.com/netbirdio/netbird/client/wasm/internal/rdp"
 	"github.com/netbirdio/netbird/client/wasm/internal/ssh"
+	nbwebsocket "github.com/netbirdio/netbird/client/wasm/internal/websocket"
 	"github.com/netbirdio/netbird/util"
 )
 
@@ -516,6 +517,7 @@ func createClientObject(client *netbird.Client) js.Value {
 	obj["createSSHConnection"] = createSSHMethod(client)
 	obj["proxyRequest"] = createProxyRequestMethod(client)
 	obj["createRDPProxy"] = createRDPProxyMethod(client)
+	obj["dialWebSocket"] = createDialWebSocketMethod(client)
 	obj["status"] = createStatusMethod(client)
 	obj["statusSummary"] = createStatusSummaryMethod(client)
 	obj["statusDetail"] = createStatusDetailMethod(client)
@@ -525,6 +527,31 @@ func createClientObject(client *netbird.Client) js.Value {
 	return js.ValueOf(obj)
 }
 
+const dialWebSocketTimeout = 30 * time.Second
+
+func createDialWebSocketMethod(client *netbird.Client) js.Func {
+	return js.FuncOf(func(_ js.Value, args []js.Value) any {
+		if len(args) < 1 || args[0].Type() != js.TypeString {
+			return js.ValueOf("error: dialWebSocket requires a URL string argument")
+		}
+
+		url := args[0].String()
+
+		return createPromise(func(resolve, reject js.Value) {
+			ctx, cancel := context.WithTimeout(context.Background(), dialWebSocketTimeout)
+			defer cancel()
+
+			conn, err := nbwebsocket.Dial(ctx, client, url)
+			if err != nil {
+				reject.Invoke(js.ValueOf(fmt.Sprintf("dial websocket: %v", err)))
+				return
+			}
+
+			resolve.Invoke(nbwebsocket.NewJSInterface(conn))
+		})
+	})
+}
+
 // netBirdClientConstructor acts as a JavaScript constructor function
 func netBirdClientConstructor(_ js.Value, args []js.Value) any {
 	return js.Global().Get("Promise").New(js.FuncOf(func(_ js.Value, promiseArgs []js.Value) any {

client/wasm/internal/websocket/websocket.go

@@ -0,0 +1,261 @@
+//go:build js
+
+package websocket
+
+import (
+	"context"
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"sync"
+	"syscall/js"
+
+	netbird "github.com/netbirdio/netbird/client/embed"
+	"github.com/gobwas/ws"
+	"github.com/gobwas/ws/wsutil"
+	log "github.com/sirupsen/logrus"
+)
+
+type closeError struct {
+	code   uint16
+	reason string
+}
+
+func (e *closeError) Error() string {
+	return fmt.Sprintf("websocket closed: %d %s", e.code, e.reason)
+}
+
+// Conn wraps a WebSocket connection over a NetBird TCP connection.
+type Conn struct {
+	conn      net.Conn
+	mu        sync.Mutex
+	closed    chan struct{}
+	closeOnce sync.Once
+	closeErr  error
+}
+
+// Dial establishes a WebSocket connection to the given URL through the NetBird network.
+func Dial(ctx context.Context, client *netbird.Client, rawURL string) (*Conn, error) {
+	d := ws.Dialer{
+		NetDial: client.Dial,
+	}
+
+	conn, br, _, err := d.Dial(ctx, rawURL)
+	if err != nil {
+		return nil, fmt.Errorf("websocket dial: %w", err)
+	}
+
+	if br != nil {
+		ws.PutReader(br)
+	}
+
+	return &Conn{
+		conn:   conn,
+		closed: make(chan struct{}),
+	}, nil
+}
+
+// ReadMessage reads the next WebSocket message, handling control frames automatically.
+func (c *Conn) ReadMessage() (ws.OpCode, []byte, error) {
+	for {
+		msgs, err := wsutil.ReadServerMessage(c.conn, nil)
+		if err != nil {
+			return 0, nil, err
+		}
+
+		for _, msg := range msgs {
+			if msg.OpCode.IsControl() {
+				if err := c.handleControl(msg); err != nil {
+					return 0, nil, err
+				}
+				continue
+			}
+			return msg.OpCode, msg.Payload, nil
+		}
+	}
+}
+
+func (c *Conn) handleControl(msg wsutil.Message) error {
+	switch msg.OpCode {
+	case ws.OpPing:
+		c.mu.Lock()
+		defer c.mu.Unlock()
+		return wsutil.WriteClientMessage(c.conn, ws.OpPong, msg.Payload)
+	case ws.OpClose:
+		code, reason := parseClosePayload(msg.Payload)
+		return &closeError{code: code, reason: reason}
+	default:
+		return nil
+	}
+}
+
+// WriteText sends a text WebSocket message.
+func (c *Conn) WriteText(data []byte) error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return wsutil.WriteClientMessage(c.conn, ws.OpText, data)
+}
+
+// WriteBinary sends a binary WebSocket message.
+func (c *Conn) WriteBinary(data []byte) error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return wsutil.WriteClientMessage(c.conn, ws.OpBinary, data)
+}
+
+// Close sends a close frame and closes the underlying connection.
+func (c *Conn) Close() error {
+	var first bool
+	c.closeOnce.Do(func() {
+		first = true
+		close(c.closed)
+
+		c.mu.Lock()
+		_ = wsutil.WriteClientMessage(c.conn, ws.OpClose,
+			ws.NewCloseFrameBody(ws.StatusNormalClosure, ""),
+		)
+		c.mu.Unlock()
+
+		c.closeErr = c.conn.Close()
+	})
+
+	if !first {
+		return net.ErrClosed
+	}
+	return c.closeErr
+}
+
+// NewJSInterface creates a JavaScript object wrapping the WebSocket connection.
+// It exposes: send(string|Uint8Array), close(), and callback properties
+// onmessage, onclose, onerror.
+//
+// Callback properties may be set from the JS thread while the read loop
+// goroutine reads them. In WASM this is safe because Go and JS share a
+// single thread, but the design would need synchronization on
+// multi-threaded runtimes.
+func NewJSInterface(conn *Conn) js.Value {
+	obj := js.Global().Get("Object").Call("create", js.Null())
+
+	obj.Set("send", js.FuncOf(func(_ js.Value, args []js.Value) any {
+		if len(args) < 1 {
+			return js.ValueOf("send requires a data argument")
+		}
+
+		data := args[0]
+		switch data.Type() {
+		case js.TypeString:
+			if err := conn.WriteText([]byte(data.String())); err != nil {
+				log.Errorf("failed to send websocket text: %v", err)
+				return js.ValueOf(false)
+			}
+		default:
+			buf, err := jsToBytes(data)
+			if err != nil {
+				return js.ValueOf(err.Error())
+			}
+			if err := conn.WriteBinary(buf); err != nil {
+				log.Errorf("failed to send websocket binary: %v", err)
+				return js.ValueOf(false)
+			}
+		}
+		return js.ValueOf(true)
+	}))
+
+	obj.Set("close", js.FuncOf(func(_ js.Value, _ []js.Value) any {
+		if err := conn.Close(); err != nil {
+			log.Debugf("failed to close websocket: %v", err)
+		}
+		return js.Undefined()
+	}))
+
+	go readLoop(conn, obj)
+
+	return obj
+}
+
+func jsToBytes(data js.Value) ([]byte, error) {
+	var uint8Array js.Value
+	switch {
+	case data.InstanceOf(js.Global().Get("Uint8Array")):
+		uint8Array = data
+	case data.InstanceOf(js.Global().Get("ArrayBuffer")):
+		uint8Array = js.Global().Get("Uint8Array").New(data)
+	default:
+		return nil, fmt.Errorf("send: unsupported data type, use string or Uint8Array")
+	}
+
+	buf := make([]byte, uint8Array.Get("length").Int())
+	js.CopyBytesToGo(buf, uint8Array)
+	return buf, nil
+}
+
+func readLoop(conn *Conn, obj js.Value) {
+	var closeCode uint16
+	var closeReason string
+	var gotCloseFrame bool
+
+	defer func() {
+		onclose := obj.Get("onclose")
+		if !onclose.Truthy() {
+			return
+		}
+		if gotCloseFrame {
+			onclose.Invoke(js.ValueOf(int(closeCode)), js.ValueOf(closeReason))
+			return
+		}
+		onclose.Invoke()
+	}()
+
+	for {
+		select {
+		case <-conn.closed:
+			return
+		default:
+		}
+
+		op, payload, err := conn.ReadMessage()
+		if err != nil {
+			var ce *closeError
+			if errors.As(err, &ce) {
+				gotCloseFrame = true
+				closeCode = ce.code
+				closeReason = ce.reason
+				// Respond to server close per RFC 6455.
+				if err := conn.Close(); err != nil {
+					log.Debugf("failed to close websocket after server close frame: %v", err)
+				}
+				return
+			}
+			if err != io.EOF {
+				if onerror := obj.Get("onerror"); onerror.Truthy() {
+					onerror.Invoke(js.ValueOf(err.Error()))
+				}
+			}
+			return
+		}
+
+		onmessage := obj.Get("onmessage")
+		if !onmessage.Truthy() {
+			continue
+		}
+
+		switch op {
+		case ws.OpText:
+			onmessage.Invoke(js.ValueOf(string(payload)))
+		case ws.OpBinary:
+			uint8Array := js.Global().Get("Uint8Array").New(len(payload))
+			js.CopyBytesToJS(uint8Array, payload)
+			onmessage.Invoke(uint8Array)
+		}
+	}
+}
+
+func parseClosePayload(payload []byte) (uint16, string) {
+	if len(payload) < 2 {
+		return 1005, "" // RFC 6455: No Status Rcvd
+	}
+	code := binary.BigEndian.Uint16(payload[:2])
+	return code, string(payload[2:])
+}

flow/client/client.go

@@ -14,6 +14,7 @@ import (
 	log "github.com/sirupsen/logrus"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/keepalive"
@@ -25,22 +26,11 @@ import (
 	"github.com/netbirdio/netbird/util/wsproxy"
 )
 
-var ErrClientClosed = errors.New("client is closed")
-
-// minHealthyDuration is the minimum time a stream must survive before a failure
-// resets the backoff timer. Streams that fail faster are considered unhealthy and
-// should not reset backoff, so that MaxElapsedTime can eventually stop retries.
-const minHealthyDuration = 5 * time.Second
-
 type GRPCClient struct {
 	realClient proto.FlowServiceClient
 	clientConn *grpc.ClientConn
 	stream     proto.FlowService_EventsClient
-	target     string
-	opts       []grpc.DialOption
-	closed     bool       // prevent creating conn in the middle of the Close
-	receiving  bool       // prevent concurrent Receive calls
-	mu         sync.Mutex // protects clientConn, realClient, stream, closed, and receiving
+	streamMu   sync.Mutex
 }
 
 func NewClient(addr, payload, signature string, interval time.Duration) (*GRPCClient, error) {
@@ -75,8 +65,7 @@ func NewClient(addr, payload, signature string, interval time.Duration) (*GRPCCl
 		grpc.WithDefaultServiceConfig(`{"healthCheckConfig": {"serviceName": ""}}`),
 	)
 
-	target := parsedURL.Host
-	conn, err := grpc.NewClient(target, opts...)
+	conn, err := grpc.NewClient(fmt.Sprintf("%s:%s", parsedURL.Hostname(), parsedURL.Port()), opts...)
 	if err != nil {
 		return nil, fmt.Errorf("creating new grpc client: %w", err)
 	}
@@ -84,73 +73,30 @@ func NewClient(addr, payload, signature string, interval time.Duration) (*GRPCCl
 	return &GRPCClient{
 		realClient: proto.NewFlowServiceClient(conn),
 		clientConn: conn,
-		target:     target,
-		opts:       opts,
 	}, nil
 }
 
 func (c *GRPCClient) Close() error {
-	c.mu.Lock()
-	c.closed = true
+	c.streamMu.Lock()
+	defer c.streamMu.Unlock()
+
 	c.stream = nil
-	conn := c.clientConn
-	c.clientConn = nil
-	c.mu.Unlock()
-
-	if conn == nil {
-		return nil
-	}
-
-	if err := conn.Close(); err != nil && !errors.Is(err, context.Canceled) {
+	if err := c.clientConn.Close(); err != nil && !errors.Is(err, context.Canceled) {
 		return fmt.Errorf("close client connection: %w", err)
 	}
 
 	return nil
 }
 
-func (c *GRPCClient) Send(event *proto.FlowEvent) error {
-	c.mu.Lock()
-	stream := c.stream
-	c.mu.Unlock()
-
-	if stream == nil {
-		return errors.New("stream not initialized")
-	}
-
-	if err := stream.Send(event); err != nil {
-		return fmt.Errorf("send flow event: %w", err)
-	}
-
-	return nil
-}
-
 func (c *GRPCClient) Receive(ctx context.Context, interval time.Duration, msgHandler func(msg *proto.FlowEventAck) error) error {
-	c.mu.Lock()
-	if c.receiving {
-		c.mu.Unlock()
-		return errors.New("concurrent Receive calls are not supported")
-	}
-	c.receiving = true
-	c.mu.Unlock()
-	defer func() {
-		c.mu.Lock()
-		c.receiving = false
-		c.mu.Unlock()
-	}()
-
 	backOff := defaultBackoff(ctx, interval)
 	operation := func() error {
-		stream, err := c.establishStream(ctx)
-		if err != nil {
-			log.Errorf("failed to establish flow stream, retrying: %v", err)
-			return c.handleRetryableError(err, time.Time{}, backOff)
-		}
-
-		streamStart := time.Now()
-
-		if err := c.receive(stream, msgHandler); err != nil {
+		if err := c.establishStreamAndReceive(ctx, msgHandler); err != nil {
+			if s, ok := status.FromError(err); ok && s.Code() == codes.Canceled {
+				return fmt.Errorf("receive: %w: %w", err, context.Canceled)
+			}
 			log.Errorf("receive failed: %v", err)
-			return c.handleRetryableError(err, streamStart, backOff)
+			return fmt.Errorf("receive: %w", err)
 		}
 		return nil
 	}
@@ -162,106 +108,37 @@ func (c *GRPCClient) Receive(ctx context.Context, interval time.Duration, msgHan
 	return nil
 }
 
-// handleRetryableError resets the backoff timer if the stream was healthy long
-// enough and recreates the underlying ClientConn so that gRPC's internal
-// subchannel backoff does not accumulate and compete with our own retry timer.
-// A zero streamStart means the stream was never established.
-func (c *GRPCClient) handleRetryableError(err error, streamStart time.Time, backOff backoff.BackOff) error {
-	if isContextDone(err) {
-		return backoff.Permanent(err)
+func (c *GRPCClient) establishStreamAndReceive(ctx context.Context, msgHandler func(msg *proto.FlowEventAck) error) error {
+	if c.clientConn.GetState() == connectivity.Shutdown {
+		return errors.New("connection to flow receiver has been shut down")
 	}
 
-	var permErr *backoff.PermanentError
-	if errors.As(err, &permErr) {
-		return err
-	}
-
-	// Reset the backoff so the next retry starts with a short delay instead of
-	// continuing the already-elapsed timer. Only do this if the stream was healthy
-	// long enough; short-lived connect/drop cycles must not defeat MaxElapsedTime.
-	if !streamStart.IsZero() && time.Since(streamStart) >= minHealthyDuration {
-		backOff.Reset()
-	}
-
-	if recreateErr := c.recreateConnection(); recreateErr != nil {
-		log.Errorf("recreate connection: %v", recreateErr)
-		return recreateErr
-	}
-
-	log.Infof("connection recreated, retrying stream")
-	return fmt.Errorf("retrying after error: %w", err)
-}
-
-func (c *GRPCClient) recreateConnection() error {
-	c.mu.Lock()
-	if c.closed {
-		c.mu.Unlock()
-		return backoff.Permanent(ErrClientClosed)
-	}
-
-	conn, err := grpc.NewClient(c.target, c.opts...)
+	stream, err := c.realClient.Events(ctx, grpc.WaitForReady(true))
 	if err != nil {
-		c.mu.Unlock()
-		return fmt.Errorf("create new connection: %w", err)
+		return fmt.Errorf("create event stream: %w", err)
 	}
 
-	old := c.clientConn
-	c.clientConn = conn
-	c.realClient = proto.NewFlowServiceClient(conn)
-	c.stream = nil
-	c.mu.Unlock()
-
-	_ = old.Close()
-
-	return nil
-}
-
-func (c *GRPCClient) establishStream(ctx context.Context) (proto.FlowService_EventsClient, error) {
-	c.mu.Lock()
-	if c.closed {
-		c.mu.Unlock()
-		return nil, backoff.Permanent(ErrClientClosed)
-	}
-	cl := c.realClient
-	c.mu.Unlock()
-
-	// open stream outside the lock — blocking operation
-	stream, err := cl.Events(ctx)
+	err = stream.Send(&proto.FlowEvent{IsInitiator: true})
 	if err != nil {
-		return nil, fmt.Errorf("create event stream: %w", err)
-	}
-	streamReady := false
-	defer func() {
-		if !streamReady {
-			_ = stream.CloseSend()
-		}
-	}()
-
-	if err = stream.Send(&proto.FlowEvent{IsInitiator: true}); err != nil {
-		return nil, fmt.Errorf("send initiator: %w", err)
+		log.Infof("failed to send initiator message to flow receiver but will attempt to continue. Error: %s", err)
 	}
 
 	if err = checkHeader(stream); err != nil {
-		return nil, fmt.Errorf("check header: %w", err)
+		return fmt.Errorf("check header: %w", err)
 	}
 
-	c.mu.Lock()
-	if c.closed {
-		c.mu.Unlock()
-		return nil, backoff.Permanent(ErrClientClosed)
-	}
+	c.streamMu.Lock()
 	c.stream = stream
-	c.mu.Unlock()
-	streamReady = true
+	c.streamMu.Unlock()
 
-	return stream, nil
+	return c.receive(stream, msgHandler)
 }
 
 func (c *GRPCClient) receive(stream proto.FlowService_EventsClient, msgHandler func(msg *proto.FlowEventAck) error) error {
 	for {
 		msg, err := stream.Recv()
 		if err != nil {
-			return err
+			return fmt.Errorf("receive from stream: %w", err)
 		}
 
 		if msg.IsInitiator {
@@ -292,7 +169,7 @@ func checkHeader(stream proto.FlowService_EventsClient) error {
 func defaultBackoff(ctx context.Context, interval time.Duration) backoff.BackOff {
 	return backoff.WithContext(&backoff.ExponentialBackOff{
 		InitialInterval:     800 * time.Millisecond,
-		RandomizationFactor: 0.5,
+		RandomizationFactor: 1,
 		Multiplier:          1.7,
 		MaxInterval:         interval / 2,
 		MaxElapsedTime:      3 * 30 * 24 * time.Hour, // 3 months
@@ -301,12 +178,18 @@ func defaultBackoff(ctx context.Context, interval time.Duration) backoff.BackOff
 	}, ctx)
 }
 
-func isContextDone(err error) bool {
-	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
-		return true
+func (c *GRPCClient) Send(event *proto.FlowEvent) error {
+	c.streamMu.Lock()
+	stream := c.stream
+	c.streamMu.Unlock()
+
+	if stream == nil {
+		return errors.New("stream not initialized")
 	}
-	if s, ok := status.FromError(err); ok {
-		return s.Code() == codes.Canceled || s.Code() == codes.DeadlineExceeded
+
+	if err := stream.Send(event); err != nil {
+		return fmt.Errorf("send flow event: %w", err)
 	}
-	return false
+
+	return nil
 }

flow/client/client_test.go

@@ -2,11 +2,8 @@ package client_test
 
 import (
 	"context"
-	"encoding/binary"
 	"errors"
 	"net"
-	"sync"
-	"sync/atomic"
 	"testing"
 	"time"
 
@@ -14,8 +11,6 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc"
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/status"
 
 	flow "github.com/netbirdio/netbird/flow/client"
 	"github.com/netbirdio/netbird/flow/proto"
@@ -23,89 +18,21 @@ import (
 
 type testServer struct {
 	proto.UnimplementedFlowServiceServer
-	events         chan *proto.FlowEvent
-	acks           chan *proto.FlowEventAck
-	grpcSrv        *grpc.Server
-	addr           string
-	listener       *connTrackListener
-	closeStream    chan struct{} // signal server to close the stream
-	handlerDone    chan struct{} // signaled each time Events() exits
-	handlerStarted chan struct{} // signaled each time Events() begins
-}
-
-// connTrackListener wraps a net.Listener to track accepted connections
-// so tests can forcefully close them to simulate PROTOCOL_ERROR/RST_STREAM.
-type connTrackListener struct {
-	net.Listener
-	mu    sync.Mutex
-	conns []net.Conn
-}
-
-func (l *connTrackListener) Accept() (net.Conn, error) {
-	c, err := l.Listener.Accept()
-	if err != nil {
-		return nil, err
-	}
-	l.mu.Lock()
-	l.conns = append(l.conns, c)
-	l.mu.Unlock()
-	return c, nil
-}
-
-// sendRSTStream writes a raw HTTP/2 RST_STREAM frame with PROTOCOL_ERROR
-// (error code 0x1) on every tracked connection. This produces the exact error:
-//
-//	rpc error: code = Internal desc = stream terminated by RST_STREAM with error code: PROTOCOL_ERROR
-//
-// HTTP/2 RST_STREAM frame format (9-byte header + 4-byte payload):
-//
-//	Length (3 bytes): 0x000004
-//	Type   (1 byte):  0x03 (RST_STREAM)
-//	Flags  (1 byte):  0x00
-//	Stream ID (4 bytes): target stream (must have bit 31 clear)
-//	Error Code (4 bytes): 0x00000001 (PROTOCOL_ERROR)
-func (l *connTrackListener) connCount() int {
-	l.mu.Lock()
-	defer l.mu.Unlock()
-	return len(l.conns)
-}
-
-func (l *connTrackListener) sendRSTStream(streamID uint32) {
-	l.mu.Lock()
-	defer l.mu.Unlock()
-
-	frame := make([]byte, 13) // 9-byte header + 4-byte payload
-	// Length = 4 (3 bytes, big-endian)
-	frame[0], frame[1], frame[2] = 0, 0, 4
-	// Type = RST_STREAM (0x03)
-	frame[3] = 0x03
-	// Flags = 0
-	frame[4] = 0x00
-	// Stream ID (4 bytes, big-endian, bit 31 reserved = 0)
-	binary.BigEndian.PutUint32(frame[5:9], streamID)
-	// Error Code = PROTOCOL_ERROR (0x1)
-	binary.BigEndian.PutUint32(frame[9:13], 0x1)
-
-	for _, c := range l.conns {
-		_, _ = c.Write(frame)
-	}
+	events  chan *proto.FlowEvent
+	acks    chan *proto.FlowEventAck
+	grpcSrv *grpc.Server
+	addr    string
 }
 
 func newTestServer(t *testing.T) *testServer {
-	rawListener, err := net.Listen("tcp", "127.0.0.1:0")
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
 	require.NoError(t, err)
 
-	listener := &connTrackListener{Listener: rawListener}
-
 	s := &testServer{
-		events:         make(chan *proto.FlowEvent, 100),
-		acks:           make(chan *proto.FlowEventAck, 100),
-		grpcSrv:        grpc.NewServer(),
-		addr:           rawListener.Addr().String(),
-		listener:       listener,
-		closeStream:    make(chan struct{}, 1),
-		handlerDone:    make(chan struct{}, 10),
-		handlerStarted: make(chan struct{}, 10),
+		events:  make(chan *proto.FlowEvent, 100),
+		acks:    make(chan *proto.FlowEventAck, 100),
+		grpcSrv: grpc.NewServer(),
+		addr:    listener.Addr().String(),
 	}
 
 	proto.RegisterFlowServiceServer(s.grpcSrv, s)
@@ -124,23 +51,11 @@ func newTestServer(t *testing.T) *testServer {
 }
 
 func (s *testServer) Events(stream proto.FlowService_EventsServer) error {
-	defer func() {
-		select {
-		case s.handlerDone <- struct{}{}:
-		default:
-		}
-	}()
-
 	err := stream.Send(&proto.FlowEventAck{IsInitiator: true})
 	if err != nil {
 		return err
 	}
 
-	select {
-	case s.handlerStarted <- struct{}{}:
-	default:
-	}
-
 	ctx, cancel := context.WithCancel(stream.Context())
 	defer cancel()
 
@@ -176,8 +91,6 @@ func (s *testServer) Events(stream proto.FlowService_EventsServer) error {
 			if err := stream.Send(ack); err != nil {
 				return err
 			}
-		case <-s.closeStream:
-			return status.Errorf(codes.Internal, "server closing stream")
 		case <-ctx.Done():
 			return ctx.Err()
 		}
@@ -197,13 +110,16 @@ func TestReceive(t *testing.T) {
 		assert.NoError(t, err, "failed to close flow")
 	})
 
-	var ackCount atomic.Int32
+	receivedAcks := make(map[string]bool)
 	receiveDone := make(chan struct{})
 
 	go func() {
 		err := client.Receive(ctx, 1*time.Second, func(msg *proto.FlowEventAck) error {
 			if !msg.IsInitiator && len(msg.EventId) > 0 {
-				if ackCount.Add(1) >= 3 {
+				id := string(msg.EventId)
+				receivedAcks[id] = true
+
+				if len(receivedAcks) >= 3 {
 					close(receiveDone)
 				}
 			}
@@ -214,11 +130,7 @@ func TestReceive(t *testing.T) {
 		}
 	}()
 
-	select {
-	case <-server.handlerStarted:
-	case <-time.After(3 * time.Second):
-		t.Fatal("timeout waiting for stream to be established")
-	}
+	time.Sleep(500 * time.Millisecond)
 
 	for i := 0; i < 3; i++ {
 		eventID := uuid.New().String()
@@ -241,7 +153,7 @@ func TestReceive(t *testing.T) {
 		t.Fatal("timeout waiting for acks to be processed")
 	}
 
-	assert.Equal(t, int32(3), ackCount.Load())
+	assert.Equal(t, 3, len(receivedAcks))
 }
 
 func TestReceive_ContextCancellation(t *testing.T) {
@@ -342,195 +254,3 @@ func TestSend(t *testing.T) {
 		t.Fatal("timeout waiting for ack to be received by flow")
 	}
 }
-
-func TestNewClient_PermanentClose(t *testing.T) {
-	server := newTestServer(t)
-
-	client, err := flow.NewClient("http://"+server.addr, "test-payload", "test-signature", 1*time.Second)
-	require.NoError(t, err)
-
-	err = client.Close()
-	require.NoError(t, err)
-
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-	t.Cleanup(cancel)
-
-	done := make(chan error, 1)
-	go func() {
-		done <- client.Receive(ctx, 1*time.Second, func(msg *proto.FlowEventAck) error {
-			return nil
-		})
-	}()
-
-	select {
-	case err := <-done:
-		require.ErrorIs(t, err, flow.ErrClientClosed)
-	case <-time.After(2 * time.Second):
-		t.Fatal("Receive did not return after Close — stuck in retry loop")
-	}
-}
-
-func TestNewClient_CloseVerify(t *testing.T) {
-	server := newTestServer(t)
-
-	client, err := flow.NewClient("http://"+server.addr, "test-payload", "test-signature", 1*time.Second)
-	require.NoError(t, err)
-
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-	t.Cleanup(cancel)
-
-	done := make(chan error, 1)
-	go func() {
-		done <- client.Receive(ctx, 1*time.Second, func(msg *proto.FlowEventAck) error {
-			return nil
-		})
-	}()
-
-	closeDone := make(chan struct{}, 1)
-	go func() {
-		_ = client.Close()
-		closeDone <- struct{}{}
-	}()
-
-	select {
-	case err := <-done:
-		require.Error(t, err)
-	case <-time.After(2 * time.Second):
-		t.Fatal("Receive did not return after Close — stuck in retry loop")
-	}
-
-	select {
-	case <-closeDone:
-		return
-	case <-time.After(2 * time.Second):
-		t.Fatal("Close did not return — blocked in retry loop")
-	}
-
-}
-
-func TestClose_WhileReceiving(t *testing.T) {
-	server := newTestServer(t)
-	client, err := flow.NewClient("http://"+server.addr, "test-payload", "test-signature", 1*time.Second)
-	require.NoError(t, err)
-
-	ctx := context.Background() // no timeout — intentional
-	receiveDone := make(chan struct{})
-	go func() {
-		_ = client.Receive(ctx, 1*time.Second, func(msg *proto.FlowEventAck) error {
-			return nil
-		})
-		close(receiveDone)
-	}()
-
-	// Wait for the server-side handler to confirm the stream is established.
-	select {
-	case <-server.handlerStarted:
-	case <-time.After(3 * time.Second):
-		t.Fatal("timeout waiting for stream to be established")
-	}
-
-	closeDone := make(chan struct{})
-	go func() {
-		_ = client.Close()
-		close(closeDone)
-	}()
-
-	select {
-	case <-closeDone:
-		// Close returned — good
-	case <-time.After(2 * time.Second):
-		t.Fatal("Close blocked forever — Receive stuck in retry loop")
-	}
-
-	select {
-	case <-receiveDone:
-	case <-time.After(2 * time.Second):
-		t.Fatal("Receive did not exit after Close")
-	}
-}
-
-func TestReceive_ProtocolErrorStreamReconnect(t *testing.T) {
-	server := newTestServer(t)
-
-	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-	t.Cleanup(cancel)
-
-	client, err := flow.NewClient("http://"+server.addr, "test-payload", "test-signature", 1*time.Second)
-	require.NoError(t, err)
-	t.Cleanup(func() {
-		err := client.Close()
-		assert.NoError(t, err, "failed to close flow")
-	})
-
-	// Track acks received before and after server-side stream close
-	var ackCount atomic.Int32
-	receivedFirst := make(chan struct{})
-	receivedAfterReconnect := make(chan struct{})
-
-	go func() {
-		err := client.Receive(ctx, 1*time.Second, func(msg *proto.FlowEventAck) error {
-			if msg.IsInitiator || len(msg.EventId) == 0 {
-				return nil
-			}
-			n := ackCount.Add(1)
-			if n == 1 {
-				close(receivedFirst)
-			}
-			if n == 2 {
-				close(receivedAfterReconnect)
-			}
-			return nil
-		})
-		if err != nil && !errors.Is(err, context.Canceled) {
-			t.Logf("receive error: %v", err)
-		}
-	}()
-
-	// Wait for stream to be established, then send first ack
-	select {
-	case <-server.handlerStarted:
-	case <-time.After(3 * time.Second):
-		t.Fatal("timeout waiting for stream to be established")
-	}
-	server.acks <- &proto.FlowEventAck{EventId: []byte("before-close")}
-
-	select {
-	case <-receivedFirst:
-	case <-time.After(3 * time.Second):
-		t.Fatal("timeout waiting for first ack")
-	}
-
-	// Snapshot connection count before injecting the fault.
-	connsBefore := server.listener.connCount()
-
-	// Send a raw HTTP/2 RST_STREAM frame with PROTOCOL_ERROR on the TCP connection.
-	// gRPC multiplexes streams on stream IDs 1, 3, 5, ... (odd, client-initiated).
-	// Stream ID 1 is the client's first stream (our Events bidi stream).
-	// This produces the exact error the client sees in production:
-	//   "stream terminated by RST_STREAM with error code: PROTOCOL_ERROR"
-	server.listener.sendRSTStream(1)
-
-	// Wait for the old Events() handler to fully exit so it can no longer
-	// drain s.acks and drop our injected ack on a broken stream.
-	select {
-	case <-server.handlerDone:
-	case <-time.After(5 * time.Second):
-		t.Fatal("old Events() handler did not exit after RST_STREAM")
-	}
-
-	require.Eventually(t, func() bool {
-		return server.listener.connCount() > connsBefore
-	}, 5*time.Second, 50*time.Millisecond, "client did not open a new TCP connection after RST_STREAM")
-
-	server.acks <- &proto.FlowEventAck{EventId: []byte("after-close")}
-
-	select {
-	case <-receivedAfterReconnect:
-		// Client successfully reconnected and received ack after server-side stream close
-	case <-time.After(5 * time.Second):
-		t.Fatal("timeout waiting for ack after server-side stream close — client did not reconnect")
-	}
-
-	assert.GreaterOrEqual(t, int(ackCount.Load()), 2, "should have received acks before and after stream close")
-	assert.GreaterOrEqual(t, server.listener.connCount(), 2, "client should have created at least 2 TCP connections (original + reconnect)")
-}

go.mod

@@ -190,6 +190,9 @@ require (
 	github.com/go-sql-driver/mysql v1.9.3 // indirect
 	github.com/go-text/render v0.2.0 // indirect
 	github.com/go-text/typesetting v0.2.1 // indirect
+	github.com/gobwas/httphead v0.1.0 // indirect
+	github.com/gobwas/pool v0.2.1 // indirect
+	github.com/gobwas/ws v1.3.2 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/google/btree v1.1.2 // indirect
 	github.com/google/s2a-go v0.1.9 // indirect

go.sum

@@ -203,6 +203,12 @@ github.com/go-text/typesetting v0.2.1 h1:x0jMOGyO3d1qFAPI0j4GSsh7M0Q3Ypjzr4+CEVg
 github.com/go-text/typesetting v0.2.1/go.mod h1:mTOxEwasOFpAMBjEQDhdWRckoLLeI/+qrQeBCTGEt6M=
 github.com/go-text/typesetting-utils v0.0.0-20241103174707-87a29e9e6066 h1:qCuYC+94v2xrb1PoS4NIDe7DGYtLnU2wWiQe9a1B1c0=
 github.com/go-text/typesetting-utils v0.0.0-20241103174707-87a29e9e6066/go.mod h1:DDxDdQEnB70R8owOx3LVpEFvpMK9eeH1o2r0yZhFI9o=
+github.com/gobwas/httphead v0.1.0 h1:exrUm0f4YX0L7EBwZHuCF4GDp8aJfVeBrlLQrs6NqWU=
+github.com/gobwas/httphead v0.1.0/go.mod h1:O/RXo79gxV8G+RqlR/otEwx4Q36zl9rqC5u12GKvMCM=
+github.com/gobwas/pool v0.2.1 h1:xfeeEhW7pwmX8nuLVlqbzVc7udMDrwetjEv+TZIz1og=
+github.com/gobwas/pool v0.2.1/go.mod h1:q8bcK0KcYlCgd9e7WYLm9LpyS+YeLd8JVDW6WezmKEw=
+github.com/gobwas/ws v1.3.2 h1:zlnbNHxumkRvfPWgfXu8RBwyNR1x8wh9cf5PTOCqs9Q=
+github.com/gobwas/ws v1.3.2/go.mod h1:hRKAFb8wOxFROYNsT1bqfWnhX+b5MFeJM9r2ZSwg/KY=
 github.com/godbus/dbus/v5 v5.1.0 h1:4KLkAxT3aOY8Li4FRJe/KvhoNFFxo0m6fNuFUO8QJUk=
 github.com/godbus/dbus/v5 v5.1.0/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
 github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
@@ -738,6 +744,7 @@ golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.2.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.7.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.10.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=