Enable and fix AuditOn.

integration/helpers/fixture.go

@@ -51,7 +51,8 @@ func NewFixture(t *testing.T) *Fixture {
 	require.NoError(t, err)
 
 	// Find AllocatePortsNum free listening ports to use.
-	fixture.Me, _ = user.Current()
+	fixture.Me, err = user.Current()
+	require.NoError(t, err)
 
 	// close & re-open stdin because 'go test' runs with os.stdin connected to /dev/null
 	stdin, err := os.Open("/dev/tty")

integration/integration_test.go

@@ -282,14 +282,6 @@ func testAuthLocalNodeControlStream(t *testing.T, suite *integrationTestSuite) {
 // testAuditOn creates a live session, records a bunch of data through it
 // and then reads it back and compares against simulated reality.
 func testAuditOn(t *testing.T, suite *integrationTestSuite) {
-	// TODO(jakule): Re-enable once the fix is ready.
-	t.Skip("This test is flaky, skip for now.")
-
-	ctx := context.Background()
-
-	tr := utils.NewTracer(utils.ThisFunction()).Start()
-	defer tr.Stop()
-
 	tests := []struct {
 		comment          string
 		inRecordLocation string
@@ -327,6 +319,11 @@ func testAuditOn(t *testing.T, suite *integrationTestSuite) {
 
 	for _, tt := range tests {
 		t.Run(tt.comment, func(t *testing.T) {
+			tr := utils.NewTracer(utils.ThisFunction()).Start()
+			t.Cleanup(func() {
+				tr.Stop()
+			})
+
 			makeConfig := func() (*testing.T, []string, []*helpers.InstanceSecrets, *service.Config) {
 				auditConfig, err := types.NewClusterAuditConfig(types.ClusterAuditConfigSpecV2{
 					AuditSessionsURI: tt.auditSessionsURI,
@@ -349,7 +346,10 @@ func testAuditOn(t *testing.T, suite *integrationTestSuite) {
 				return t, nil, nil, tconf
 			}
 			teleport := suite.NewTeleportWithConfig(makeConfig())
-			defer teleport.StopAll()
+			t.Cleanup(func() {
+				err := teleport.StopAll()
+				require.NoError(t, err)
+			})
 
 			// Start a node.
 			nodeConf := suite.defaultServiceConfig()
@@ -364,6 +364,8 @@ func testAuditOn(t *testing.T, suite *integrationTestSuite) {
 			site := teleport.GetSiteAPI(helpers.Site)
 			require.NotNil(t, site)
 
+			ctx := context.Background()
+
 			// wait 10 seconds for both nodes to show up, otherwise
 			// we'll have trouble connecting to the node below.
 			waitForNodes := func(site auth.ClientI, count int) error {
@@ -437,9 +439,8 @@ func testAuditOn(t *testing.T, suite *integrationTestSuite) {
 			// make sure it's us who joined! :)
 			require.Equal(t, suite.Me.Username, tracker.GetParticipants()[0].User)
 
-			// lets type "echo hi" followed by "enter" and then "exit" + "enter":
-
-			myTerm.Type("\aecho hi\n\r\aexit\n\r\a")
+			// let's type "echo hi" followed by "enter" and then "exit" + "enter":
+			myTerm.Type("echo hi\n\rexit\n\r")
 
 			// wait for session to end:
 			select {

lib/srv/reexec.go

@@ -304,13 +304,6 @@ func RunCommand() (errw io.Writer, code int, err error) {
 	if err != nil {
 		return errorWriter, teleport.RemoteCommandFailure, trace.Wrap(err)
 	}
-	defer func() {
-		for _, f := range cmd.ExtraFiles {
-			if err := f.Close(); err != nil {
-				log.WithError(err).Warn("Error closing extra file.")
-			}
-		}
-	}()
 
 	// Wait until the continue signal is received from Teleport signaling that
 	// the child process has been placed in a cgroup.

lib/srv/sess.go

@@ -1000,6 +1000,12 @@ func (s *session) startInteractive(ctx context.Context, ch ssh.Channel, scx *Ser
 			s.log.WithError(err).Error("Received error waiting for the interactive session to finish")
 		}
 
+		if result != nil {
+			if err := s.registry.broadcastResult(s.id, *result); err != nil {
+				s.log.Warningf("Failed to broadcast session result: %v", err)
+			}
+		}
+
 		// wait for copying from the pty to be complete or a timeout before
 		// broadcasting the result (which will close the pty) if it has not been
 		// closed already.
@@ -1013,14 +1019,10 @@ func (s *session) startInteractive(ctx context.Context, ch ssh.Channel, scx *Ser
 			emitExecAuditEvent(scx, execRequest.GetCommand(), err)
 		}
 
-		if result != nil {
-			if err := s.registry.broadcastResult(s.id, *result); err != nil {
-				s.log.Warningf("Failed to broadcast session result: %v", err)
-			}
-		}
-
 		s.emitSessionEndEvent()
-		s.Close()
+		if err := s.Close(); err != nil {
+			s.log.Warnf("Failed to close session: %v", err)
+		}
 	}()
 
 	return nil

lib/srv/term.go

@@ -168,10 +168,8 @@ func (t *terminal) AddParty(delta int) {
 }
 
 // Run will run the terminal.
-func (t *terminal) Run(ctx context.Context) error {
+func (t *terminal) Run(_ context.Context) error {
 	var err error
-	defer t.closeTTY()
-
 	// Create the command that will actually execute.
 	t.cmd, err = ConfigureCommand(t.ctx)
 	if err != nil {
@@ -224,7 +222,9 @@ func (t *terminal) Wait() (*ExecResult, error) {
 // Continue will resume execution of the process after it completes its
 // pre-processing routine (placed in a cgroup).
 func (t *terminal) Continue() {
-	t.ctx.contw.Close()
+	if err := t.ctx.contw.Close(); err != nil {
+		t.log.Warnf("failed to close server context")
+	}
 }
 
 // Kill will force kill the terminal.
@@ -269,10 +269,14 @@ func (t *terminal) Close() error {
 }
 
 func (t *terminal) closeTTY() error {
+	t.log.Debugf("Closing TTY")
+	defer t.log.Debugf("Closed TTY")
+
 	t.mu.Lock()
 	defer t.mu.Unlock()
 
 	if t.tty == nil {
+		t.log.Debugf("TTY already closed")
 		return nil
 	}
 
@@ -287,14 +291,21 @@ func (t *terminal) closeTTY() error {
 }
 
 func (t *terminal) closePTY() {
+	defer t.log.Debugf("Closed PTY")
+
 	t.mu.Lock()
 	defer t.mu.Unlock()
-	defer t.log.Debugf("Closed PTY")
 
 	// wait until all copying is over (all participants have left)
 	t.wg.Wait()
 
-	t.pty.Close()
+	if t.pty == nil {
+		return
+	}
+
+	if err := t.pty.Close(); err != nil {
+		t.log.Warnf("Failed to close PTY: %v", err)
+	}
 	t.pty = nil
 }
 
@@ -458,7 +469,7 @@ func (b *ptyBuffer) Write(p []byte) (n int, err error) {
 }
 
 func (t *remoteTerminal) Run(ctx context.Context) error {
-	// prepare the remote remote session by setting environment variables
+	// prepare the remote session by setting environment variables
 	t.prepareRemoteSession(ctx, t.session, t.ctx)
 
 	// combine stdout and stderr
@@ -560,6 +571,7 @@ func (t *remoteTerminal) PID() int {
 }
 
 func (t *remoteTerminal) Close() error {
+	t.wg.Wait()
 	// this closes the underlying stdin,stdout,stderr which is what ptyBuffer is
 	// hooked to directly
 	err := t.session.Close()