refactor(ssh,pkg): convert recorded session to events

Author: henrybarreto

api/routes/session.go

@@ -133,7 +133,7 @@ func (h *Handler) EventSession(c gateway.Context) error {
 
 	return h.service.EventSession(c.Ctx(), models.UID(req.UID), &models.SessionEvent{
 		Session:   req.UID,
-		Type:      req.Type,
+		Type:      models.SessionEventType(req.Type),
 		Timestamp: req.Timestamp,
 		Data:      req.Data,
 		Seat:      req.Seat,

pkg/models/session.go

@@ -52,25 +52,48 @@ type Status struct {
 }
 
 type SessionRecorded struct {
-	UID       string `json:"uid"`
-	Seat      int    `json:"seat"`
-	Namespace string `json:"namespace" bson:"namespace"`
-	Message   string `json:"message" bson:"message"`
-	Width     int    `json:"width" bson:"width,omitempty"`
-	Height    int    `json:"height" bson:"height,omitempty"`
+	UID    string `json:"uid"`
+	Output string `json:"output" bson:"output"`
 }
 
 type SessionUpdate struct {
 	Authenticated *bool   `json:"authenticated"`
 	Type          *string `json:"type"`
 }
 
+type SessionEventType string
+
+const (
+	// ShellHub custom requests.
+	SessionEventTypePtyOutput SessionEventType = "pty-output"
+
+	// Terminal (PTY) request types
+	SessionEventTypePtyRequest   SessionEventType = "pty-req"
+	SessionEventTypeWindowChange SessionEventType = "window-change"
+	SessionEventTypeExitCode     SessionEventType = "exit-code"
+
+	// Process-related requests
+	SessionEventTypeExitStatus SessionEventType = "exit-status"
+	SessionEventTypeExitSignal SessionEventType = "exit-signal"
+
+	// Environment and Shell requests
+	SessionEventTypeEnv       SessionEventType = "env"
+	SessionEventTypeShell     SessionEventType = "shell"
+	SessionEventTypeExec      SessionEventType = "exec"
+	SessionEventTypeSubsystem SessionEventType = "subsystem"
+
+	// Signal and forwarding requests
+	SessionEventTypeSignal       SessionEventType = "signal"
+	SessionEventTypeTcpipForward SessionEventType = "tcpip-forward"
+	SessionEventTypeAuthAgentReq SessionEventType = "auth-agent-req"
+)
+
 // SessionEvent represents a session event.
 type SessionEvent struct {
 	// Session is the session UID where the event occurred.
 	Session string `json:"session" bson:"session,omitempty"`
 	// Type of the session. Normally, it is the SSH request name.
-	Type string `json:"type" bson:"type"`
+	Type SessionEventType `json:"type" bson:"type"`
 	// Timestamp contains the time when the event was logged.
 	Timestamp time.Time `json:"timestamp" bson:"timestamp"`
 	// Data is a generic structure containing data of the event, normally the unmarshaling data of the request.

ssh/server/channels/session.go

@@ -137,7 +137,7 @@ func DefaultSessionHandler() gliderssh.ChannelHandler {
 
 		defer agent.Close()
 
-		go pipe(ctx, sess, client.Channel, agent.Channel, seat)
+		go pipe(sess, client.Channel, agent.Channel, seat)
 
 		// TODO: Add middleware to block certain types of requests.
 		for {

ssh/server/channels/utils.go

@@ -5,7 +5,6 @@ import (
 	"sync"
 
 	"github.com/Masterminds/semver"
-	gliderssh "github.com/gliderlabs/ssh"
 	"github.com/shellhub-io/shellhub/pkg/envs"
 	"github.com/shellhub-io/shellhub/pkg/models"
 	"github.com/shellhub-io/shellhub/ssh/session"
@@ -14,88 +13,48 @@ import (
 )
 
 type Recorder struct {
-	queue   chan string
+	// channels is the source of data read, recorded and redirected to client.
 	channel gossh.Channel
+	// session is the session between Agent and Client.
+	session *session.Session
+	// seat is the current identifier of session's.
+	seat int
 }
 
-func NewRecorder(channel gossh.Channel, sess *session.Session, camera *session.Camera, seat int) (io.WriteCloser, error) {
-	// NOTE: The queue's size is a random number.
-	queue := make(chan string, 100)
-
-	go func() {
-		for {
-			msg, ok := <-queue
-			if !ok {
-				log.WithFields(log.Fields{"session": sess.UID, "sshid": sess.SSHID}).
-					Warning("recorder queue is closed")
-
-				return
-			}
-
-			if err := camera.WriteFrame(&models.SessionRecorded{ //nolint:errcheck
-				UID:       sess.UID,
-				Seat:      seat,
-				Namespace: sess.Lookup["domain"],
-				Message:   msg,
-				Width:     int(sess.Pty.Columns),
-				Height:    int(sess.Pty.Rows),
-			}); err != nil {
-				log.WithError(err).
-					WithFields(log.Fields{"session": sess.UID, "sshid": sess.SSHID}).
-					Warning("failed to send the session frame to record")
-
-					// NOTE: When a frame isn't sent correctly, we stop the writing loop, only reading from the queue,
-					// and discarding the messages to avoid stuck the go routine.
-				break
-			}
-		}
-
-		for {
-			// NOTE: Reads the queue and discards the data to avoid stuck the go routine.
-			if _, ok := <-queue; !ok {
-				log.WithFields(log.Fields{"session": sess.UID, "sshid": sess.SSHID}).
-					Warning("recorder queue is closed")
-
-				return
-			}
-		}
-	}()
-
+func NewRecorder(channel gossh.Channel, session *session.Session, seat int) (io.WriteCloser, error) {
 	return &Recorder{
-		queue:   queue,
 		channel: channel,
+		session: session,
+		seat:    seat,
 	}, nil
 }
 
-// record enqueues a session frame to be recorded. If the queue is closed, nothing is done.
-func (c *Recorder) record(msg string) {
-	select {
-	case c.queue <- msg:
-	default:
-		log.Trace("the message couldn't sent to the record queue")
-	}
-}
+// PtyOutputEventType is the event's type for an output.
+const PtyOutputEventType = "pty-output"
 
-func (c *Recorder) Write(data []byte) (int, error) {
-	read, err := c.channel.Write(data)
+func (c *Recorder) Write(output []byte) (int, error) {
+	read, err := c.channel.Write(output)
 	if err != nil {
 		return read, err
 	}
 
-	c.record(string(data))
+	// NOTE: Writes the event into the event stream to be processed and send to target endpoint.
+	c.session.Event(PtyOutputEventType, &models.SessionRecorded{
+		UID:    c.session.UID,
+		Output: string(output),
+	}, c.seat)
 
 	return read, nil
 }
 
+// Close closes the internal channel.
 func (c *Recorder) Close() error {
-	close(c.queue)
-
 	return c.channel.CloseWrite()
 }
 
 // pipe function pipes data between client and agent, and vice versa, recording each frame when ShellHub instance are
 // Cloud or Enterprise.
-func pipe(ctx gliderssh.Context, sess *session.Session, client gossh.Channel, agent gossh.Channel, seat int) {
+func pipe(sess *session.Session, client gossh.Channel, agent gossh.Channel, seat int) {
 	defer log.
 		WithFields(log.Fields{"session": sess.UID, "sshid": sess.SSHID}).
 		Trace("data pipe between client and agent has done")
@@ -110,23 +69,10 @@ func pipe(ctx gliderssh.Context, sess *session.Session, client gossh.Channel, ag
 		defer wg.Done()
 
 		if envs.IsEnterprise() || envs.IsCloud() {
-			recordURL := ctx.Value("RECORD_URL").(string)
-			if recordURL == "" {
-				log.WithFields(log.Fields{"session": sess.UID, "sshid": sess.SSHID, "record_url": recordURL}).
-					Warning("failed to start session's record because the record URL is empty")
-
-				goto normal
-			}
-
-			camera, err := sess.Record(ctx, recordURL, seat)
-			if err != nil {
-				goto normal
-			}
-
-			recorder, err := NewRecorder(client, sess, camera, seat)
+			recorder, err := NewRecorder(client, sess, seat)
 			if err != nil {
 				log.WithError(err).
-					WithFields(log.Fields{"session": sess.UID, "sshid": sess.SSHID, "record_url": recordURL}).
+					WithFields(log.Fields{"session": sess.UID, "sshid": sess.SSHID}).
 					Warning("failed to connect to session record endpoint")
 
 				goto normal

ssh/session/session.go

@@ -156,6 +156,8 @@ type Session struct {
 
 	api    internalclient.Client
 	tunnel *httptunnel.Tunnel
+	// Events is a channel used to send and process Events of a session.
+	Events chan *models.SessionEvent
 
 	once *sync.Once
 
@@ -247,6 +249,7 @@ func NewSession(ctx gliderssh.Context, tunnel *httptunnel.Tunnel, cache cache.Ca
 		UID:    ctx.SessionID(),
 		api:    api,
 		tunnel: tunnel,
+		Events: make(chan *models.SessionEvent, 100),
 		Data: Data{
 			IPAddress: hos.Host,
 			Target:    target,
@@ -269,6 +272,8 @@ func NewSession(ctx gliderssh.Context, tunnel *httptunnel.Tunnel, cache cache.Ca
 
 	snap.save(session, StateCreated)
 
+	go session.NewEventStream(ctx)
+
 	return session, nil
 }
 
@@ -565,13 +570,13 @@ func (s *Session) NewSeat() (int, error) {
 
 // Events register an event to the session.
 func (s *Session) Event(t string, data any, seat int) {
-	go s.api.EventSession(s.UID, &models.SessionEvent{ //nolint:errcheck
+	s.Events <- &models.SessionEvent{
 		Session:   s.UID,
-		Type:      t,
+		Type:      models.SessionEventType(t),
 		Timestamp: clock.Now(),
 		Data:      data,
 		Seat:      seat,
-	})
+	}
 }
 
 func Event[D any](sess *Session, t string, data []byte, seat int) {
@@ -580,13 +585,33 @@ func Event[D any](sess *Session, t string, data []byte, seat int) {
 		return
 	}
 
-	go sess.api.EventSession(sess.UID, &models.SessionEvent{ //nolint:errcheck
+	sess.Events <- &models.SessionEvent{
 		Session:   sess.UID,
-		Type:      t,
+		Type:      models.SessionEventType(t),
 		Timestamp: clock.Now(),
-		Data:      d,
+		Data:      data,
 		Seat:      seat,
-	})
+	}
+}
+
+func (s *Session) NewEventStream(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
+			log.WithFields(log.Fields{
+				"session": s.UID,
+			}).Debug("event stream loop done")
+
+			return
+		case event := <-s.Events:
+			log.WithFields(log.Fields{
+				"session": s.UID,
+				"event":   event,
+			}).Trace("event received on event stream")
+
+			go s.api.EventSession(s.UID, event) //nolint:errcheck
+		}
+	}
 }
 
 func (s *Session) KeepAlive() error {