From 69b88a1cae0a61bd22530c384f40166b37b9f1ea Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sat, 18 Sep 2021 18:43:19 +0300 Subject: remote connector is now an interface --- internal/clients/connectors/connector.go | 17 ++ internal/clients/connectors/serverconnection.go | 225 ++++++++++++++++++++++++ 2 files changed, 242 insertions(+) create mode 100644 internal/clients/connectors/connector.go create mode 100644 internal/clients/connectors/serverconnection.go (limited to 'internal/clients/connectors') diff --git a/internal/clients/connectors/connector.go b/internal/clients/connectors/connector.go new file mode 100644 index 0000000..3ab6a08 --- /dev/null +++ b/internal/clients/connectors/connector.go @@ -0,0 +1,17 @@ +package connectors + +import ( + "context" + + "github.com/mimecast/dtail/internal/clients/handlers" +) + +// Connector interface. +type Connector interface { + // Start the connection. + Start(ctx context.Context, cancel context.CancelFunc, throttleCh, statsCh chan struct{}) + // Server hostname. + Server() string + // Handler for the connection. + Handler() handlers.Handler +} diff --git a/internal/clients/connectors/serverconnection.go b/internal/clients/connectors/serverconnection.go new file mode 100644 index 0000000..fab2f87 --- /dev/null +++ b/internal/clients/connectors/serverconnection.go @@ -0,0 +1,225 @@ +package connectors + +import ( + "context" + "fmt" + "io" + "strconv" + "strings" + "time" + + "github.com/mimecast/dtail/internal/clients/handlers" + "github.com/mimecast/dtail/internal/config" + "github.com/mimecast/dtail/internal/io/logger" + "github.com/mimecast/dtail/internal/ssh/client" + + "golang.org/x/crypto/ssh" +) + +// ServerConnection represents a client connection connection to a single server. +type ServerConnection struct { + // The remote server's hostname connected to. + server string + // The remote server's port connected to. + port int + // The SSH client configuration used. + config *ssh.ClientConfig + // The SSH client handler to use. + handler handlers.Handler + // DTail commands sent from client to server. When client loses + // connection to the server it re-connects automatically and sends the + // same commands again. + commands []string + // Is it a persistent connection or a one-off? + isOneOff bool + // To deal with SSH server host keys + hostKeyCallback client.HostKeyCallback + // To determine if connection throttling has finished or not + throttlingDone bool +} + +// NewServerConnection returns a new connection. +func NewServerConnection(server string, userName string, authMethods []ssh.AuthMethod, hostKeyCallback client.HostKeyCallback, handler handlers.Handler, commands []string) *ServerConnection { + logger.Debug(server, "Creating new connection") + + c := ServerConnection{ + hostKeyCallback: hostKeyCallback, + server: server, + handler: handler, + commands: commands, + config: &ssh.ClientConfig{ + User: userName, + Auth: authMethods, + HostKeyCallback: hostKeyCallback.Wrap(), + Timeout: time.Second * 2, + }, + } + + c.initServerPort() + return &c +} + +// NewOneOffServerConnection creates new one-off connection (only for sending a series of commands and then quit). +func NewOneOffServerConnection(server string, userName string, authMethods []ssh.AuthMethod, handler handlers.Handler, commands []string) *ServerConnection { + c := ServerConnection{ + server: server, + handler: handler, + commands: commands, + config: &ssh.ClientConfig{ + User: userName, + Auth: authMethods, + HostKeyCallback: ssh.InsecureIgnoreHostKey(), + }, + isOneOff: true, + } + + c.initServerPort() + return &c +} + +// Server hostname +func (c *ServerConnection) Server() string { + return c.server +} + +// Handler for the connection +func (c *ServerConnection) Handler() handlers.Handler { + return c.handler +} + +// Attempt to parse the server port address from the provided server FQDN. +func (c *ServerConnection) initServerPort() { + c.port = config.Common.SSHPort + parts := strings.Split(c.server, ":") + + if len(parts) == 2 { + logger.Debug("Parsing port from hostname", parts) + port, err := strconv.Atoi(parts[1]) + if err != nil { + logger.FatalExit("Unable to parse client port", c.server, parts, err) + } + c.server = parts[0] + c.port = port + } +} + +// Start the server connection. Build up SSH session and send some DTail commands. +func (c *ServerConnection) Start(ctx context.Context, cancel context.CancelFunc, throttleCh, statsCh chan struct{}) { + // Throttle how many connections can be established concurrently (based on ch length) + logger.Debug(c.server, "Throttling connection", len(throttleCh), cap(throttleCh)) + + select { + case throttleCh <- struct{}{}: + case <-ctx.Done(): + logger.Debug(c.server, "Not establishing connection as context is done", len(throttleCh), cap(throttleCh)) + return + } + + logger.Debug(c.server, "Throttling says that the connection can be established", len(throttleCh), cap(throttleCh)) + + go func() { + defer func() { + if !c.throttlingDone { + logger.Debug(c.server, "Unthrottling connection (1)", len(throttleCh), cap(throttleCh)) + c.throttlingDone = true + <-throttleCh + } + cancel() + }() + + if err := c.dial(ctx, cancel, throttleCh, statsCh); err != nil { + logger.Warn(c.server, c.port, err) + if c.hostKeyCallback.Untrusted(fmt.Sprintf("%s:%d", c.server, c.port)) { + logger.Debug(c.server, "Not trusting host") + } + } + }() + + <-ctx.Done() +} + +// Dail into a new SSH connection. Close connection in case of an error. +func (c *ServerConnection) dial(ctx context.Context, cancel context.CancelFunc, throttleCh, statsCh chan struct{}) error { + logger.Debug(c.server, "Incrementing connection stats") + statsCh <- struct{}{} + defer func() { + logger.Debug(c.server, "Decrementing connection stats") + <-statsCh + }() + + logger.Debug(c.server, "Dialing into the connection") + address := fmt.Sprintf("%s:%d", c.server, c.port) + + client, err := ssh.Dial("tcp", address, c.config) + if err != nil { + return err + } + defer client.Close() + + return c.session(ctx, cancel, client, throttleCh) +} + +// Create the SSH session. Close the session in case of an error. +func (c *ServerConnection) session(ctx context.Context, cancel context.CancelFunc, client *ssh.Client, throttleCh chan struct{}) error { + logger.Debug(c.server, "session") + + session, err := client.NewSession() + if err != nil { + return err + } + defer session.Close() + + return c.handle(ctx, cancel, session, throttleCh) +} + +func (c *ServerConnection) handle(ctx context.Context, cancel context.CancelFunc, session *ssh.Session, throttleCh chan struct{}) error { + logger.Debug(c.server, "handle") + + stdinPipe, err := session.StdinPipe() + if err != nil { + return err + } + + stdoutPipe, err := session.StdoutPipe() + if err != nil { + return err + } + + if err := session.Shell(); err != nil { + return err + } + + go func() { + io.Copy(stdinPipe, c.handler) + cancel() + }() + + go func() { + io.Copy(c.handler, stdoutPipe) + cancel() + }() + + go func() { + select { + case <-c.handler.Done(): + case <-ctx.Done(): + } + cancel() + }() + + // Send all commands to client. + for _, command := range c.commands { + logger.Debug(command) + c.handler.SendMessage(command) + } + + if !c.throttlingDone { + logger.Debug(c.server, "Unthrottling connection (2)", len(throttleCh), cap(throttleCh)) + c.throttlingDone = true + <-throttleCh + } + + <-ctx.Done() + c.handler.Shutdown() + return nil +} -- cgit v1.2.3 From abeac87aec44249bf67f1b0eca471a31086265ca Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sat, 18 Sep 2021 19:27:50 +0300 Subject: fix auto reconnect --- internal/clients/connectors/serverconnection.go | 38 ++++------- internal/clients/connectors/serverless.go | 90 +++++++++++++++++++++++++ 2 files changed, 103 insertions(+), 25 deletions(-) create mode 100644 internal/clients/connectors/serverless.go (limited to 'internal/clients/connectors') diff --git a/internal/clients/connectors/serverconnection.go b/internal/clients/connectors/serverconnection.go index fab2f87..0904ba1 100644 --- a/internal/clients/connectors/serverconnection.go +++ b/internal/clients/connectors/serverconnection.go @@ -16,31 +16,21 @@ import ( "golang.org/x/crypto/ssh" ) -// ServerConnection represents a client connection connection to a single server. +// ServerConnection represents a connection to a single remote dtail server via SSH protocol. type ServerConnection struct { - // The remote server's hostname connected to. - server string - // The remote server's port connected to. - port int - // The SSH client configuration used. - config *ssh.ClientConfig - // The SSH client handler to use. - handler handlers.Handler - // DTail commands sent from client to server. When client loses - // connection to the server it re-connects automatically and sends the - // same commands again. - commands []string - // Is it a persistent connection or a one-off? - isOneOff bool - // To deal with SSH server host keys + server string + port int + config *ssh.ClientConfig + handler handlers.Handler + commands []string + isOneOff bool hostKeyCallback client.HostKeyCallback - // To determine if connection throttling has finished or not - throttlingDone bool + throttlingDone bool } -// NewServerConnection returns a new connection. +// NewServerConnection returns a new DTail SSH server connection. func NewServerConnection(server string, userName string, authMethods []ssh.AuthMethod, hostKeyCallback client.HostKeyCallback, handler handlers.Handler, commands []string) *ServerConnection { - logger.Debug(server, "Creating new connection") + logger.Debug(server, "Creating new connection", server, handler, commands) c := ServerConnection{ hostKeyCallback: hostKeyCallback, @@ -77,12 +67,10 @@ func NewOneOffServerConnection(server string, userName string, authMethods []ssh return &c } -// Server hostname func (c *ServerConnection) Server() string { return c.server } -// Handler for the connection func (c *ServerConnection) Handler() handlers.Handler { return c.handler } @@ -103,7 +91,6 @@ func (c *ServerConnection) initServerPort() { } } -// Start the server connection. Build up SSH session and send some DTail commands. func (c *ServerConnection) Start(ctx context.Context, cancel context.CancelFunc, throttleCh, statsCh chan struct{}) { // Throttle how many connections can be established concurrently (based on ch length) logger.Debug(c.server, "Throttling connection", len(throttleCh), cap(throttleCh)) @@ -161,7 +148,7 @@ func (c *ServerConnection) dial(ctx context.Context, cancel context.CancelFunc, // Create the SSH session. Close the session in case of an error. func (c *ServerConnection) session(ctx context.Context, cancel context.CancelFunc, client *ssh.Client, throttleCh chan struct{}) error { - logger.Debug(c.server, "session") + logger.Debug(c.server, "Creating SSH session") session, err := client.NewSession() if err != nil { @@ -173,7 +160,7 @@ func (c *ServerConnection) session(ctx context.Context, cancel context.CancelFun } func (c *ServerConnection) handle(ctx context.Context, cancel context.CancelFunc, session *ssh.Session, throttleCh chan struct{}) error { - logger.Debug(c.server, "handle") + logger.Debug(c.server, "Creating handler for SSH session") stdinPipe, err := session.StdinPipe() if err != nil { @@ -221,5 +208,6 @@ func (c *ServerConnection) handle(ctx context.Context, cancel context.CancelFunc <-ctx.Done() c.handler.Shutdown() + return nil } diff --git a/internal/clients/connectors/serverless.go b/internal/clients/connectors/serverless.go new file mode 100644 index 0000000..0500645 --- /dev/null +++ b/internal/clients/connectors/serverless.go @@ -0,0 +1,90 @@ +package connectors + +import ( + "context" + "io" + + "github.com/mimecast/dtail/internal/clients/handlers" + "github.com/mimecast/dtail/internal/config" + "github.com/mimecast/dtail/internal/io/logger" + serverHandlers "github.com/mimecast/dtail/internal/server/handlers" + user "github.com/mimecast/dtail/internal/user/server" +) + +// Serverless creates a server object directly without TCP. +type Serverless struct { + handler handlers.Handler + commands []string + userName string +} + +// NewServerConnection returns a new connection. +func NewServerless(userName string, handler handlers.Handler, commands []string) *Serverless { + s := Serverless{ + userName: userName, + handler: handler, + commands: commands, + } + + logger.Debug("Creating new serverless connector", handler, commands) + return &s +} + +func (s *Serverless) Server() string { + return "local(serverless)" +} + +func (s *Serverless) Handler() handlers.Handler { + return s.handler +} + +func (s *Serverless) Start(ctx context.Context, cancel context.CancelFunc, throttleCh, statsCh chan struct{}) { + go func() { + defer cancel() + + if err := s.handle(ctx, cancel); err != nil { + logger.Warn(err) + } + }() + + <-ctx.Done() +} + +func (s *Serverless) handle(ctx context.Context, cancel context.CancelFunc) error { + logger.Debug("Creating server handler for a serverless session") + + serverHandler := serverHandlers.NewServerHandler( + user.New(s.userName, s.Server()), + make(chan struct{}, config.Server.MaxConcurrentCats), + make(chan struct{}, config.Server.MaxConcurrentTails), + ) + + go func() { + io.Copy(serverHandler, s.handler) + cancel() + }() + + go func() { + io.Copy(s.handler, serverHandler) + cancel() + }() + + go func() { + select { + case <-s.handler.Done(): + case <-ctx.Done(): + } + cancel() + }() + + // Send all commands to client. + for _, command := range s.commands { + logger.Debug(command) + s.handler.SendMessage(command) + } + + <-ctx.Done() + s.handler.Shutdown() + + return nil +} -- cgit v1.2.3 From fe3e68afd99d8ea246be52893730f987e138ec24 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sun, 19 Sep 2021 13:22:59 +0300 Subject: move args to config package logger package rewrite as dlog --- internal/clients/connectors/serverconnection.go | 34 ++++++++++++------------- internal/clients/connectors/serverless.go | 21 +++++++++------ 2 files changed, 30 insertions(+), 25 deletions(-) (limited to 'internal/clients/connectors') diff --git a/internal/clients/connectors/serverconnection.go b/internal/clients/connectors/serverconnection.go index 0904ba1..5bc63ee 100644 --- a/internal/clients/connectors/serverconnection.go +++ b/internal/clients/connectors/serverconnection.go @@ -10,7 +10,7 @@ import ( "github.com/mimecast/dtail/internal/clients/handlers" "github.com/mimecast/dtail/internal/config" - "github.com/mimecast/dtail/internal/io/logger" + "github.com/mimecast/dtail/internal/io/dlog" "github.com/mimecast/dtail/internal/ssh/client" "golang.org/x/crypto/ssh" @@ -30,7 +30,7 @@ type ServerConnection struct { // NewServerConnection returns a new DTail SSH server connection. func NewServerConnection(server string, userName string, authMethods []ssh.AuthMethod, hostKeyCallback client.HostKeyCallback, handler handlers.Handler, commands []string) *ServerConnection { - logger.Debug(server, "Creating new connection", server, handler, commands) + dlog.Client.Debug(server, "Creating new connection", server, handler, commands) c := ServerConnection{ hostKeyCallback: hostKeyCallback, @@ -81,10 +81,10 @@ func (c *ServerConnection) initServerPort() { parts := strings.Split(c.server, ":") if len(parts) == 2 { - logger.Debug("Parsing port from hostname", parts) + dlog.Client.Debug("Parsing port from hostname", parts) port, err := strconv.Atoi(parts[1]) if err != nil { - logger.FatalExit("Unable to parse client port", c.server, parts, err) + dlog.Client.FatalPanic("Unable to parse client port", c.server, parts, err) } c.server = parts[0] c.port = port @@ -93,21 +93,21 @@ func (c *ServerConnection) initServerPort() { func (c *ServerConnection) Start(ctx context.Context, cancel context.CancelFunc, throttleCh, statsCh chan struct{}) { // Throttle how many connections can be established concurrently (based on ch length) - logger.Debug(c.server, "Throttling connection", len(throttleCh), cap(throttleCh)) + dlog.Client.Debug(c.server, "Throttling connection", len(throttleCh), cap(throttleCh)) select { case throttleCh <- struct{}{}: case <-ctx.Done(): - logger.Debug(c.server, "Not establishing connection as context is done", len(throttleCh), cap(throttleCh)) + dlog.Client.Debug(c.server, "Not establishing connection as context is done", len(throttleCh), cap(throttleCh)) return } - logger.Debug(c.server, "Throttling says that the connection can be established", len(throttleCh), cap(throttleCh)) + dlog.Client.Debug(c.server, "Throttling says that the connection can be established", len(throttleCh), cap(throttleCh)) go func() { defer func() { if !c.throttlingDone { - logger.Debug(c.server, "Unthrottling connection (1)", len(throttleCh), cap(throttleCh)) + dlog.Client.Debug(c.server, "Unthrottling connection (1)", len(throttleCh), cap(throttleCh)) c.throttlingDone = true <-throttleCh } @@ -115,9 +115,9 @@ func (c *ServerConnection) Start(ctx context.Context, cancel context.CancelFunc, }() if err := c.dial(ctx, cancel, throttleCh, statsCh); err != nil { - logger.Warn(c.server, c.port, err) + dlog.Client.Warn(c.server, c.port, err) if c.hostKeyCallback.Untrusted(fmt.Sprintf("%s:%d", c.server, c.port)) { - logger.Debug(c.server, "Not trusting host") + dlog.Client.Debug(c.server, "Not trusting host") } } }() @@ -127,14 +127,14 @@ func (c *ServerConnection) Start(ctx context.Context, cancel context.CancelFunc, // Dail into a new SSH connection. Close connection in case of an error. func (c *ServerConnection) dial(ctx context.Context, cancel context.CancelFunc, throttleCh, statsCh chan struct{}) error { - logger.Debug(c.server, "Incrementing connection stats") + dlog.Client.Debug(c.server, "Incrementing connection stats") statsCh <- struct{}{} defer func() { - logger.Debug(c.server, "Decrementing connection stats") + dlog.Client.Debug(c.server, "Decrementing connection stats") <-statsCh }() - logger.Debug(c.server, "Dialing into the connection") + dlog.Client.Debug(c.server, "Dialing into the connection") address := fmt.Sprintf("%s:%d", c.server, c.port) client, err := ssh.Dial("tcp", address, c.config) @@ -148,7 +148,7 @@ func (c *ServerConnection) dial(ctx context.Context, cancel context.CancelFunc, // Create the SSH session. Close the session in case of an error. func (c *ServerConnection) session(ctx context.Context, cancel context.CancelFunc, client *ssh.Client, throttleCh chan struct{}) error { - logger.Debug(c.server, "Creating SSH session") + dlog.Client.Debug(c.server, "Creating SSH session") session, err := client.NewSession() if err != nil { @@ -160,7 +160,7 @@ func (c *ServerConnection) session(ctx context.Context, cancel context.CancelFun } func (c *ServerConnection) handle(ctx context.Context, cancel context.CancelFunc, session *ssh.Session, throttleCh chan struct{}) error { - logger.Debug(c.server, "Creating handler for SSH session") + dlog.Client.Debug(c.server, "Creating handler for SSH session") stdinPipe, err := session.StdinPipe() if err != nil { @@ -196,12 +196,12 @@ func (c *ServerConnection) handle(ctx context.Context, cancel context.CancelFunc // Send all commands to client. for _, command := range c.commands { - logger.Debug(command) + dlog.Client.Debug(command) c.handler.SendMessage(command) } if !c.throttlingDone { - logger.Debug(c.server, "Unthrottling connection (2)", len(throttleCh), cap(throttleCh)) + dlog.Client.Debug(c.server, "Unthrottling connection (2)", len(throttleCh), cap(throttleCh)) c.throttlingDone = true <-throttleCh } diff --git a/internal/clients/connectors/serverless.go b/internal/clients/connectors/serverless.go index 0500645..c7b5f62 100644 --- a/internal/clients/connectors/serverless.go +++ b/internal/clients/connectors/serverless.go @@ -6,7 +6,7 @@ import ( "github.com/mimecast/dtail/internal/clients/handlers" "github.com/mimecast/dtail/internal/config" - "github.com/mimecast/dtail/internal/io/logger" + "github.com/mimecast/dtail/internal/io/dlog" serverHandlers "github.com/mimecast/dtail/internal/server/handlers" user "github.com/mimecast/dtail/internal/user/server" ) @@ -26,7 +26,7 @@ func NewServerless(userName string, handler handlers.Handler, commands []string) commands: commands, } - logger.Debug("Creating new serverless connector", handler, commands) + dlog.Client.Debug("Creating new serverless connector", handler, commands) return &s } @@ -43,7 +43,7 @@ func (s *Serverless) Start(ctx context.Context, cancel context.CancelFunc, throt defer cancel() if err := s.handle(ctx, cancel); err != nil { - logger.Warn(err) + dlog.Client.Warn(err) } }() @@ -51,7 +51,7 @@ func (s *Serverless) Start(ctx context.Context, cancel context.CancelFunc, throt } func (s *Serverless) handle(ctx context.Context, cancel context.CancelFunc) error { - logger.Debug("Creating server handler for a serverless session") + dlog.Client.Debug("Creating server handler for a serverless session") serverHandler := serverHandlers.NewServerHandler( user.New(s.userName, s.Server()), @@ -59,14 +59,19 @@ func (s *Serverless) handle(ctx context.Context, cancel context.CancelFunc) erro make(chan struct{}, config.Server.MaxConcurrentTails), ) + terminate := func() { + serverHandler.Shutdown() + cancel() + } + go func() { io.Copy(serverHandler, s.handler) - cancel() + terminate() }() go func() { io.Copy(s.handler, serverHandler) - cancel() + terminate() }() go func() { @@ -74,12 +79,12 @@ func (s *Serverless) handle(ctx context.Context, cancel context.CancelFunc) erro case <-s.handler.Done(): case <-ctx.Done(): } - cancel() + terminate() }() // Send all commands to client. for _, command := range s.commands { - logger.Debug(command) + dlog.Client.Debug(command) s.handler.SendMessage(command) } -- cgit v1.2.3 From fcaa94c7453efa0d74e330128c0f5c2cde8f11b3 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sun, 26 Sep 2021 16:42:47 +0300 Subject: refactor config reader - also looks in additional search paths for config file unless NONE is specified --- internal/clients/connectors/serverless.go | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) (limited to 'internal/clients/connectors') diff --git a/internal/clients/connectors/serverless.go b/internal/clients/connectors/serverless.go index c7b5f62..7740aab 100644 --- a/internal/clients/connectors/serverless.go +++ b/internal/clients/connectors/serverless.go @@ -53,8 +53,13 @@ func (s *Serverless) Start(ctx context.Context, cancel context.CancelFunc, throt func (s *Serverless) handle(ctx context.Context, cancel context.CancelFunc) error { dlog.Client.Debug("Creating server handler for a serverless session") + user, err := user.New(s.userName, s.Server()) + if err != nil { + return err + } + serverHandler := serverHandlers.NewServerHandler( - user.New(s.userName, s.Server()), + user, make(chan struct{}, config.Server.MaxConcurrentCats), make(chan struct{}, config.Server.MaxConcurrentTails), ) -- cgit v1.2.3 From f70622f307629a2542ea5eb128dea8c1043d3a40 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Tue, 5 Oct 2021 10:00:38 +0300 Subject: more on this --- internal/clients/connectors/serverconnection.go | 19 ------------------- internal/clients/connectors/serverless.go | 22 +++++++++++++--------- 2 files changed, 13 insertions(+), 28 deletions(-) (limited to 'internal/clients/connectors') diff --git a/internal/clients/connectors/serverconnection.go b/internal/clients/connectors/serverconnection.go index 5bc63ee..1666a79 100644 --- a/internal/clients/connectors/serverconnection.go +++ b/internal/clients/connectors/serverconnection.go @@ -23,7 +23,6 @@ type ServerConnection struct { config *ssh.ClientConfig handler handlers.Handler commands []string - isOneOff bool hostKeyCallback client.HostKeyCallback throttlingDone bool } @@ -49,24 +48,6 @@ func NewServerConnection(server string, userName string, authMethods []ssh.AuthM return &c } -// NewOneOffServerConnection creates new one-off connection (only for sending a series of commands and then quit). -func NewOneOffServerConnection(server string, userName string, authMethods []ssh.AuthMethod, handler handlers.Handler, commands []string) *ServerConnection { - c := ServerConnection{ - server: server, - handler: handler, - commands: commands, - config: &ssh.ClientConfig{ - User: userName, - Auth: authMethods, - HostKeyCallback: ssh.InsecureIgnoreHostKey(), - }, - isOneOff: true, - } - - c.initServerPort() - return &c -} - func (c *ServerConnection) Server() string { return c.server } diff --git a/internal/clients/connectors/serverless.go b/internal/clients/connectors/serverless.go index 7740aab..ae72c9b 100644 --- a/internal/clients/connectors/serverless.go +++ b/internal/clients/connectors/serverless.go @@ -20,14 +20,12 @@ type Serverless struct { // NewServerConnection returns a new connection. func NewServerless(userName string, handler handlers.Handler, commands []string) *Serverless { - s := Serverless{ + dlog.Client.Debug("Creating new serverless connector", handler, commands) + return &Serverless{ userName: userName, handler: handler, commands: commands, } - - dlog.Client.Debug("Creating new serverless connector", handler, commands) - return &s } func (s *Serverless) Server() string { @@ -58,11 +56,17 @@ func (s *Serverless) handle(ctx context.Context, cancel context.CancelFunc) erro return err } - serverHandler := serverHandlers.NewServerHandler( - user, - make(chan struct{}, config.Server.MaxConcurrentCats), - make(chan struct{}, config.Server.MaxConcurrentTails), - ) + var serverHandler serverHandlers.Handler + switch s.userName { + case config.ControlUser: + serverHandler = serverHandlers.NewControlHandler(user) + default: + serverHandler = serverHandlers.NewServerHandler( + user, + make(chan struct{}, config.Server.MaxConcurrentCats), + make(chan struct{}, config.Server.MaxConcurrentTails), + ) + } terminate := func() { serverHandler.Shutdown() -- cgit v1.2.3 From 9f395a03f25941d8ed98ec43035688daa1e8877f Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Tue, 5 Oct 2021 22:39:58 +0300 Subject: more on this --- internal/clients/connectors/serverless.go | 1 + 1 file changed, 1 insertion(+) (limited to 'internal/clients/connectors') diff --git a/internal/clients/connectors/serverless.go b/internal/clients/connectors/serverless.go index ae72c9b..2a1cec4 100644 --- a/internal/clients/connectors/serverless.go +++ b/internal/clients/connectors/serverless.go @@ -69,6 +69,7 @@ func (s *Serverless) handle(ctx context.Context, cancel context.CancelFunc) erro } terminate := func() { + dlog.Client.Debug("Terminating serverless connection") serverHandler.Shutdown() cancel() } -- cgit v1.2.3 From fab5dc3e70434ea0abc7a0976487a1973b662331 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Wed, 6 Oct 2021 09:50:41 +0300 Subject: enable faster shutdown - useful for dgrep/dmap and dcat commands --- internal/clients/connectors/serverless.go | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) (limited to 'internal/clients/connectors') diff --git a/internal/clients/connectors/serverless.go b/internal/clients/connectors/serverless.go index 2a1cec4..768a5ce 100644 --- a/internal/clients/connectors/serverless.go +++ b/internal/clients/connectors/serverless.go @@ -37,6 +37,7 @@ func (s *Serverless) Handler() handlers.Handler { } func (s *Serverless) Start(ctx context.Context, cancel context.CancelFunc, throttleCh, statsCh chan struct{}) { + dlog.Client.Debug("Starting serverless connector") go func() { defer cancel() @@ -44,7 +45,6 @@ func (s *Serverless) Start(ctx context.Context, cancel context.CancelFunc, throt dlog.Client.Warn(err) } }() - <-ctx.Done() } @@ -58,9 +58,11 @@ func (s *Serverless) handle(ctx context.Context, cancel context.CancelFunc) erro var serverHandler serverHandlers.Handler switch s.userName { - case config.ControlUser: - serverHandler = serverHandlers.NewControlHandler(user) + case config.HealthUser: + dlog.Client.Debug("Creating serverless health handler") + serverHandler = serverHandlers.NewHealthHandler(user) default: + dlog.Client.Debug("Creating serverless server handler") serverHandler = serverHandlers.NewServerHandler( user, make(chan struct{}, config.Server.MaxConcurrentCats), @@ -76,29 +78,34 @@ func (s *Serverless) handle(ctx context.Context, cancel context.CancelFunc) erro go func() { io.Copy(serverHandler, s.handler) + dlog.Client.Trace("io.Copy(serverHandler, s.handler) => done") terminate() }() go func() { io.Copy(s.handler, serverHandler) + dlog.Client.Trace("io.Copy(s.handler, serverHandler) => done") terminate() }() go func() { select { case <-s.handler.Done(): + dlog.Client.Trace("<-s.handler.Done()") case <-ctx.Done(): + dlog.Client.Trace("<-ctx.Done()") } terminate() }() // Send all commands to client. for _, command := range s.commands { - dlog.Client.Debug(command) + dlog.Client.Debug("Sending command to serverless server", command) s.handler.SendMessage(command) } <-ctx.Done() + dlog.Client.Trace("s.handler.Shutdown()") s.handler.Shutdown() return nil -- cgit v1.2.3 From 97747ea0f3178f7f5890512d483fdccaa82846b0 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sat, 9 Oct 2021 21:10:29 +0300 Subject: vetting and linting and some code restyling --- internal/clients/connectors/serverconnection.go | 48 +++++++++++++++---------- internal/clients/connectors/serverless.go | 16 +++++---- 2 files changed, 39 insertions(+), 25 deletions(-) (limited to 'internal/clients/connectors') diff --git a/internal/clients/connectors/serverconnection.go b/internal/clients/connectors/serverconnection.go index 1666a79..2d7b45a 100644 --- a/internal/clients/connectors/serverconnection.go +++ b/internal/clients/connectors/serverconnection.go @@ -16,7 +16,8 @@ import ( "golang.org/x/crypto/ssh" ) -// ServerConnection represents a connection to a single remote dtail server via SSH protocol. +// ServerConnection represents a connection to a single remote dtail server via +// SSH protocol. type ServerConnection struct { server string port int @@ -28,9 +29,11 @@ type ServerConnection struct { } // NewServerConnection returns a new DTail SSH server connection. -func NewServerConnection(server string, userName string, authMethods []ssh.AuthMethod, hostKeyCallback client.HostKeyCallback, handler handlers.Handler, commands []string) *ServerConnection { - dlog.Client.Debug(server, "Creating new connection", server, handler, commands) +func NewServerConnection(server string, userName string, + authMethods []ssh.AuthMethod, hostKeyCallback client.HostKeyCallback, + handler handlers.Handler, commands []string) *ServerConnection { + dlog.Client.Debug(server, "Creating new connection", server, handler, commands) c := ServerConnection{ hostKeyCallback: hostKeyCallback, server: server, @@ -48,10 +51,12 @@ func NewServerConnection(server string, userName string, authMethods []ssh.AuthM return &c } +// Server returns the server hostname connected to. func (c *ServerConnection) Server() string { return c.server } +// Handler returns the handler used for the connection. func (c *ServerConnection) Handler() handlers.Handler { return c.handler } @@ -72,23 +77,29 @@ func (c *ServerConnection) initServerPort() { } } -func (c *ServerConnection) Start(ctx context.Context, cancel context.CancelFunc, throttleCh, statsCh chan struct{}) { +// Start the connection to the server. +func (c *ServerConnection) Start(ctx context.Context, cancel context.CancelFunc, + throttleCh, statsCh chan struct{}) { + // Throttle how many connections can be established concurrently (based on ch length) dlog.Client.Debug(c.server, "Throttling connection", len(throttleCh), cap(throttleCh)) select { case throttleCh <- struct{}{}: case <-ctx.Done(): - dlog.Client.Debug(c.server, "Not establishing connection as context is done", len(throttleCh), cap(throttleCh)) + dlog.Client.Debug(c.server, "Not establishing connection as context is done", + len(throttleCh), cap(throttleCh)) return } - dlog.Client.Debug(c.server, "Throttling says that the connection can be established", len(throttleCh), cap(throttleCh)) + dlog.Client.Debug(c.server, "Throttling says that the connection can be established", + len(throttleCh), cap(throttleCh)) go func() { defer func() { if !c.throttlingDone { - dlog.Client.Debug(c.server, "Unthrottling connection (1)", len(throttleCh), cap(throttleCh)) + dlog.Client.Debug(c.server, "Unthrottling connection (1)", + len(throttleCh), cap(throttleCh)) c.throttlingDone = true <-throttleCh } @@ -107,7 +118,9 @@ func (c *ServerConnection) Start(ctx context.Context, cancel context.CancelFunc, } // Dail into a new SSH connection. Close connection in case of an error. -func (c *ServerConnection) dial(ctx context.Context, cancel context.CancelFunc, throttleCh, statsCh chan struct{}) error { +func (c *ServerConnection) dial(ctx context.Context, cancel context.CancelFunc, + throttleCh, statsCh chan struct{}) error { + dlog.Client.Debug(c.server, "Incrementing connection stats") statsCh <- struct{}{} defer func() { @@ -128,31 +141,30 @@ func (c *ServerConnection) dial(ctx context.Context, cancel context.CancelFunc, } // Create the SSH session. Close the session in case of an error. -func (c *ServerConnection) session(ctx context.Context, cancel context.CancelFunc, client *ssh.Client, throttleCh chan struct{}) error { - dlog.Client.Debug(c.server, "Creating SSH session") +func (c *ServerConnection) session(ctx context.Context, cancel context.CancelFunc, + client *ssh.Client, throttleCh chan struct{}) error { + dlog.Client.Debug(c.server, "Creating SSH session") session, err := client.NewSession() if err != nil { return err } defer session.Close() - return c.handle(ctx, cancel, session, throttleCh) } -func (c *ServerConnection) handle(ctx context.Context, cancel context.CancelFunc, session *ssh.Session, throttleCh chan struct{}) error { - dlog.Client.Debug(c.server, "Creating handler for SSH session") +func (c *ServerConnection) handle(ctx context.Context, cancel context.CancelFunc, + session *ssh.Session, throttleCh chan struct{}) error { + dlog.Client.Debug(c.server, "Creating handler for SSH session") stdinPipe, err := session.StdinPipe() if err != nil { return err } - stdoutPipe, err := session.StdoutPipe() if err != nil { return err } - if err := session.Shell(); err != nil { return err } @@ -161,12 +173,10 @@ func (c *ServerConnection) handle(ctx context.Context, cancel context.CancelFunc io.Copy(stdinPipe, c.handler) cancel() }() - go func() { io.Copy(c.handler, stdoutPipe) cancel() }() - go func() { select { case <-c.handler.Done(): @@ -182,13 +192,13 @@ func (c *ServerConnection) handle(ctx context.Context, cancel context.CancelFunc } if !c.throttlingDone { - dlog.Client.Debug(c.server, "Unthrottling connection (2)", len(throttleCh), cap(throttleCh)) + dlog.Client.Debug(c.server, "Unthrottling connection (2)", + len(throttleCh), cap(throttleCh)) c.throttlingDone = true <-throttleCh } <-ctx.Done() c.handler.Shutdown() - return nil } diff --git a/internal/clients/connectors/serverless.go b/internal/clients/connectors/serverless.go index 768a5ce..2ff490a 100644 --- a/internal/clients/connectors/serverless.go +++ b/internal/clients/connectors/serverless.go @@ -18,8 +18,10 @@ type Serverless struct { userName string } -// NewServerConnection returns a new connection. -func NewServerless(userName string, handler handlers.Handler, commands []string) *Serverless { +// NewServerless starts a new serverless session. +func NewServerless(userName string, handler handlers.Handler, + commands []string) *Serverless { + dlog.Client.Debug("Creating new serverless connector", handler, commands) return &Serverless{ userName: userName, @@ -28,15 +30,20 @@ func NewServerless(userName string, handler handlers.Handler, commands []string) } } +// Server returns serverless server indicator. func (s *Serverless) Server() string { return "local(serverless)" } +// Handler returns the handler used for the serverless connection. func (s *Serverless) Handler() handlers.Handler { return s.handler } -func (s *Serverless) Start(ctx context.Context, cancel context.CancelFunc, throttleCh, statsCh chan struct{}) { +// Start the serverless connection. +func (s *Serverless) Start(ctx context.Context, cancel context.CancelFunc, + throttleCh, statsCh chan struct{}) { + dlog.Client.Debug("Starting serverless connector") go func() { defer cancel() @@ -81,13 +88,11 @@ func (s *Serverless) handle(ctx context.Context, cancel context.CancelFunc) erro dlog.Client.Trace("io.Copy(serverHandler, s.handler) => done") terminate() }() - go func() { io.Copy(s.handler, serverHandler) dlog.Client.Trace("io.Copy(s.handler, serverHandler) => done") terminate() }() - go func() { select { case <-s.handler.Done(): @@ -107,6 +112,5 @@ func (s *Serverless) handle(ctx context.Context, cancel context.CancelFunc) erro <-ctx.Done() dlog.Client.Trace("s.handler.Shutdown()") s.handler.Shutdown() - return nil } -- cgit v1.2.3 From 71f89dc7ec7cf993d1eca98771212afe6310e9c8 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sun, 10 Oct 2021 19:42:48 +0300 Subject: refactor --- internal/clients/connectors/serverconnection.go | 12 ++++-------- 1 file changed, 4 insertions(+), 8 deletions(-) (limited to 'internal/clients/connectors') diff --git a/internal/clients/connectors/serverconnection.go b/internal/clients/connectors/serverconnection.go index 2d7b45a..2737ede 100644 --- a/internal/clients/connectors/serverconnection.go +++ b/internal/clients/connectors/serverconnection.go @@ -37,6 +37,7 @@ func NewServerConnection(server string, userName string, c := ServerConnection{ hostKeyCallback: hostKeyCallback, server: server, + port: config.Common.SSHPort, handler: handler, commands: commands, config: &ssh.ClientConfig{ @@ -47,25 +48,20 @@ func NewServerConnection(server string, userName string, }, } + // TODO: After reconnecting the port is wrong! Due to string slicing? c.initServerPort() return &c } // Server returns the server hostname connected to. -func (c *ServerConnection) Server() string { - return c.server -} +func (c *ServerConnection) Server() string { return c.server } // Handler returns the handler used for the connection. -func (c *ServerConnection) Handler() handlers.Handler { - return c.handler -} +func (c *ServerConnection) Handler() handlers.Handler { return c.handler } // Attempt to parse the server port address from the provided server FQDN. func (c *ServerConnection) initServerPort() { - c.port = config.Common.SSHPort parts := strings.Split(c.server, ":") - if len(parts) == 2 { dlog.Client.Debug("Parsing port from hostname", parts) port, err := strconv.Atoi(parts[1]) -- cgit v1.2.3 From c0f4ebc9b3773c37e22c686024c8cc7ce4e71f9f Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Mon, 11 Oct 2021 11:51:30 +0300 Subject: add dtail integration test --- internal/clients/connectors/serverconnection.go | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 deletions(-) (limited to 'internal/clients/connectors') diff --git a/internal/clients/connectors/serverconnection.go b/internal/clients/connectors/serverconnection.go index 2737ede..1df4d73 100644 --- a/internal/clients/connectors/serverconnection.go +++ b/internal/clients/connectors/serverconnection.go @@ -19,7 +19,11 @@ import ( // ServerConnection represents a connection to a single remote dtail server via // SSH protocol. type ServerConnection struct { - server string + // The full server string as received from the server discovery (can be with port number) + server string + // Only the hostname or FQDN (without the port number) + hostname string + // Only the port number. port int config *ssh.ClientConfig handler handlers.Handler @@ -37,7 +41,6 @@ func NewServerConnection(server string, userName string, c := ServerConnection{ hostKeyCallback: hostKeyCallback, server: server, - port: config.Common.SSHPort, handler: handler, commands: commands, config: &ssh.ClientConfig{ @@ -48,7 +51,6 @@ func NewServerConnection(server string, userName string, }, } - // TODO: After reconnecting the port is wrong! Due to string slicing? c.initServerPort() return &c } @@ -61,6 +63,7 @@ func (c *ServerConnection) Handler() handlers.Handler { return c.handler } // Attempt to parse the server port address from the provided server FQDN. func (c *ServerConnection) initServerPort() { + c.port = config.Common.SSHPort parts := strings.Split(c.server, ":") if len(parts) == 2 { dlog.Client.Debug("Parsing port from hostname", parts) @@ -68,7 +71,7 @@ func (c *ServerConnection) initServerPort() { if err != nil { dlog.Client.FatalPanic("Unable to parse client port", c.server, parts, err) } - c.server = parts[0] + c.hostname = parts[0] c.port = port } } @@ -103,8 +106,8 @@ func (c *ServerConnection) Start(ctx context.Context, cancel context.CancelFunc, }() if err := c.dial(ctx, cancel, throttleCh, statsCh); err != nil { - dlog.Client.Warn(c.server, c.port, err) - if c.hostKeyCallback.Untrusted(fmt.Sprintf("%s:%d", c.server, c.port)) { + dlog.Client.Warn(c.server, err) + if c.hostKeyCallback.Untrusted(c.server) { dlog.Client.Debug(c.server, "Not trusting host") } } @@ -125,7 +128,7 @@ func (c *ServerConnection) dial(ctx context.Context, cancel context.CancelFunc, }() dlog.Client.Debug(c.server, "Dialing into the connection") - address := fmt.Sprintf("%s:%d", c.server, c.port) + address := fmt.Sprintf("%s:%d", c.hostname, c.port) client, err := ssh.Dial("tcp", address, c.config) if err != nil { -- cgit v1.2.3 From 3c16894e997bdb235f1c0ce1291cc85a7b56118c Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Tue, 19 Oct 2021 20:33:59 +0300 Subject: Bugfix: Use correct hostname when no port specified --- internal/clients/connectors/serverconnection.go | 23 +++++++++++++---------- 1 file changed, 13 insertions(+), 10 deletions(-) (limited to 'internal/clients/connectors') diff --git a/internal/clients/connectors/serverconnection.go b/internal/clients/connectors/serverconnection.go index 1df4d73..aeb2a41 100644 --- a/internal/clients/connectors/serverconnection.go +++ b/internal/clients/connectors/serverconnection.go @@ -63,17 +63,20 @@ func (c *ServerConnection) Handler() handlers.Handler { return c.handler } // Attempt to parse the server port address from the provided server FQDN. func (c *ServerConnection) initServerPort() { - c.port = config.Common.SSHPort parts := strings.Split(c.server, ":") - if len(parts) == 2 { - dlog.Client.Debug("Parsing port from hostname", parts) - port, err := strconv.Atoi(parts[1]) - if err != nil { - dlog.Client.FatalPanic("Unable to parse client port", c.server, parts, err) - } - c.hostname = parts[0] - c.port = port + if len(parts) == 1 { + c.hostname = c.server + c.port = config.Common.SSHPort + return + } + + dlog.Client.Debug("Parsing port from hostname", parts) + port, err := strconv.Atoi(parts[1]) + if err != nil { + dlog.Client.FatalPanic("Unable to parse client port", c.server, parts, err) } + c.hostname = parts[0] + c.port = port } // Start the connection to the server. @@ -127,8 +130,8 @@ func (c *ServerConnection) dial(ctx context.Context, cancel context.CancelFunc, <-statsCh }() - dlog.Client.Debug(c.server, "Dialing into the connection") address := fmt.Sprintf("%s:%d", c.hostname, c.port) + dlog.Client.Debug(c.server, "Dialing into the connection", address) client, err := ssh.Dial("tcp", address, c.config) if err != nil { -- cgit v1.2.3