diff options
Diffstat (limited to 'internal/server')
| -rw-r--r-- | internal/server/background/background.go | 126 | ||||
| -rw-r--r-- | internal/server/handlers/controlhandler.go | 2 | ||||
| -rw-r--r-- | internal/server/handlers/runcommand.go | 118 | ||||
| -rw-r--r-- | internal/server/handlers/serverhandler.go | 269 | ||||
| -rw-r--r-- | internal/server/server.go | 50 |
5 files changed, 465 insertions, 100 deletions
diff --git a/internal/server/background/background.go b/internal/server/background/background.go new file mode 100644 index 0000000..225b82a --- /dev/null +++ b/internal/server/background/background.go @@ -0,0 +1,126 @@ +package background + +import ( + "context" + "errors" + "fmt" + "strings" + "sync" + + "github.com/mimecast/dtail/internal/io/logger" +) + +type job struct { + cancel context.CancelFunc + wg *sync.WaitGroup +} + +// Background specifies a job or command run in background on server side. +// This does not require an active DTail client SSH connection/session. +type Background struct { + mutex *sync.Mutex + jobs map[string]job +} + +// New returns a new background manager. +func New() Background { + return Background{ + jobs: make(map[string]job), + mutex: &sync.Mutex{}, + } +} + +// Add a background job. +func (b Background) Add(userName, jobName string, cancel context.CancelFunc, wg *sync.WaitGroup) error { + key := b.key(userName, jobName) + logger.Debug("background", "Add", key) + + b.mutex.Lock() + defer b.mutex.Unlock() + + if _, ok := b.jobs[key]; ok { + return errors.New("job already exists") + } + + b.jobs[key] = job{cancel, wg} + + // Clean up background job database. + go func() { + wg.Wait() + b.cancel(key) + }() + + return nil +} + +// Cancel a background job. +func (b Background) Cancel(userName, jobName string) error { + key := b.key(userName, jobName) + logger.Debug("background", "Cancel", key) + + return b.cancel(key) +} + +func (b Background) cancel(key string) error { + job, ok := b.get(key) + logger.Debug("background", "cancel", key, job, ok) + + if !ok { + return errors.New("no job to cancel") + } + + logger.Debug("background", "cancel", "run job.cancel()") + job.cancel() + logger.Debug("background", "cancel", "run job.wg.Wait()") + job.wg.Wait() + logger.Debug("background", "cancel", "run b.delete(key)") + b.delete(key) + + return nil +} + +// ListJobsC returns a channel listing all jobs of the given user. +func (b Background) ListJobsC(userName string) <-chan string { + logger.Debug("background", "ListJobC", userName) + + ch := make(chan string) + + go func() { + defer close(ch) + + b.mutex.Lock() + defer b.mutex.Unlock() + + for k, _ := range b.jobs { + logger.Debug("ListJobsC", k, userName) + if strings.HasPrefix(k, fmt.Sprintf("%s.", userName)) { + ch <- k + } + } + }() + + return ch +} + +func (b Background) get(key string) (job, bool) { + logger.Debug("background", "get", key) + + b.mutex.Lock() + defer b.mutex.Unlock() + + job, ok := b.jobs[key] + return job, ok +} + +func (b Background) delete(key string) { + logger.Debug("background", "delete", key) + + b.mutex.Lock() + defer b.mutex.Unlock() + + delete(b.jobs, key) +} + +func (Background) key(userName, jobName string) string { + return fmt.Sprintf("%s.%s", userName, jobName) +} diff --git a/internal/server/handlers/controlhandler.go b/internal/server/handlers/controlhandler.go index a33a78b..daa9835 100644 --- a/internal/server/handlers/controlhandler.go +++ b/internal/server/handlers/controlhandler.go @@ -87,6 +87,6 @@ func (h *ControlHandler) handleCommand(ctx context.Context, command string) { case "debug": h.serverMessages <- logger.Debug(h.user, "Receiving debug command", command, s) default: - h.serverMessages <- logger.Warn(h.user, "Received unknown command", command, s) + h.serverMessages <- logger.Warn(h.user, "Received unknown control command", command, s) } } diff --git a/internal/server/handlers/runcommand.go b/internal/server/handlers/runcommand.go index e260060..8e5895b 100644 --- a/internal/server/handlers/runcommand.go +++ b/internal/server/handlers/runcommand.go @@ -2,10 +2,16 @@ package handlers import ( "context" + "errors" "fmt" + "io/ioutil" + "os" "os/exec" "strings" + "sync" + "time" + "github.com/mimecast/dtail/internal/config" "github.com/mimecast/dtail/internal/io/logger" "github.com/mimecast/dtail/internal/io/run" ) @@ -21,53 +27,85 @@ func newRunCommand(server *ServerHandler) runCommand { } } -func (r runCommand) Start(ctx context.Context, argc int, args []string) { +func (r runCommand) StartBackground(ctx context.Context, wg *sync.WaitGroup, argc int, args, outerArgs []string) error { if argc < 2 { - r.server.sendServerMessage(logger.Warn(r.server.user, commandParseWarning, args, argc)) - return + return fmt.Errorf("%s: args:%v argc:%d", commandParseWarning, args, argc) } - commands := strings.Split(strings.Join(args[1:], " "), ";") - r.start(ctx, commands) -} -func (r runCommand) start(ctx context.Context, commands []string) { - for _, command := range commands { - command = strings.TrimSpace(command) - if len(command) == 0 { - continue - } - splitted := strings.Split(command, " ") - path := splitted[0] - args := splitted[1:] - - qualifiedPath, err := exec.LookPath(path) - if err != nil { - logger.Error(r.server.user, err) - r.server.sendServerMessage(logger.Warn(r.server.user, "Unable to execute command(s), check server logs")) - r.server.sendServerMessage(fmt.Sprintf(".run exitstatus -%d", -1)) - return - } + ec := make(chan int, 1) + var pid int + var err error - if !r.server.user.HasFilePermission(qualifiedPath, "runcommands") { - logger.Error(r.server.user, "No permission to execute path", qualifiedPath) - r.server.sendServerMessage(logger.Warn(r.server.user, "Unable to execute command(s), check server logs")) - r.server.sendServerMessage(fmt.Sprintf(".run exitstatus -%d", -1)) - return + command := strings.Join(args[1:], " ") + if strings.Contains(command, ";") || strings.Contains(command, "\n") { + if pid, err = r.startScript(ctx, wg, ec, command, outerArgs); err != nil { + r.server.sendServerMessage(".run exitstatus 255") + return err } + return nil + } - r.run = run.New(qualifiedPath, args) - pid, ec, err := r.run.Start(ctx, r.server.lines) + if pid, err = r.start(ctx, wg, ec, strings.TrimSpace(command), outerArgs); err != nil { + r.server.sendServerMessage(".run exitstatus 255") + return err + } - if err != nil { - message := fmt.Sprintf("Unable to execute remote command '%s'", command) - logger.Error(r.server.user, message, ec, pid, err) - r.server.sendServerMessage(logger.Error(message, ec, pid, err)) - r.server.sendServerMessage(fmt.Sprintf(".run exitstatus -%d", ec)) - return - } + exitCode := <-ec + r.server.sendServerMessage(fmt.Sprintf(".run exitstatus %d", exitCode)) + r.server.sendServerMessage(logger.Info(fmt.Sprintf("Process %d exited with status %d", pid, exitCode))) + + return nil +} + +func (r runCommand) startScript(ctx context.Context, wg *sync.WaitGroup, ec chan<- int, script string, outerArgs []string) (int, error) { + if _, err := os.Stat(config.Common.TmpDir); os.IsNotExist(err) { + return -1, err + } + + timestamp := time.Now().UnixNano() + scriptPath := fmt.Sprintf("%s/%s_%v.sh", config.Common.TmpDir, r.server.user.Name, timestamp) + + // TODO: On dserver startup delete all previously written scripts (there might be left overs due to a crash or so) + logger.Debug(r.server.user, "Writing temp script", scriptPath) + + script = fmt.Sprintf("#!/bin/sh\n%s", script) + if err := ioutil.WriteFile(scriptPath, []byte(script), 0700); err != nil { + return -1, err + } + + pid, err := r.start(ctx, wg, ec, scriptPath, outerArgs) + go func() { + wg.Wait() + logger.Debug("Deleting script", scriptPath) + os.Remove(scriptPath) + }() + + return pid, err +} + +func (r runCommand) start(ctx context.Context, wg *sync.WaitGroup, ec chan<- int, command string, outerArgs []string) (int, error) { + if len(command) == 0 { + return -1, errors.New("Empty command provided") + } + + splitted := strings.Split(command, " ") + path := splitted[0] + args := splitted[1:] + args = append(args, outerArgs...) + + qualifiedPath, err := exec.LookPath(path) + if err != nil { + return -1, err + } + + if !r.server.user.HasFilePermission(qualifiedPath, "runcommands") { + return -1, fmt.Errorf("No permission to execute path: %s", qualifiedPath) + } - message := fmt.Sprintf("Remote process '%d' exited with status '%d'", pid, ec) - r.server.sendServerMessage(fmt.Sprintf(".run exitstatus %d", ec)) - r.server.sendServerMessage(logger.Info("run", pid, ec, message)) + r.run = run.New(qualifiedPath, args) + pid, err := r.run.StartBackground(ctx, wg, ec, r.server.lines) + if err != nil { + return pid, err } + return pid, nil } diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go index 2979dd5..939388c 100644 --- a/internal/server/handlers/serverhandler.go +++ b/internal/server/handlers/serverhandler.go @@ -7,8 +7,10 @@ import ( "fmt" "io" "os" + "strconv" "strings" "sync" + "sync/atomic" "time" "github.com/mimecast/dtail/internal/config" @@ -16,6 +18,7 @@ import ( "github.com/mimecast/dtail/internal/io/logger" "github.com/mimecast/dtail/internal/mapr/server" "github.com/mimecast/dtail/internal/omode" + "github.com/mimecast/dtail/internal/server/background" user "github.com/mimecast/dtail/internal/user/server" "github.com/mimecast/dtail/internal/version" ) @@ -28,7 +31,6 @@ const ( // the Bi-directional communication between SSH client and server. // This handler implements the handler of the SSH server. type ServerHandler struct { - mutex *sync.Mutex lines chan line.Line regex string aggregate *server.Aggregate @@ -37,28 +39,35 @@ type ServerHandler struct { payload []byte hostname string user *user.User - catLimiter chan struct{} - tailLimiter chan struct{} - ackCloseReceived chan struct{} - ctx context.Context - done chan struct{} - activeReaders int + // TODO: Move all these channels into a separate struct for readability! + catLimiter chan struct{} + tailLimiter chan struct{} + globalServerWaitFor chan struct{} + ackCloseReceived chan struct{} + serverCtx context.Context + handlerCtx context.Context + done chan struct{} + activeCommands int32 + activeReaders int32 + background background.Background } // NewServerHandler returns the server handler. -func NewServerHandler(ctx context.Context, user *user.User, catLimiter chan struct{}, tailLimiter chan struct{}) (*ServerHandler, <-chan struct{}) { +func NewServerHandler(handlerCtx, serverCtx context.Context, user *user.User, catLimiter, tailLimiter, globalServerWaitFor chan struct{}, background background.Background) (*ServerHandler, <-chan struct{}) { h := ServerHandler{ - ctx: ctx, - done: make(chan struct{}), - mutex: &sync.Mutex{}, - lines: make(chan line.Line, 100), - serverMessages: make(chan string, 10), - aggregatedMessages: make(chan string, 10), - ackCloseReceived: make(chan struct{}), - catLimiter: catLimiter, - tailLimiter: tailLimiter, - regex: ".", - user: user, + serverCtx: serverCtx, + handlerCtx: handlerCtx, + done: make(chan struct{}), + lines: make(chan line.Line, 100), + serverMessages: make(chan string, 10), + aggregatedMessages: make(chan string, 10), + ackCloseReceived: make(chan struct{}), + catLimiter: catLimiter, + tailLimiter: tailLimiter, + globalServerWaitFor: globalServerWaitFor, + regex: ".", + user: user, + background: background, } fqdn, err := os.Hostname() @@ -108,7 +117,7 @@ func (h *ServerHandler) Read(p []byte) (n int, err error) { case <-time.After(time.Second): // Once in a while check whether we are done. select { - case <-h.ctx.Done(): + case <-h.handlerCtx.Done(): return 0, io.EOF default: } @@ -122,7 +131,7 @@ func (h *ServerHandler) Write(p []byte) (n int, err error) { switch c { case ';': commandStr := strings.TrimSpace(string(h.payload)) - h.handleCommand(h.ctx, commandStr) + h.handleCommand(h.handlerCtx, commandStr) h.payload = nil default: h.payload = append(h.payload, c) @@ -135,6 +144,7 @@ func (h *ServerHandler) Write(p []byte) (n int, err error) { func (h *ServerHandler) handleCommand(ctx context.Context, commandStr string) { logger.Debug(h.user, commandStr) + var timeout time.Duration args, argc, err := h.handleProtocolVersion(strings.Split(commandStr, " ")) if err != nil { @@ -148,12 +158,30 @@ func (h *ServerHandler) handleCommand(ctx context.Context, commandStr string) { return } + args, argc, timeout, err = h.handleTimeout(args, argc) + if err != nil { + h.send(h.serverMessages, logger.Error(h.user, err)) + return + } + if h.user.Name == config.ControlUser { h.handleControlCommand(argc, args) return } - h.handleUserCommand(ctx, argc, args) + if timeout > 0 { + logger.Info(h.user, "Command with timeout context", argc, args, timeout) + commandCtx, cancel := context.WithTimeout(ctx, timeout) + go func() { + <-commandCtx.Done() + logger.Info(h.user, "Command timed out, canceling it", args, args, timeout) + cancel() + }() + h.handleUserCommand(commandCtx, argc, args, timeout) + return + } + + h.handleUserCommand(ctx, argc, args, timeout) } func (h *ServerHandler) handleProtocolVersion(args []string) ([]string, int, error) { @@ -191,37 +219,70 @@ func (h *ServerHandler) handleBase64(args []string, argc int) ([]string, int, er return args, argc, nil } +func (h *ServerHandler) handleTimeout(args []string, argc int) ([]string, int, time.Duration, error) { + if argc <= 2 || args[0] != "timeout" { + // No timeout specified + return args, argc, time.Duration(0) * time.Second, nil + } + + timeout, err := strconv.Atoi(args[1]) + return args[2:], argc - 2, time.Duration(timeout) * time.Second, err +} + func (h *ServerHandler) handleControlCommand(argc int, args []string) { switch args[0] { case "debug": h.send(h.serverMessages, logger.Debug(h.user, "Receiving debug command", argc, args)) default: - logger.Warn(h.user, "Received unknown command", argc, args) + logger.Warn(h.user, "Received unknown control command", argc, args) } } -func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args []string) { +func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args []string, timeout time.Duration) { logger.Debug(h.user, "handleUserCommand", argc, args) - switch args[0] { + h.incrementActiveCommands() + commandFinished := func() { + if h.decrementActiveCommands() == 0 { + h.shutdown() + } + } + readerFinished := func() { + if h.decrementActiveReaders() == 0 { + if h.aggregate == nil { + return + } + h.aggregate.Cancel() + } + } + + splitted := strings.Split(args[0], ":") + commandName := splitted[0] + + options, err := readOptions(splitted[1:]) + if err != nil { + h.sendServerMessage(logger.Error(h.user, err)) + commandFinished() + return + } + + switch commandName { case "grep", "cat": command := newReadCommand(h, omode.CatClient) - h.incrementActiveReaders() go func() { + h.incrementActiveReaders() command.Start(ctx, argc, args) - if h.decrementActiveReaders() == 0 { - h.shutdown() - } + readerFinished() + commandFinished() }() case "tail": command := newReadCommand(h, omode.TailClient) - h.incrementActiveReaders() go func() { + h.incrementActiveReaders() command.Start(ctx, argc, args) - if h.decrementActiveReaders() == 0 { - h.shutdown() - } + readerFinished() + commandFinished() }() case "map": @@ -229,31 +290,103 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args [] if err != nil { h.sendServerMessage(err.Error()) logger.Error(h.user, err) - h.shutdown() + commandFinished() return } h.aggregate = aggregate go func() { command.Start(ctx, h.aggregatedMessages) - h.shutdown() + commandFinished() }() case "run": + // TODO: Refactor this "run" case, move code to runcommand.go command := newRunCommand(h) - h.incrementActiveReaders() - go func() { - command.Start(ctx, argc, args) - if h.decrementActiveReaders() == 0 { - h.shutdown() + + jobName, _ := options["jobName"] + logger.Debug(h.user, "run", options) + + if val, ok := options["background"]; ok && (val == "cancel" || val == "stop") { + if err := h.background.Cancel(h.user.Name, jobName); err != nil { + h.sendServerMessage(logger.Error(h.user, err, jobName, args)) + } else { + h.sendServerMessage(logger.Info(h.user, "job cancelled", jobName)) + } + commandFinished() + return + } + + if val, ok := options["background"]; ok && val == "list" { + h.sendServerMessage("Listing jobs") + count := 0 + for jobName := range h.background.ListJobsC(h.user.Name) { + h.sendServerMessage(jobName) + count++ } + h.sendServerMessage(fmt.Sprintf("Found %d jobs", count)) + commandFinished() + return + } + + str, _ := options["outerArgs"] + outerArgs := strings.Split(str, " ") + + var background bool + if val, ok := options["background"]; ok && val == "start" { + background = true + } + + var wg sync.WaitGroup + wg.Add(1) + + if background { + if timeout == 0 { + // Set default background timeout. + timeout = time.Hour * 1 + } + // Use a new context based on the server context, so that background job does not get + // terminated when handler/SSH connection terminates. + commandCtx, cancel := context.WithTimeout(h.serverCtx, timeout) + + if err := h.background.Add(h.user.Name, jobName, cancel, &wg); err != nil { + h.sendServerMessage(logger.Error(h.user, err, jobName, args)) + commandFinished() + return + } + ctx = commandCtx + } + + if err := command.StartBackground(ctx, &wg, argc, args, outerArgs); err != nil { + h.sendServerMessage(logger.Error(h.user, "Unable to execute command", argc, args, err)) + commandFinished() + return + } + + // Make sure that server waits for all sub-processes to finish on shutdown + go func() { h.globalServerWaitFor <- struct{}{} }() + go func() { + wg.Wait() + <-h.globalServerWaitFor }() + if background { + h.sendServerMessage(logger.Info(h.user, jobName, "job started in background")) + commandFinished() + return + } + + // Command run in foreground, wait for it to complete before finishing the connection. + wg.Wait() + commandFinished() + case "ack", ".ack": h.handleAckCommand(argc, args) + commandFinished() default: - h.sendServerMessage(logger.Error(h.user, "Received unknown command", argc, args)) + h.sendServerMessage(logger.Error(h.user, "Received unknown user command", commandName, argc, args, options)) + commandFinished() } } @@ -270,7 +403,7 @@ func (h *ServerHandler) handleAckCommand(argc int, args []string) { func (h *ServerHandler) send(ch chan<- string, message string) { select { case ch <- message: - case <-h.ctx.Done(): + case <-h.handlerCtx.Done(): } } @@ -292,7 +425,6 @@ func (h *ServerHandler) flush() { unsentMessages := func() int { return len(h.lines) + len(h.serverMessages) + len(h.aggregatedMessages) } - for i := 0; i < 3; i++ { if unsentMessages() == 0 { logger.Debug(h.user, "All lines sent") @@ -312,7 +444,7 @@ func (h *ServerHandler) shutdown() { go func() { select { case h.serverMessageC() <- ".syn close connection": - case <-h.ctx.Done(): + case <-h.handlerCtx.Done(): } }() @@ -320,7 +452,7 @@ func (h *ServerHandler) shutdown() { case <-h.ackCloseReceived: case <-time.After(time.Second * 5): logger.Debug(h.user, "Shutdown timeout reached, enforcing shutdown") - case <-h.ctx.Done(): + case <-h.handlerCtx.Done(): } select { @@ -329,17 +461,46 @@ func (h *ServerHandler) shutdown() { } } +func (h *ServerHandler) incrementActiveCommands() { + atomic.AddInt32(&h.activeCommands, 1) +} + +func (h *ServerHandler) decrementActiveCommands() int32 { + atomic.AddInt32(&h.activeCommands, -1) + return atomic.LoadInt32(&h.activeCommands) +} + func (h *ServerHandler) incrementActiveReaders() { - // TODO: Use atomic counter variable instead, so we can get rid of the mutex - h.mutex.Lock() - defer h.mutex.Unlock() - h.activeReaders++ + atomic.AddInt32(&h.activeReaders, 1) +} + +func (h *ServerHandler) decrementActiveReaders() int32 { + atomic.AddInt32(&h.activeReaders, -1) + return atomic.LoadInt32(&h.activeReaders) } -func (h *ServerHandler) decrementActiveReaders() int { - h.mutex.Lock() - defer h.mutex.Unlock() - h.activeReaders-- +func readOptions(opts []string) (map[string]string, error) { + options := make(map[string]string, len(opts)) + + for _, o := range opts { + kv := strings.SplitN(o, "=", 2) + if len(kv) != 2 { + return options, fmt.Errorf("Unable to parse options: %v", kv) + } + key := kv[0] + val := kv[1] + + if strings.HasPrefix(val, "base64%") { + s := strings.SplitN(val, "%", 2) + decoded, err := base64.StdEncoding.DecodeString(s[1]) + if err != nil { + return options, err + } + val = string(decoded) + } + + options[key] = val + } - return h.activeReaders + return options, nil } diff --git a/internal/server/server.go b/internal/server/server.go index 4ffe4d9..8e791c8 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -7,9 +7,11 @@ import ( "io" "net" "strings" + "time" "github.com/mimecast/dtail/internal/config" "github.com/mimecast/dtail/internal/io/logger" + "github.com/mimecast/dtail/internal/server/background" "github.com/mimecast/dtail/internal/server/handlers" "github.com/mimecast/dtail/internal/ssh/server" user "github.com/mimecast/dtail/internal/user/server" @@ -25,11 +27,15 @@ type Server struct { // SSH server configuration. sshServerConfig *gossh.ServerConfig // To control the max amount of concurrent cats (which can cause a lot of I/O on the server) - catLimiterCh chan struct{} + catLimiter chan struct{} // To control the max amount of concurrent tails - tailLimiterCh chan struct{} + tailLimiter chan struct{} // To run scheduled tasks (if configured) sched *scheduler + // Wait counter, e.g. there might be still subprocesses (forked by drun) to be killed. + shutdownWaitFor chan struct{} + // Background jobs + background background.Background } // New returns a new server. @@ -38,9 +44,11 @@ func New() *Server { s := Server{ sshServerConfig: &gossh.ServerConfig{}, - catLimiterCh: make(chan struct{}, config.Server.MaxConcurrentCats), - tailLimiterCh: make(chan struct{}, config.Server.MaxConcurrentTails), + catLimiter: make(chan struct{}, config.Server.MaxConcurrentCats), + tailLimiter: make(chan struct{}, config.Server.MaxConcurrentTails), + shutdownWaitFor: make(chan struct{}, 1000), sched: newScheduler(), + background: background.New(), } s.sshServerConfig.PasswordCallback = s.backgroundUserCallback @@ -61,6 +69,7 @@ func (s *Server) Start(ctx context.Context) int { bindAt := fmt.Sprintf("%s:%d", config.Server.SSHBindAddress, config.Common.SSHPort) logger.Info("Binding server", bindAt) + listener, err := net.Listen("tcp", bindAt) if err != nil { logger.FatalExit("Failed to open listening TCP socket", err) @@ -68,10 +77,40 @@ func (s *Server) Start(ctx context.Context) int { go s.stats.start(ctx) go s.sched.start(ctx) + go s.listenerLoop(ctx, listener) + + select { + case <-ctx.Done(): + // Wait until all commands/jobs/children are no more! + s.wait() + } + + // For future use. + return 0 +} + +func (s *Server) wait() { + for { + num := len(s.shutdownWaitFor) + logger.Debug("Waiting for stuff to finish", num) + if num <= 0 { + return + } + time.Sleep(time.Second) + } +} + +func (s *Server) listenerLoop(ctx context.Context, listener net.Listener) { + logger.Debug("Starting listener loop") for { conn, err := listener.Accept() // Blocking if err != nil { + select { + case <-ctx.Done(): + return + default: + } logger.Error("Failed to accept incoming connection", err) continue } @@ -144,12 +183,13 @@ func (s *Server) handleRequests(ctx context.Context, sshConn gossh.Conn, in <-ch case config.ControlUser: handler, done = handlers.NewControlHandler(handlerCtx, user) default: - handler, done = handlers.NewServerHandler(handlerCtx, user, s.catLimiterCh, s.tailLimiterCh) + handler, done = handlers.NewServerHandler(handlerCtx, ctx, user, s.catLimiter, s.tailLimiter, s.shutdownWaitFor, s.background) } go func() { // Handler finished work, cancel all remaining routines defer cancel() + <-done }() |