Skip to content

fix(ssh): sync agent exiting code with data pipe #4880

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
gustavosbarreto merged 4 commits into from
Jun 3, 2025
Merged

fix(ssh): sync agent exiting code with data pipe #4880

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
b100ba5
refactor(ssh): use multiwriter to record session on cloud
henrybarreto Jun 2, 2025
f82fe84
feat(ssh): split ssh requests to avoid client closed before piping done
henrybarreto Jun 3, 2025
f9a1178
fix(ssh): avoid panic when session's event connection is closed
henrybarreto Jun 3, 2025
4e91e9d
fix(ssh): don't record when session doesn't have pty
henrybarreto Jun 3, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
301 changes: 173 additions & 128 deletions ssh/server/channels/session.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,7 +89,7 @@ func DefaultSessionHandler() gliderssh.ChannelHandler {
sess, _ := session.ObtainSession(ctx)

go func() {
// NOTICE: As [gossh.ServerConn] is shared by all channels calls, close it after a channel close block any
// NOTE: As [gossh.ServerConn] is shared by all channels calls, close it after a channel close block any
// other channel invocation. To avoid it, we wait for the connection to be closed to finish the session.
conn.Wait() //nolint:errcheck

Expand Down Expand Up @@ -139,186 +139,231 @@ func DefaultSessionHandler() gliderssh.ChannelHandler {

defer agent.Close()

var wg sync.WaitGroup

done := make(chan bool)

oncePipe := sync.OnceFunc(func() {
go pipe(sess, client.Channel, agent.Channel, seat)
go pipe(sess, client.Channel, agent.Channel, seat, done)
})

// TODO: Add middleware to block certain types of requests.
for {
select {
case <-ctx.Done():
logger.Info("context has done")

return
case req, ok := <-sess.Agent.Requests:
if !ok {
logger.Trace("global requests is closed")
wg.Add(3)

return
}
go func() {
defer wg.Done()

logger.Debugf("global request from agent: %s", req.Type)
for {
select {
case <-ctx.Done():
logger.Info("context has done (global requests)")

switch {
// NOTICE: The Agent sends "keepalive" requests to the server to avoid the Web Socket being closed due
// to inactivity. Through the time, the request type sent from agent to server changed its name, but
// always keeping the prefix "keepalive". So, to maintain the retro compatibility, we check if this
// prefix exists and perform the necessary operations.
case strings.HasPrefix(req.Type, KeepAliveRequestTypePrefix):
if _, err := client.Channel.SendRequest(KeepAliveRequestType, req.WantReply, req.Payload); err != nil {
logger.Error("failed to send the keepalive request received from agent to client")
return
case req, ok := <-sess.Agent.Requests:
if !ok {
logger.Trace("global requests is closed")

return
}

if err := sess.KeepAlive(); err != nil {
logger.WithError(err).Error("failed to send the API request to inform that the session is open")
logger.Debugf("global request from agent: %s", req.Type)

return
}
default:
if req.WantReply {
if err := req.Reply(false, nil); err != nil {
logger.WithError(err).Error(err)
switch {
// NOTE: The Agent sends "keepalive" requests to the server to avoid the Web Socket being closed due
// to inactivity. Through the time, the request type sent from agent to server changed its name, but
// always keeping the prefix "keepalive". So, to maintain the retro compatibility, we check if this
// prefix exists and perform the necessary operations.
case strings.HasPrefix(req.Type, KeepAliveRequestTypePrefix):
if _, err := client.Channel.SendRequest(KeepAliveRequestType, req.WantReply, req.Payload); err != nil {
logger.Error("failed to send the keepalive request received from agent to client")

return
}

if err := sess.KeepAlive(); err != nil {
logger.WithError(err).Error("failed to send the API request to inform that the session is open")

return
}
default:
if req.WantReply {
if err := req.Reply(false, nil); err != nil {
logger.WithError(err).Error(err)
}
}
}
}
case req, ok := <-agent.Requests:
if !ok {
logger.Trace("agent requests is closed")
}
}()

go func() {
defer wg.Done()
defer func() {
logger.Debug("agent waiting for data done to close client")

<-done
client.Close()
}()

for {
select {
case <-ctx.Done():
logger.Info("context has done (agent requests)")

return
}
case req, ok := <-agent.Requests:
if !ok {
logger.Trace("agent requests is closed")

switch req.Type {
case ExitStatusRequest:
session.Event[models.SSHExitStatus](sess, req.Type, req.Payload, seat)
case ExitSignalRequest:
session.Event[models.SSHSignal](sess, req.Type, req.Payload, seat)
default:
sess.Event(req.Type, req.Payload, seat)
}
return
}

switch req.Type {
case ExitStatusRequest:
session.Event[models.SSHExitStatus](sess, req.Type, req.Payload, seat)
case ExitSignalRequest:
session.Event[models.SSHSignal](sess, req.Type, req.Payload, seat)
default:
sess.Event(req.Type, req.Payload, seat)
}

logger.Debugf("request from agent to client: %s", req.Type)
logger.Debugf("request from agent to client: %s", req.Type)

ok, err := client.Channel.SendRequest(req.Type, req.WantReply, req.Payload)
if err != nil {
logger.WithError(err).Error("failed to send the request from agent to client")
ok, err := client.Channel.SendRequest(req.Type, req.WantReply, req.Payload)
if err != nil {
logger.WithError(err).Error("failed to send the request from agent to client")

continue
}
continue
}

if req.WantReply {
if err := req.Reply(ok, nil); err != nil {
logger.WithError(err).Error(err)
if req.WantReply {
if err := req.Reply(ok, nil); err != nil {
logger.WithError(err).Error(err)
}
}
}
case req, ok := <-client.Requests:
if !ok {
logger.Trace("client requests is closed")
}
}()

go func() {
defer wg.Done()

for {
select {
case <-ctx.Done():
logger.Info("context has done (client requests)")

return
}
case req, ok := <-client.Requests:
if !ok {
logger.Trace("client requests is closed")

switch req.Type {
case ShellRequestType:
if sess.Pty.Term != "" {
if err := sess.Announce(client.Channel); err != nil {
logger.WithError(err).Warn("failed to get the namespace announcement")
}
return
}

sess.Event(req.Type, req.Payload, seat)
case ExecRequestType, SubsystemRequestType:
session.Event[models.SSHCommand](sess, req.Type, req.Payload, seat)
switch req.Type {
case ShellRequestType:
if sess.Pty.Term != "" {
if err := sess.Announce(client.Channel); err != nil {
logger.WithError(err).Warn("failed to get the namespace announcement")
}
}

sess.Type = ExecRequestType
case PtyRequestType:
var pty models.SSHPty
sess.Event(req.Type, req.Payload, seat)
case ExecRequestType, SubsystemRequestType:
session.Event[models.SSHCommand](sess, req.Type, req.Payload, seat)

if err := gossh.Unmarshal(req.Payload, &pty); err != nil {
reject(nil, "failed to recover the session dimensions")
}
sess.Type = ExecRequestType
case PtyRequestType:
var pty models.SSHPty

sess.Pty = pty
if err := gossh.Unmarshal(req.Payload, &pty); err != nil {
reject(nil, "failed to recover the session dimensions")
}

sess.Event(req.Type, pty, seat) //nolint:errcheck
case WindowChangeRequestType:
var dimensions models.SSHWindowChange
sess.Pty = pty

if err := gossh.Unmarshal(req.Payload, &dimensions); err != nil {
reject(nil, "failed to recover the session dimensions")
}
sess.Event(req.Type, pty, seat) //nolint:errcheck
case WindowChangeRequestType:
var dimensions models.SSHWindowChange

sess.Pty.Columns = dimensions.Columns
sess.Pty.Rows = dimensions.Rows
if err := gossh.Unmarshal(req.Payload, &dimensions); err != nil {
reject(nil, "failed to recover the session dimensions")
}

sess.Event(req.Type, dimensions, seat) //nolint:errcheck
case AuthRequestOpenSSHRequest:
gliderssh.SetAgentRequested(ctx)
sess.Pty.Columns = dimensions.Columns
sess.Pty.Rows = dimensions.Rows

sess.Event(req.Type, req.Payload, seat)
go func() {
clientConn := ctx.Value(gliderssh.ContextKeyConn).(gossh.Conn)
agentChannels := sess.Agent.Client.HandleChannelOpen(AuthRequestOpenSSHChannel)
sess.Event(req.Type, dimensions, seat) //nolint:errcheck
case AuthRequestOpenSSHRequest:
gliderssh.SetAgentRequested(ctx)

for {
newAgentChannel, ok := <-agentChannels
if !ok {
reject(nil, "channel for agent forwarding done")
sess.Event(req.Type, req.Payload, seat)
go func() {
clientConn := ctx.Value(gliderssh.ContextKeyConn).(gossh.Conn)
agentChannels := sess.Agent.Client.HandleChannelOpen(AuthRequestOpenSSHChannel)

return
}
for {
newAgentChannel, ok := <-agentChannels
if !ok {
reject(nil, "channel for agent forwarding done")

agentChannel, agentReqs, err := newAgentChannel.Accept()
if err != nil {
reject(nil, "failed to accept the chanel request from agent on auth request")
return
}

return
}
agentChannel, agentReqs, err := newAgentChannel.Accept()
if err != nil {
reject(nil, "failed to accept the chanel request from agent on auth request")

defer agentChannel.Close()
go gossh.DiscardRequests(agentReqs)
return
}

clientChannel, clientReqs, err := clientConn.OpenChannel(AuthRequestOpenSSHChannel, nil)
if err != nil {
reject(nil, "failed to open the auth request channel from agent to client")
defer agentChannel.Close()
go gossh.DiscardRequests(agentReqs)

return
}
clientChannel, clientReqs, err := clientConn.OpenChannel(AuthRequestOpenSSHChannel, nil)
if err != nil {
reject(nil, "failed to open the auth request channel from agent to client")

defer clientChannel.Close()
go gossh.DiscardRequests(clientReqs)
return
}

hose(sess, agentChannel, clientChannel)
defer clientChannel.Close()
go gossh.DiscardRequests(clientReqs)

logger.WithError(err).Trace("auth request channel piping done")
}
}()
default:
sess.Event(req.Type, req.Payload, seat)
}
hose(sess, agentChannel, clientChannel)

logger.Debugf("request from client to agent: %s", req.Type)
logger.WithError(err).Trace("auth request channel piping done")
}
}()
default:
sess.Event(req.Type, req.Payload, seat)
}

ok, err := agent.Channel.SendRequest(req.Type, req.WantReply, req.Payload)
if err != nil {
logger.WithError(err).Error("failed to send the request from client to agent")
logger.Debugf("request from client to agent: %s", req.Type)

continue
}
ok, err := agent.Channel.SendRequest(req.Type, req.WantReply, req.Payload)
if err != nil {
logger.WithError(err).Error("failed to send the request from client to agent")

if req.WantReply {
if err := req.Reply(ok, nil); err != nil {
logger.WithError(err).Error(err)
continue
}

if req.WantReply {
if err := req.Reply(ok, nil); err != nil {
logger.WithError(err).Error(err)
}
}
}

switch req.Type {
case PtyRequestType, ExecRequestType, SubsystemRequestType:
oncePipe()
switch req.Type {
case PtyRequestType, ExecRequestType, SubsystemRequestType:
oncePipe()
}
}
}
}
}()

wg.Wait()

logger.Debug("session done after waiting")
}
}
Loading
Loading