command listener removed

go.mod

@@ -3,7 +3,7 @@ module github.com/k8shell-io/ssh-proxy
 go 1.24.5
 
 require (
-	github.com/k8shell-io/common v0.37.0
+	github.com/k8shell-io/common v0.40.0
 	github.com/nats-io/nats.go v1.47.0
 	github.com/rs/zerolog v1.34.0
 	golang.org/x/crypto v0.43.0

go.sum

@@ -31,10 +31,12 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
 github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
 github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
-github.com/k8shell-io/common v0.36.0 h1:fkMH1XfYRLzDxqhIq5/luHusWWPGnCGJUXSTEIhEDzI=
-github.com/k8shell-io/common v0.36.0/go.mod h1:E8dsb9ta4v3ne61AJgtRyTTbTkMMmKeCMAcXD+/9+cY=
 github.com/k8shell-io/common v0.37.0 h1:whq66WosIJECKErUKZF1RQep7tdpOfI6GtP4hXREpsQ=
 github.com/k8shell-io/common v0.37.0/go.mod h1:E8dsb9ta4v3ne61AJgtRyTTbTkMMmKeCMAcXD+/9+cY=
+github.com/k8shell-io/common v0.39.0 h1:hfrKZYX2lBonornGrfK35rSY/+5o2sK+SakgUPKDL24=
+github.com/k8shell-io/common v0.39.0/go.mod h1:E8dsb9ta4v3ne61AJgtRyTTbTkMMmKeCMAcXD+/9+cY=
+github.com/k8shell-io/common v0.40.0 h1:MhQPVI5oe+JSdRJhWrtvCTJf0MaJDjM58KR7Nq3+lsM=
+github.com/k8shell-io/common v0.40.0/go.mod h1:E8dsb9ta4v3ne61AJgtRyTTbTkMMmKeCMAcXD+/9+cY=
 github.com/k8shell-io/crypto v0.41.1-ssh-proxy h1:8+q6Ofc2ky23Oc9iNyiq8aeiQBIP+y3+O6zzHqe1f48=
 github.com/k8shell-io/crypto v0.41.1-ssh-proxy/go.mod h1:RVZeOJCpqtogniULztSXQESKJCfcI8WCxsS0FagMA8U=
 github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=

internal/server/auth.go

@@ -253,6 +253,9 @@ func (s *Server) updateUser(ctx context.Context, connInfo *Connection) {
 	if err != nil {
 		if status.Code(err) != codes.NotFound {
 			s.log.Error().Msgf("Failed to get user %s: %v", connInfo.userStr.Username(), err)
+			if status.Code(err) == codes.Unavailable {
+				connInfo.suppressFailureEvent = true
+			}
 			return
 		}
 		// when user is not found, we will check onboarding capability

internal/server/connection.go

@@ -25,7 +25,6 @@ import (
 	"github.com/k8shell-io/common/pkg/api/client/k8shelld"
 	sessionc "github.com/k8shell-io/common/pkg/api/client/session"
 	identityv1 "github.com/k8shell-io/common/pkg/api/gen/go/identity/v1"
-	provisionerv1 "github.com/k8shell-io/common/pkg/api/gen/go/provisioner/v1"
 	sessionv1 "github.com/k8shell-io/common/pkg/api/gen/go/session/v1"
 	"github.com/k8shell-io/common/pkg/authz"
 	"github.com/k8shell-io/common/pkg/gapi"
@@ -38,40 +37,41 @@ import (
 
 // Connection represents the connection information for a user
 type Connection struct {
-	ctx              context.Context               // context for managing the connection
-	seqNumberGen     int64                         // sequence number for exec commands
-	connId           string                        // session key for the connection
-	connKey          string                        // key under which this Connection is stored in connStates
-	cancel           context.CancelFunc            // function to cancel the context
-	log              *zerolog.Logger               // logger instance, reused from server
-	transport        io.Closer                     // underlying SSH transport; set once the handshake completes, used to force-terminate the session
-	userStr          *userstr.UserStr              // user string information
-	clientIP         string                        // client IP address (detected from proxy protocol if available)
-	clientPort       int                           // client port (detected from proxy protocol if available)
-	identity         *identity.IdentityClient      // identity client for interacting with the identity service
-	sessionClient    *sessionc.Client              // gRPC client for session tracking
-	k8shelldCfg      gapi.ClientConfig             // k8shelld client configuration
-	k8shelld         workspace.K8shelldClient      // client for interacting with the workspace k8shelld daemon
-	k8shelldVer      string                        // version of the k8shelld daemon
-	onboardMu        sync.RWMutex                  // mutex for synchronizing access to onboardInfo and onboardCap
-	onboardCap       *models.OnboardCapability     // onboarding capabilities
-	onboardInfo      *models.OnboardUserDeviceFlow // onboarding information
-	user             *models.User                  // user information
-	mu               sync.RWMutex                  // mutex for synchronizing access
-	session          *Session                      // SSH session information
-	directTCPIP      *sync.Map                     // direct TCP/IP connection information
-	directTCPIPCount int64                         // current count of direct TCP/IP connections
-	counters         *k8shelld.ConnCounters        // connection counters
-	workspaceName    string                        // name of the workspace
-	channelInfoMu    sync.RWMutex                  // mutex for synchronizing access to channelInfo
-	channelInfo      []string                      // channel information
-	failureInfo      []string                      // failure information
-	reportStopCh     chan struct{}                 // channel to signal report goroutine to stop
-	reportWg         sync.WaitGroup                // wait group for report goroutine
-	ptyName          string                        // name of the allocated pseudo-terminal (if any)
-	authMethodsMu    sync.RWMutex                  // mutex for synchronizing access to authMethods
-	authMethods      []authz.UserAuthMethod        // SSH authentication methods permitted by policy (resolved once per connection)
-	authMethodsSet   bool                          // whether authMethods has been resolved
+	ctx                  context.Context               // context for managing the connection
+	seqNumberGen         int64                         // sequence number for exec commands
+	connId               string                        // session key for the connection
+	connKey              string                        // key under which this Connection is stored in connStates
+	cancel               context.CancelFunc            // function to cancel the context
+	log                  *zerolog.Logger               // logger instance, reused from server
+	transport            io.Closer                     // underlying SSH transport
+	userStr              *userstr.UserStr              // user string information
+	clientIP             string                        // client IP address (detected from proxy protocol if available)
+	clientPort           int                           // client port (detected from proxy protocol if available)
+	identity             *identity.IdentityClient      // identity client for interacting with the identity service
+	sessionClient        *sessionc.Client              // gRPC client for session tracking
+	k8shelldCfg          gapi.ClientConfig             // k8shelld client configuration
+	k8shelld             workspace.K8shelldClient      // client for interacting with the workspace k8shelld daemon
+	k8shelldVer          string                        // version of the k8shelld daemon
+	onboardMu            sync.RWMutex                  // mutex for synchronizing access to onboardInfo and onboardCap
+	onboardCap           *models.OnboardCapability     // onboarding capabilities
+	onboardInfo          *models.OnboardUserDeviceFlow // onboarding information
+	user                 *models.User                  // user information
+	mu                   sync.RWMutex                  // mutex for synchronizing access
+	session              *Session                      // SSH session information
+	directTCPIP          *sync.Map                     // direct TCP/IP connection information
+	directTCPIPCount     int64                         // current count of direct TCP/IP connections
+	counters             *k8shelld.ConnCounters        // connection counters
+	workspaceName        string                        // name of the workspace
+	channelInfoMu        sync.RWMutex                  // mutex for synchronizing access to channelInfo
+	channelInfo          []string                      // channel information
+	failureInfo          []string                      // failure information
+	reportStopCh         chan struct{}                 // channel to signal report goroutine to stop
+	reportWg             sync.WaitGroup                // wait group for report goroutine
+	ptyName              string                        // name of the allocated pseudo-terminal (if any)
+	authMethodsMu        sync.RWMutex                  // mutex for synchronizing access to authMethods
+	authMethods          []authz.UserAuthMethod        // SSH authentication methods permitted by policy
+	authMethodsSet       bool                          // whether authMethods has been resolved
+	suppressFailureEvent bool                          // when true, skip publishing an SSH failure event to NATS
 }
 
 // Session holds information about a user's SSH session
@@ -248,6 +248,14 @@ func (c *Connection) AddFailureInfo(info string, err error) {
 	}
 }
 
+// ShouldSuppressFailureEvent reports whether the SSH failure event should be
+// suppressed for this connection.
+func (c *Connection) ShouldSuppressFailureEvent() bool {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	return c.suppressFailureEvent
+}
+
 // AddChannelInfo appends channel information to the Connection object
 func (c *Connection) AddChannelInfo(info string) {
 	c.channelInfoMu.Lock()
@@ -538,16 +546,6 @@ func (c *Connection) Handshake(writer io.Writer, writerOptions *workspace.InfoWr
 		return nil, fmt.Errorf("handshake with k8shelld failed for user %s", c.user.Username)
 	}
 
-	go func() {
-		c.log.Debug().Msgf("Running k8shelld command processor for user %s", c.user.Username)
-		err = k8shelld.RunCommandProcessor(c.ctx, c.getCommandHandler(backends, status.Name))
-		if err != nil {
-			c.log.Error().Msgf("Failed to run k8shelld command processor for user %s: %v", c.user.Username, err)
-		} else {
-			c.log.Debug().Msgf("k8shelld command processor stopped for user %s", c.user.Username)
-		}
-	}()
-
 	c.k8shelld = k8shelld
 	c.k8shelldVer = status.AppVersion
 	c.workspaceName = status.Name
@@ -560,50 +558,6 @@ func (c *Connection) Handshake(writer io.Writer, writerOptions *workspace.InfoWr
 	return c.k8shelld, nil
 }
 
-func (c *Connection) getCommandHandler(backends workspace.Backends, workspaceName string) k8shelld.CommandHandler {
-	return func(ctx context.Context, command string) (string, error) {
-		parts := strings.SplitN(command, " ", 2)
-		switch parts[0] {
-		case "shutdown":
-			action := "stop"
-			if len(parts) == 2 {
-				action = parts[1]
-			}
-			c.log.Debug().Msgf("Received k8shelld shutdown command (action=%s) for user %s, workspace %s",
-				action, c.user.Username, workspaceName)
-			switch action {
-			case "delete":
-				_, err := backends.Provisioner().DeleteWorkspace(c.ctx,
-					&provisionerv1.DeleteWorkspaceRequest{Workspace: workspaceName, DelaySeconds: 2})
-				if err != nil {
-					c.log.Debug().Msgf("Failed to delete workspace for user %s, workspace %s: %v",
-						c.user.Username, workspaceName, err)
-					return "Cannot delete workspace due to an error.",
-						fmt.Errorf("failed to delete workspace: %w", err)
-				}
-				return "Workspace deletion has been initiated.", nil
-			case "stop":
-				_, err := backends.Provisioner().StopWorkspace(c.ctx,
-					&provisionerv1.StopWorkspaceRequest{Workspace: workspaceName, DelaySeconds: 2})
-				if err != nil {
-					c.log.Debug().Msgf("Failed to stop workspace for user %s, workspace %s: %v",
-						c.user.Username, workspaceName, err)
-					return "Cannot stop workspace due to an error.",
-						fmt.Errorf("failed to stop workspace: %w", err)
-				}
-				return "Workspace stop has been initiated.", nil
-			default:
-				c.log.Error().Msgf("Received unknown shutdown action %q for user %s, workspace %s",
-					action, c.user.Username, workspaceName)
-				return "", fmt.Errorf("unknown shutdown action %q, expected \"delete\" or \"stop\"", action)
-			}
-		}
-		c.log.Error().Msgf("Received unknown k8shelld command %q for user %s, workspace %s",
-			command, c.user.Username, workspaceName)
-		return "", fmt.Errorf("unknown command")
-	}
-}
-
 // IncrementDirectTCPIPCount atomically increments the direct TCP/IP count
 func (c *Connection) IncrementDirectTCPIPCount(maxLimit int) bool {
 	for {

internal/server/ssh.go

@@ -356,7 +356,7 @@ func (s *Server) handleConnection(netConn net.Conn, isDirect bool) {
 		if connInfo != nil {
 			s.log.Debug().Msgf("Failed connection info: %v", connInfo.failureInfo)
 
-			if s.fpub != nil {
+			if s.fpub != nil && !connInfo.ShouldSuppressFailureEvent() {
 				failureInfo := []string{}
 				failureInfo = append(failureInfo, connInfo.failureInfo...)
 				failureInfo = append(failureInfo, string(err.Error()))

internal/workspace/client.go

@@ -32,7 +32,6 @@ type K8shelldClient interface {
 		enableRecording bool) (int32, error)
 	RunSFTP(ctx context.Context, userToken string, asUser string, upstream k8shelldClient.BufferedReadWriter,
 		sessionID string, command string, envVars []string, enableRecording bool) (int32, error)
-	RunCommandProcessor(ctx context.Context, handler k8shelldClient.CommandHandler) error
 	Close() error
 }