From 68c08cc5e2da8e2972add46b47851900f1b2ebf4 Mon Sep 17 00:00:00 2001 From: n/a Date: Wed, 12 Aug 2026 22:57:39 +0200 Subject: [PATCH 1/3] command listener removed --- go.mod | 2 +- go.sum | 4 +-- internal/server/connection.go | 55 ----------------------------------- internal/workspace/client.go | 1 - 4 files changed, 3 insertions(+), 59 deletions(-) diff --git a/go.mod b/go.mod index eebd920..7294c8f 100644 --- a/go.mod +++ b/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.39.0 github.com/nats-io/nats.go v1.47.0 github.com/rs/zerolog v1.34.0 golang.org/x/crypto v0.43.0 diff --git a/go.sum b/go.sum index 5751b9a..8809234 100644 --- a/go.sum +++ b/go.sum @@ -31,10 +31,10 @@ 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/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= diff --git a/internal/server/connection.go b/internal/server/connection.go index a4da7fd..efe3e79 100644 --- a/internal/server/connection.go +++ b/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" @@ -538,16 +537,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 +549,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 { diff --git a/internal/workspace/client.go b/internal/workspace/client.go index d4201ec..fe549a2 100644 --- a/internal/workspace/client.go +++ b/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 } From 032810b6a79f13c9942e26139786484254ca2441 Mon Sep 17 00:00:00 2001 From: n/a Date: Mon, 17 Aug 2026 17:33:49 +0200 Subject: [PATCH 2/3] [skip ci] suppress option failure nats publish --- internal/server/auth.go | 3 ++ internal/server/connection.go | 77 +++++++++++++++++++---------------- internal/server/ssh.go | 2 +- 3 files changed, 47 insertions(+), 35 deletions(-) diff --git a/internal/server/auth.go b/internal/server/auth.go index 925cdaa..d7d71f7 100644 --- a/internal/server/auth.go +++ b/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 diff --git a/internal/server/connection.go b/internal/server/connection.go index efe3e79..70bbb68 100644 --- a/internal/server/connection.go +++ b/internal/server/connection.go @@ -37,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 @@ -247,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() diff --git a/internal/server/ssh.go b/internal/server/ssh.go index 718a48c..ba0a75b 100644 --- a/internal/server/ssh.go +++ b/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())) From 6fda65e02999a88b7d11c29f8e99d8261ff83fb9 Mon Sep 17 00:00:00 2001 From: n/a Date: Mon, 17 Aug 2026 23:20:59 +0200 Subject: [PATCH 3/3] common v bump --- go.mod | 2 +- go.sum | 2 ++ 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/go.mod b/go.mod index 7294c8f..73671a4 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module github.com/k8shell-io/ssh-proxy go 1.24.5 require ( - github.com/k8shell-io/common v0.39.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 diff --git a/go.sum b/go.sum index 8809234..08ee3f0 100644 --- a/go.sum +++ b/go.sum @@ -35,6 +35,8 @@ github.com/k8shell-io/common v0.37.0 h1:whq66WosIJECKErUKZF1RQep7tdpOfI6GtP4hXRE 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=