[tctl] introduce tctl recordings download (#62431) (#63726)

* [tctl] introduce `tctl recordings download`

This PR introduces a new commnd to download session recordings directly
from auth server using stream.

This command aims to solve problems where a user needs to download a
specific session recording but doesn't have access to the underlying
storage - i.e. teleport saas cloud.

* use os.Getpwd

* add docs

Author: tigrato

docs/pages/reference/architecture/session-recording.mdx

@@ -49,7 +49,7 @@ informally referred to as "chunks."
 
 ### Database sessions
 
-Teleport captures database queries and a stream of audit events 
+Teleport captures database queries and a stream of audit events
 related to the database being accessed.
 
 ## Session recording configurations
@@ -140,7 +140,7 @@ The Proxy Service then establishes its own connection to the destination server
 - For [OpenSSH servers](../../enroll-resources/server-access/openssh/openssh-agentless.mdx), the Proxy Service requests a certificate signed by the Teleport OpenSSH CA, which will be trusted by the OpenSSH server.
 
 With both sides of the connections decrypted, the Proxy Service acts as a pipe between the client
-and the destination server, performing RBAC checks, recording the session, and emitting audit events 
+and the destination server, performing RBAC checks, recording the session, and emitting audit events
 in the process as shown below:
 
 ![recording-proxy](../../../img/recording-proxy.svg)
@@ -256,6 +256,18 @@ This is sometimes surprising to Teleport users, because even though the recordin
 and audit log are stored in separate backends, both need to be operational in
 order to play recordings.
 
+## Download
+
+Session recordings can be downloaded from the cluster storage using the
+[`tctl recordings download`](../cli/tctl.mdx#tctl-recordings-download) command.
+
+```code
+$ tctl recordings download [--output-dir <output-dir>] <session_id>
+```
+
+All downloaded recordings are decrypted during the streaming process if session
+recording encryption is enabled.
+
 ## Upload completer
 
 Every Teleport process runs a service called the *upload completer* which periodically
@@ -338,3 +350,4 @@ defined when the key is first provisioned.
 - [SSH recording modes](../deployment/monitoring/audit.mdx)
 - [Session recording for desktops](../../enroll-resources/desktop-access/reference/sessions.mdx)
 - [Encrypted Session Recordings](../../enroll-resources/server-access/guides/encrypted-session-recordings/encrypted-session-recordings.mdx)
+- [Session recording summaries](../../identity-security/session-summaries.mdx)

docs/pages/reference/cli/tctl.mdx

@@ -1536,6 +1536,19 @@ Flags:
 |---|---|---|
 |`--format`|`yaml`|Output format: 'yaml', 'json' or 'text'|
 
+## tctl recordings download
+
+Download session recordings from the cluster's recording storage to a local file.
+If you omit an output directory, recordings are saved to the current working directory.
+
+If session recording encryption is enabled in the cluster, recordings are decrypted during download.
+
+### Example
+
+```code
+$ tctl recordings download [--output-dir <output-dir>] <session_id>
+```
+
 ## tctl recordings encryption complete-rotation
 
 Completes an in-progress encryption key rotation.

lib/client/api.go

@@ -1637,7 +1637,7 @@ func (tc *TeleportClient) GetTargetNode(ctx context.Context, clt authclient.Clie
 		Labels:              tc.Labels,
 	})
 	switch {
-	//TODO(tross): DELETE IN v20.0.0
+	// TODO(tross): DELETE IN v20.0.0
 	case trace.IsNotImplemented(err):
 		resources, err := client.GetAllUnifiedResources(ctx, clt, &proto.ListUnifiedResourcesRequest{
 			Kinds:               []string{types.KindNode},
@@ -2570,7 +2570,7 @@ func playSession(ctx context.Context, sessionID string, speed float64, streamer
 		}
 	}
 
-	if err := player.Err(); err != nil {
+	if err := player.Err(); err != nil && !errors.Is(err, io.EOF) {
 		return trace.Wrap(err)
 	}
 

tool/tctl/common/recordings_command.go

@@ -23,6 +23,7 @@ import (
 	"fmt"
 	"io"
 	"os"
+	"path/filepath"
 
 	"github.com/alecthomas/kingpin/v2"
 	"github.com/gravitational/trace"
@@ -34,7 +35,10 @@ import (
 	"github.com/gravitational/teleport/lib/auth/authclient"
 	"github.com/gravitational/teleport/lib/client"
 	"github.com/gravitational/teleport/lib/defaults"
+	"github.com/gravitational/teleport/lib/events"
+	"github.com/gravitational/teleport/lib/events/filesessions"
 	"github.com/gravitational/teleport/lib/service/servicecfg"
+	"github.com/gravitational/teleport/lib/session"
 	"github.com/gravitational/teleport/tool/common"
 	commonclient "github.com/gravitational/teleport/tool/tctl/common/client"
 	tctlcfg "github.com/gravitational/teleport/tool/tctl/common/config"
@@ -48,6 +52,8 @@ type RecordingsCommand struct {
 	format string
 	// recordingsList implements the "tctl recordings ls" subcommand.
 	recordingsList *kingpin.CmdClause
+	// recordingsDownload implements the "tctl recordings download" subcommand.
+	recordingsDownload *kingpin.CmdClause
 	// recordingsEncryption implements the "tctl recordings encryption" subcommand.
 	recordingsEncryption recordingsEncryptionCommand
 	// fromUTC is the start time to use for the range of recordings listed by the recorded session listing command
@@ -59,12 +65,17 @@ type RecordingsCommand struct {
 	// recordingsSince is a duration which sets the time into the past in which to list session recordings
 	recordingsSince string
 
+	// recordingsDownloadSessionID is the session ID to download recordings for
+	recordingsDownloadSessionID string
+	// recordingsDownloadOutputDir is the output directory to download session recordings to
+	recordingsDownloadOutputDir string
+
 	// stdout allows to switch standard output source for resource command. Used in tests.
 	stdout io.Writer
 }
 
 // Initialize allows RecordingsCommand to plug itself into the CLI parser
-func (c *RecordingsCommand) Initialize(app *kingpin.Application, _ *tctlcfg.GlobalCLIFlags, config *servicecfg.Config) {
+func (c *RecordingsCommand) Initialize(app *kingpin.Application, t *tctlcfg.GlobalCLIFlags, config *servicecfg.Config) {
 	if c.stdout == nil {
 		c.stdout = os.Stdout
 	}
@@ -79,6 +90,15 @@ func (c *RecordingsCommand) Initialize(app *kingpin.Application, _ *tctlcfg.Glob
 	c.recordingsList.Flag("last", "Duration into the past from which session recordings should be listed. Format 5h30m40s").StringVar(&c.recordingsSince)
 	c.recordingsEncryption.Initialize(recordings, c.stdout)
 
+	download := recordings.Command("download", "Download session recordings.")
+	download.Arg("session-id", "ID of the session to download recordings for.").Required().StringVar(&c.recordingsDownloadSessionID)
+	pwd, err := os.Getwd()
+	if err != nil {
+		pwd = "."
+	}
+	download.Flag("output-dir", "Directory to download session recordings to.").Short('o').Default(pwd).StringVar(&c.recordingsDownloadOutputDir)
+	c.recordingsDownload = download
+
 	if c.recordingsEncryption.stdout == nil {
 		c.recordingsEncryption.stdout = c.stdout
 	}
@@ -90,6 +110,8 @@ func (c *RecordingsCommand) TryRun(ctx context.Context, cmd string, clientFunc c
 	switch cmd {
 	case c.recordingsList.FullCommand():
 		commandFunc = c.ListRecordings
+	case c.recordingsDownload.FullCommand():
+		commandFunc = c.DownloadRecordings
 	default:
 		return c.recordingsEncryption.TryRun(ctx, cmd, clientFunc)
 	}
@@ -120,3 +142,77 @@ func (c *RecordingsCommand) ListRecordings(ctx context.Context, tc *authclient.C
 	}
 	return trace.Wrap(common.ShowSessions(recordings, c.format, c.stdout))
 }
+
+func (c *RecordingsCommand) DownloadRecordings(ctx context.Context, tc *authclient.Client) (err error) {
+	sessionID, err := session.ParseID(c.recordingsDownloadSessionID)
+	if err != nil {
+		return trace.BadParameter("invalid session id")
+	}
+
+	e, err := createFileWriter(ctx, *sessionID, c.recordingsDownloadOutputDir)
+	if err != nil {
+		return trace.Wrap(err, "creating file downloader")
+	}
+	defer func() {
+		completeErr := e.Complete(ctx)
+		if err == nil && completeErr == nil {
+			return
+		}
+		localRemErr := os.Remove(filepath.Join(c.recordingsDownloadOutputDir, string(*sessionID)+".tar"))
+		// ignore file not found errors
+		if os.IsNotExist(localRemErr) {
+			localRemErr = nil
+		}
+		err = trace.NewAggregate(err, completeErr, localRemErr)
+	}()
+
+	recC, errC := tc.StreamSessionEvents(ctx, *sessionID, 0)
+loop:
+	for {
+		select {
+		case rec, ok := <-recC:
+			if !ok {
+				break loop
+			}
+			prepared, err := e.PrepareSessionEvent(rec)
+			if err != nil {
+				return trace.Wrap(err, "preparing recording event")
+			}
+			err = e.RecordEvent(ctx, prepared)
+			if err != nil {
+				return trace.Wrap(err, "recording session event")
+			}
+		case err := <-errC:
+			if err != nil && !trace.IsEOF(err) {
+				return trace.Wrap(err, "downloading session recordings")
+			}
+			return nil
+		}
+	}
+	return nil
+}
+
+// createFileWriter creates a file-based session event writer that outputs to a file.
+func createFileWriter(ctx context.Context, sessionID session.ID, outputDir string) (*events.SessionWriter, error) {
+	fileStreamer, err := filesessions.NewStreamer(
+		filesessions.StreamerConfig{
+			Dir: outputDir,
+		},
+	)
+	if err != nil {
+		return nil, trace.Wrap(err, "failed to create fileStreamer")
+	}
+
+	e, err := events.NewSessionWriter(
+		events.SessionWriterConfig{
+			SessionID: sessionID,
+			Component: "downloader",
+			// Use NoOpPreparer as events are already prepared by the server.
+			Preparer: &events.NoOpPreparer{},
+			Context:  ctx,
+			Clock:    clockwork.NewRealClock(),
+			Streamer: fileStreamer,
+		},
+	)
+	return e, trace.Wrap(err, "creating session writer")
+}