Evict orphaned packet captures and annotate VNC streams

client/internal/engine_vnc.go

@@ -15,16 +15,13 @@ import (
 	"github.com/netbirdio/netbird/client/internal/metrics"
 	nftypes "github.com/netbirdio/netbird/client/internal/netflow/types"
 	"github.com/netbirdio/netbird/client/internal/peer"
-	sshauth "github.com/netbirdio/netbird/shared/sessionauth"
+	"github.com/netbirdio/netbird/client/vnc"
 	vncserver "github.com/netbirdio/netbird/client/vnc/server"
+	sshauth "github.com/netbirdio/netbird/shared/sessionauth"
 	mgmProto "github.com/netbirdio/netbird/shared/management/proto"
 	sshuserhash "github.com/netbirdio/netbird/shared/sshauth"
 )
 
-const (
-	vncExternalPort uint16 = 5900
-	vncInternalPort uint16 = 25900
-)
 
 type vncServer interface {
 	Start(ctx context.Context, addr netip.AddrPort, network netip.Prefix) error
@@ -42,10 +39,10 @@ func (e *Engine) setupVNCPortRedirection() error {
 		return errors.New("invalid local NetBird address")
 	}
 
-	if err := e.firewall.AddInboundDNAT(localAddr, firewallManager.ProtocolTCP, vncExternalPort, vncInternalPort); err != nil {
+	if err := e.firewall.AddInboundDNAT(localAddr, firewallManager.ProtocolTCP, vnc.ExternalPort, vnc.InternalPort); err != nil {
 		return fmt.Errorf("add VNC port redirection: %w", err)
 	}
-	log.Infof("VNC port redirection: %s:%d -> %s:%d", localAddr, vncExternalPort, localAddr, vncInternalPort)
+	log.Infof("VNC port redirection: %s:%d -> %s:%d", localAddr, vnc.ExternalPort, localAddr, vnc.InternalPort)
 
 	return nil
 }
@@ -60,7 +57,7 @@ func (e *Engine) cleanupVNCPortRedirection() error {
 		return errors.New("invalid local NetBird address")
 	}
 
-	if err := e.firewall.RemoveInboundDNAT(localAddr, firewallManager.ProtocolTCP, vncExternalPort, vncInternalPort); err != nil {
+	if err := e.firewall.RemoveInboundDNAT(localAddr, firewallManager.ProtocolTCP, vnc.ExternalPort, vnc.InternalPort); err != nil {
 		return fmt.Errorf("remove VNC port redirection: %w", err)
 	}
 
@@ -132,7 +129,7 @@ func (e *Engine) startVNCServer() error {
 		Approver:        &vncApprover{broker: e.approvalBroker, statusRecorder: e.statusRecorder},
 	})
 
-	listenAddr := netip.AddrPortFrom(netbirdIP, vncInternalPort)
+	listenAddr := netip.AddrPortFrom(netbirdIP, vnc.InternalPort)
 	network := e.wgInterface.Address().Network
 	if err := srv.Start(e.ctx, listenAddr, network); err != nil {
 		return fmt.Errorf("start VNC server: %w", err)
@@ -144,8 +141,8 @@ func (e *Engine) startVNCServer() error {
 		if registrar, ok := e.firewall.(interface {
 			RegisterNetstackService(protocol nftypes.Protocol, port uint16)
 		}); ok {
-			registrar.RegisterNetstackService(nftypes.TCP, vncInternalPort)
-			log.Debugf("registered VNC service with netstack for TCP:%d", vncInternalPort)
+			registrar.RegisterNetstackService(nftypes.TCP, vnc.InternalPort)
+			log.Debugf("registered VNC service with netstack for TCP:%d", vnc.InternalPort)
 		}
 	}
 
@@ -231,7 +228,7 @@ func (e *Engine) stopVNCServer() error {
 		if registrar, ok := e.firewall.(interface {
 			UnregisterNetstackService(protocol nftypes.Protocol, port uint16)
 		}); ok {
-			registrar.UnregisterNetstackService(nftypes.TCP, vncInternalPort)
+			registrar.UnregisterNetstackService(nftypes.TCP, vnc.InternalPort)
 		}
 	}
 

client/server/capture.go

@@ -190,10 +190,7 @@ func (s *Server) StartBundleCapture(_ context.Context, req *proto.StartBundleCap
 
 	s.stopBundleCaptureLocked()
 	s.cleanupBundleCapture()
-
-	if s.activeCapture != nil {
-		return nil, status.Error(codes.FailedPrecondition, "another capture is already running")
-	}
+	s.evictActiveCaptureLocked()
 
 	engine, err := s.getCaptureEngineLocked()
 	if err != nil {
@@ -304,15 +301,15 @@ func (s *Server) cleanupBundleCapture() {
 	s.bundleCapture = nil
 }
 
-// claimCapture reserves the engine's capture slot for sess. Returns
-// FailedPrecondition if another capture is already active.
+// claimCapture reserves the engine's capture slot for sess. If another
+// capture is already running it is evicted: a previous streaming session
+// whose gRPC client died and never freed the slot stays stuck otherwise,
+// and a bundle capture is just informational state.
 func (s *Server) claimCapture(sess *capture.Session) (*internal.Engine, error) {
 	s.mutex.Lock()
 	defer s.mutex.Unlock()
 
-	if s.activeCapture != nil {
-		return nil, status.Error(codes.FailedPrecondition, "another capture is already running")
-	}
+	s.evictActiveCaptureLocked()
 	engine, err := s.getCaptureEngineLocked()
 	if err != nil {
 		return nil, err
@@ -321,6 +318,28 @@ func (s *Server) claimCapture(sess *capture.Session) (*internal.Engine, error) {
 	return engine, nil
 }
 
+// evictActiveCaptureLocked tears down whatever capture currently owns
+// the engine slot so a fresh claim can succeed. Caller must hold mutex.
+func (s *Server) evictActiveCaptureLocked() {
+	if s.activeCapture == nil {
+		return
+	}
+	if s.bundleCapture != nil && s.bundleCapture.sess == s.activeCapture {
+		log.Infof("evicting running bundle capture to start a new capture")
+		s.stopBundleCaptureLocked()
+		return
+	}
+	log.Infof("evicting previous streaming capture to start a new one")
+	prev := s.activeCapture
+	if engine, err := s.getCaptureEngineLocked(); err == nil {
+		if err := engine.SetCapture(nil); err != nil {
+			log.Debugf("clear previous capture: %v", err)
+		}
+	}
+	s.activeCapture = nil
+	prev.Stop()
+}
+
 // releaseCapture clears the active-capture owner if it still matches sess.
 func (s *Server) releaseCapture(sess *capture.Session) {
 	s.mutex.Lock()

client/vnc/ports.go

@@ -0,0 +1,31 @@
+// Package vnc holds shared constants for the NetBird embedded VNC stack
+// so non-server consumers (CLI capture, debug tooling) can refer to the
+// well-known ports without depending on internal engine packages.
+package vnc
+
+// External and internal listen ports for the embedded VNC server.
+// ExternalPort is what dashboard / browser clients see; the daemon
+// DNATs it to InternalPort, where the in-process VNC server actually
+// listens. Both flow over the WireGuard interface. AgentLegacyPort is
+// the TCP port the per-session agent used before it switched to Unix
+// sockets; kept here so packet captures from older builds still get
+// tagged, and so any future on-wire agent variant has a reserved port.
+const (
+	ExternalPort     uint16 = 5900
+	InternalPort     uint16 = 25900
+	AgentLegacyPort  uint16 = 15900
+)
+
+// WellKnownPorts is the unordered set of ports a packet capture should
+// treat as carrying NetBird VNC traffic.
+var WellKnownPorts = [...]uint16{ExternalPort, InternalPort, AgentLegacyPort}
+
+// IsWellKnownPort reports whether port matches any of WellKnownPorts.
+func IsWellKnownPort(port uint16) bool {
+	for _, p := range WellKnownPorts {
+		if port == p {
+			return true
+		}
+	}
+	return false
+}

client/vnc/server/agent_darwin.go

@@ -222,8 +222,8 @@ func waitForAgent(ctx context.Context, socketPath string, wait time.Duration) er
 }
 
 // vncAgentRunning reports whether any vnc-agent process exists on the
-// system. The daemon owns the only port-15900 listener model, so any
-// match is "the" agent.
+// system. There is at most one agent per machine, so any match is "the"
+// agent.
 func vncAgentRunning() bool {
 	pids, err := vncAgentPIDs()
 	if err != nil {

util/capture/text.go

@@ -90,7 +90,7 @@ func (tw *TextWriter) writeTCP(timeStr string, dir Direction, info *packetInfo,
 	// Protocol annotation
 	var annotation string
 	if plen > 0 {
-		annotation = annotatePayload(tcp.Payload)
+		annotation = annotatePayload(tcp.Payload, info.srcPort, info.dstPort)
 	}
 
 	if !tw.verbose {
@@ -363,8 +363,11 @@ func formatTCPOptions(opts []layers.TCPOption) string {
 
 // --- Protocol annotation ---
 
-// annotatePayload returns a protocol annotation string for known application protocols.
-func annotatePayload(payload []byte) string {
+// annotatePayload returns a protocol annotation string for known
+// application protocols. srcPort/dstPort enable port-tagged
+// annotations (e.g. NetBird VNC traffic) that can't be identified from
+// the payload alone.
+func annotatePayload(payload []byte, srcPort, dstPort uint16) string {
 	if len(payload) < 4 {
 		return ""
 	}
@@ -397,9 +400,169 @@ func annotatePayload(payload []byte) string {
 		}
 	}
 
+	// NetBird VNC: tag by port and try a few payload heuristics.
+	if isVNCPort(srcPort, dstPort) {
+		return ": " + annotateVNC(payload, srcPort, dstPort)
+	}
+
 	return ""
 }
 
+// isVNCPort mirrors client/vnc/ports.go. 5900 is the external port the
+// dashboard talks to, 25900 is the internal DNAT target, 15900 is the
+// legacy agent TCP port (agents now use Unix sockets, but historical
+// captures are still useful).
+func isVNCPort(src, dst uint16) bool {
+	return src == 5900 || dst == 5900 ||
+		src == 25900 || dst == 25900 ||
+		src == 15900 || dst == 15900
+}
+
+// annotateVNC inspects a payload assumed to be on a NetBird VNC port and
+// returns a short tag. The annotation is stateless and best-effort: we
+// don't track per-flow phase, so msg-type decoding only fires when the
+// length plausibly matches a known fixed-size RFB message.
+func annotateVNC(payload []byte, srcPort, dstPort uint16) string {
+	s := string(payload)
+
+	// Banner / control-plane recognitions match in either direction.
+	switch {
+	case strings.HasPrefix(s, "RFB 003."):
+		end := strings.IndexByte(s, '\n')
+		if end > 0 && end < 32 {
+			return "VNC " + strings.TrimSpace(s[:end])
+		}
+		return "VNC RFB"
+	case strings.Contains(s, "\x00NB-VIEW-ONLY\x00"):
+		return "VNC view-only announce"
+	case isVNCRejectPrefix(s):
+		if i := strings.IndexByte(s, ':'); i > 0 && i < 32 {
+			return "VNC reject " + s[:i]
+		}
+		return "VNC reject"
+	}
+
+	// Direction by port. Well-known VNC port is always on the server
+	// side; the other end is an ephemeral client port.
+	dstIsServer := isWellKnownVNCPort(dstPort)
+	srcIsServer := isWellKnownVNCPort(srcPort)
+	switch {
+	case dstIsServer && !srcIsServer:
+		return "VNC " + annotateVNCClientToServer(payload)
+	case srcIsServer && !dstIsServer:
+		return "VNC " + annotateVNCServerToClient(payload)
+	}
+	return "VNC"
+}
+
+func isWellKnownVNCPort(p uint16) bool {
+	return p == 5900 || p == 25900 || p == 15900
+}
+
+// annotateVNCClientToServer guesses what a client-bound payload contains.
+// First-packet path: looks for the NetBird connection header. RFB
+// message-type recognitions fire only when length matches the fixed
+// size for that type, to avoid mis-tagging Noise handshake bytes.
+func annotateVNCClientToServer(p []byte) string {
+	if len(p) >= 10 && (p[0] == 0 || p[0] == 1) {
+		userLen := int(p[1])
+		// width and height are uint16 fields the dashboard often leaves
+		// zero (default). A header without an OS user has total length
+		// 10; with one, 10+userLen.
+		if 10+userLen <= len(p) {
+			mode := "attach"
+			if p[0] == 1 {
+				mode = "session"
+			}
+			tag := fmt.Sprintf("connect mode=%s", mode)
+			if userLen > 0 {
+				tag += fmt.Sprintf(" user(%d)", userLen)
+			}
+			return tag
+		}
+	}
+	switch {
+	case len(p) == 20 && p[0] == 0:
+		return "SetPixelFormat"
+	case p[0] == 2:
+		return "SetEncodings"
+	case len(p) == 10 && p[0] == 3:
+		return "FramebufferUpdateRequest"
+	case len(p) == 8 && p[0] == 4:
+		return "KeyEvent"
+	case (len(p) == 6 || len(p) == 7) && p[0] == 5:
+		return "PointerEvent"
+	case p[0] == 6:
+		return "ClientCutText"
+	case p[0] == 0xFC:
+		return "QEMUClientMsg"
+	}
+	return ""
+}
+
+// annotateVNCServerToClient guesses what a server-bound payload contains.
+// The security-failure path (numTypes=0 + 4-byte reasonLen + reason) is
+// recognised before FramebufferUpdate because both start with 0x00 and
+// the failure carries a self-describing length we can verify.
+func annotateVNCServerToClient(p []byte) string {
+	if reason, ok := matchRFBSecurityFailure(p); ok {
+		if code, _, found := strings.Cut(reason, ": "); found && isVNCRejectPrefix(reason) {
+			return "reject " + code
+		}
+		return "reject"
+	}
+	switch {
+	case len(p) >= 4 && p[0] == 0:
+		return "FramebufferUpdate"
+	case p[0] == 1:
+		return "SetColorMapEntries"
+	case len(p) == 1 && p[0] == 2:
+		return "Bell"
+	case p[0] == 3:
+		return "ServerCutText"
+	}
+	return ""
+}
+
+// matchRFBSecurityFailure recognises the RFB 3.8 security-result body the
+// server sends when authentication or session setup fails. Format:
+//   byte 0  : 0x00 (security types count = 0 = failure)
+//   bytes 1-4: uint32 reason length
+//   bytes 5+: reason text
+// Returns the reason text and ok=true when the length self-checks.
+func matchRFBSecurityFailure(p []byte) (string, bool) {
+	if len(p) < 5 || p[0] != 0 {
+		return "", false
+	}
+	reasonLen := int(p[1])<<24 | int(p[2])<<16 | int(p[3])<<8 | int(p[4])
+	if reasonLen <= 0 || reasonLen > 4096 || 5+reasonLen != len(p) {
+		return "", false
+	}
+	return string(p[5 : 5+reasonLen]), true
+}
+
+// vncRejectCodes mirrors the RejectCode* constants in
+// client/vnc/server/server.go. New codes should be added here too.
+var vncRejectCodes = [...]string{
+	"AUTH_FORBIDDEN",
+	"SESSION_ERROR",
+	"CAPTURER_ERROR",
+	"UNSUPPORTED",
+	"BAD_REQUEST",
+	"NO_CONSOLE_USER",
+	"APPROVAL_DENIED",
+	"NO_APPROVER",
+}
+
+func isVNCRejectPrefix(s string) bool {
+	for _, c := range vncRejectCodes {
+		if strings.HasPrefix(s, c+":") {
+			return true
+		}
+	}
+	return false
+}
+
 // annotateTLS returns a description for TLS handshake and alert records.
 func annotateTLS(data []byte) string {
 	if len(data) < 6 {