[v11] Prevent "session.start" from being overwritten by "session.exec" (

#19497)

* Prevent "session.start" from being overwritten by "session.exec"

The `session.exec` event was not being passed through the session
recorder, which resulted in said event having an event index of 0.
This caused the original `session.start` event which also has an
`eid` of 0 to be overwritten by the `session.exec` event.

By emitting the `session.exec` event via the same mechanism as the
`session.start` event it gets a proper event index and no longer
overwrites the `session.start`.

Closes #13622

lib/srv/ctx.go

@@ -244,6 +244,10 @@ type IdentityContext struct {
 // and other resources. SessionContext also holds a ServerContext which can be
 // used to access resources on the underlying server. SessionContext can also
 // be used to attach resources that should be closed once the session closes.
+//
+// Any events that need to be recorded should be emitted via session and not
+// ServerContext directly. Failure to use the session emitted will result in
+// incorrect event indexes that may ultimately cause events to be overwritten.
 type ServerContext struct {
 	// ConnectionContext is the parent context which manages connection-level
 	// resources.

lib/srv/exec.go

@@ -364,6 +364,11 @@ func (e *remoteExec) PID() int {
 	return 0
 }
 
+// emitExecAuditEvent emits either an SCP or exec event based on the
+// command run.
+//
+// Note: to ensure that the event is recorded ctx.session must be used
+// instead of ctx.srv.
 func emitExecAuditEvent(ctx *ServerContext, cmd string, execErr error) {
 	// Create common fields for event.
 	serverMeta := apievents.ServerMetadata{
@@ -436,13 +441,14 @@ func emitExecAuditEvent(ctx *ServerContext, cmd string, execErr error) {
 				scpEvent.Code = events.SCPDownloadCode
 			}
 		}
-		if err := ctx.srv.EmitAuditEvent(ctx.srv.Context(), scpEvent); err != nil {
+		if err := ctx.session.emitAuditEvent(ctx.srv.Context(), scpEvent); err != nil {
 			log.WithError(err).Warn("Failed to emit scp event.")
 		}
 	} else {
 		execEvent := &apievents.Exec{
 			Metadata: apievents.Metadata{
-				Type: events.ExecEvent,
+				Type:        events.ExecEvent,
+				ClusterName: ctx.ClusterName,
 			},
 			ServerMetadata:     serverMeta,
 			SessionMetadata:    sessionMeta,
@@ -455,7 +461,7 @@ func emitExecAuditEvent(ctx *ServerContext, cmd string, execErr error) {
 		} else {
 			execEvent.Code = events.ExecCode
 		}
-		if err := ctx.srv.EmitAuditEvent(ctx.srv.Context(), execEvent); err != nil {
+		if err := ctx.session.emitAuditEvent(ctx.srv.Context(), execEvent); err != nil {
 			log.WithError(err).Warn("Failed to emit exec event.")
 		}
 	}

lib/srv/exec_test.go

@@ -28,6 +28,7 @@ import (
 
 	"github.com/gravitational/teleport"
 	apievents "github.com/gravitational/teleport/api/types/events"
+	"github.com/gravitational/teleport/lib/events"
 	"github.com/gravitational/teleport/lib/sshutils"
 )
 
@@ -39,6 +40,9 @@ func TestEmitExecAuditEvent(t *testing.T) {
 	srv := newMockServer(t)
 	scx := newExecServerContext(t, srv)
 
+	rec, ok := scx.session.recorder.(*mockRecorder)
+	require.True(t, ok)
+
 	expectedUsr, err := user.Current()
 	require.NoError(t, err)
 	expectedHostname, err := os.Hostname()
@@ -86,7 +90,7 @@ func TestEmitExecAuditEvent(t *testing.T) {
 	}
 	for _, tt := range tests {
 		emitExecAuditEvent(scx, tt.inCommand, tt.inError)
-		execEvent := srv.MockEmitter.LastEvent().(*apievents.Exec)
+		execEvent := rec.emitter.LastEvent().(*apievents.Exec)
 		require.Equal(t, tt.outCommand, execEvent.Command)
 		require.Equal(t, tt.outCode, execEvent.ExitCode)
 		require.Equal(t, expectedMeta, execEvent.UserMetadata)
@@ -96,6 +100,7 @@ func TestEmitExecAuditEvent(t *testing.T) {
 		require.Equal(t, "xxx", execEvent.SessionID)
 		require.Equal(t, "10.0.0.5:4817", execEvent.RemoteAddr)
 		require.Equal(t, "127.0.0.1:3022", execEvent.LocalAddr)
+		require.NotZero(t, events.EventID)
 	}
 }
 
@@ -117,8 +122,11 @@ func newExecServerContext(t *testing.T, srv Server) *ServerContext {
 	require.NoError(t, err)
 	term.SetTermType("xterm")
 
-	scx.session = &session{id: "xxx"}
-	scx.session.term = term
+	scx.session = &session{
+		id:       "xxx",
+		term:     term,
+		recorder: &mockRecorder{done: false},
+	}
 	err = scx.SetSSHRequest(&ssh.Request{Type: sshutils.ExecRequest})
 	require.NoError(t, err)
 

lib/srv/sess_test.go

@@ -596,7 +596,8 @@ func testOpenSession(t *testing.T, reg *SessionRegistry, roleSet services.RoleSe
 
 type mockRecorder struct {
 	events.StreamWriter
-	done bool
+	emitter eventstest.MockEmitter
+	done    bool
 }
 
 func (m *mockRecorder) Done() <-chan struct{} {
@@ -608,6 +609,10 @@ func (m *mockRecorder) Done() <-chan struct{} {
 	return ch
 }
 
+func (m *mockRecorder) EmitAuditEvent(ctx context.Context, event apievents.AuditEvent) error {
+	return m.emitter.EmitAuditEvent(ctx, event)
+}
+
 type trackerService struct {
 	created     atomic.Int32
 	createError error