diff --git a/CHANGELOG.md b/CHANGELOG.md index b2261d10efa..bae076dadb6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -23,6 +23,17 @@ rely on parsing the output from multiple nodes should pass the `--log-dir` flag to `tsh ssh`, which will create a directory where the separated output of each node will be written. +#### `tsh play` now streams PTY playback + +Prior to Teleport 15, `tsh play` would download the entire session recording +before starting playback. As a result, playback of large recordings could be +slow to start. In Teleport 15 session recordings are streamed from the auth +server, allowing playback to start before the entire session is downloaded and +unpacked. + +Additionally, `tsh play` now supports a `--speed` flag for adjusting the +playback speed. + #### `drop` host user creation mode The `drop` host user creation mode has been removed in Teleport 15. It is replaced diff --git a/docs/pages/connect-your-client/tsh.mdx b/docs/pages/connect-your-client/tsh.mdx index 519a3f77849..dcf5c35c4eb 100644 --- a/docs/pages/connect-your-client/tsh.mdx +++ b/docs/pages/connect-your-client/tsh.mdx @@ -83,10 +83,10 @@ to call [`tsh login`](../reference/cli/tsh.mdx#tsh-login) in the beginning. ## Installing tsh -Follow the instructions below to install the `tsh` binary. +Follow the instructions below to install the `tsh` binary. We recommend installing `tsh` of the same major version as the version used in -your Teleport cluster. +your Teleport cluster. To find the version number, either: @@ -97,7 +97,7 @@ To find the version number, either: - Use `curl` and `jq`. Replace with your Proxy Service address (e.g. `mytenant.teleport.sh` for Teleport Enterprise Cloud): - + ```code $ curl https:///webapi/find | jq '.server_version' "(=teleport.version=)" @@ -934,8 +934,6 @@ tmpfs 1982720 0 1982720 0% /proc/acpi tmpfs 1982720 0 1982720 0% /sys/firmware root@ubuntu:/# exit exit - -end of session playback ``` ## tsh configuration files diff --git a/lib/auth/auth_with_roles_test.go b/lib/auth/auth_with_roles_test.go index 6a2143c18b7..2abe1fdfedf 100644 --- a/lib/auth/auth_with_roles_test.go +++ b/lib/auth/auth_with_roles_test.go @@ -1734,8 +1734,9 @@ func TestStreamSessionEvents_Builtin(t *testing.T) { require.Empty(t, searchEvents) } -// TestGetSessionEvents ensures that when a user streams a session's events, it emits an audit event. -func TestGetSessionEvents(t *testing.T) { +// TestStreamSessionEvents ensures that when a user streams a session's events +// a "session recording access" event is emitted. +func TestStreamSessionEvents(t *testing.T) { t.Parallel() srv := newTestTLSServer(t) @@ -1748,12 +1749,14 @@ func TestGetSessionEvents(t *testing.T) { clt, err := srv.NewClient(identity) require.NoError(t, err) + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + // ignore the response as we don't want the events or the error (the session will not exist) - _, _ = clt.GetSessionEvents(apidefaults.Namespace, "44c6cea8-362f-11ea-83aa-125400432324", 0) + clt.StreamSessionEvents(ctx, session.ID("44c6cea8-362f-11ea-83aa-125400432324"), 0) // we need to wait for a short period to ensure the event is returned time.Sleep(500 * time.Millisecond) - ctx := context.Background() searchEvents, _, err := srv.AuthServer.AuditLog.SearchEvents(ctx, events.SearchEventsRequest{ From: srv.Clock().Now().Add(-time.Hour), To: srv.Clock().Now().Add(time.Hour), diff --git a/lib/auth/http_client.go b/lib/auth/http_client.go index 17b69db2470..b8267b09da9 100644 --- a/lib/auth/http_client.go +++ b/lib/auth/http_client.go @@ -852,7 +852,10 @@ func (c *HTTPClient) ValidateGithubAuthCallback(ctx context.Context, q url.Value // GetSessionChunk allows clients to receive a byte array (chunk) from a recorded // session stream, starting from 'offset', up to 'max' in length. The upper bound // of 'max' is set to events.MaxChunkBytes +// +// Deprecated: use StreamSessionEvents API instead func (c *HTTPClient) GetSessionChunk(namespace string, sid session.ID, offsetBytes, maxBytes int) ([]byte, error) { + // DELETE IN 16(zmb3): v15 web UIs stopped calling this if namespace == "" { return nil, trace.BadParameter(MissingNamespaceError) } @@ -867,12 +870,10 @@ func (c *HTTPClient) GetSessionChunk(namespace string, sid session.ID, offsetByt return response.Bytes(), nil } -// Returns events that happen during a session sorted by time -// (oldest first). -// -// afterN allows to filter by "newer than N" value where N is the cursor ID -// of previously returned bunch (good for polling for latest) +// Deprecated: use StreamSessionEvents API instead. +// TODO(zmb3): remove from ClientI interface func (c *HTTPClient) GetSessionEvents(namespace string, sid session.ID, afterN int) (retval []events.EventFields, err error) { + // DELETE IN 16(zmb3): v15 web UIs stopped calling this if namespace == "" { return nil, trace.BadParameter(MissingNamespaceError) } diff --git a/lib/client/api.go b/lib/client/api.go index 2969fd5e8dd..0eeea09f5e7 100644 --- a/lib/client/api.go +++ b/lib/client/api.go @@ -83,6 +83,7 @@ import ( "github.com/gravitational/teleport/lib/modules" "github.com/gravitational/teleport/lib/multiplexer" "github.com/gravitational/teleport/lib/observability/tracing" + "github.com/gravitational/teleport/lib/player" "github.com/gravitational/teleport/lib/services" "github.com/gravitational/teleport/lib/session" "github.com/gravitational/teleport/lib/shell" @@ -1989,8 +1990,8 @@ func (tc *TeleportClient) Join(ctx context.Context, mode types.SessionParticipan return trace.Wrap(err) } -// Play replays the recorded session -func (tc *TeleportClient) Play(ctx context.Context, namespace, sessionID string) (err error) { +// Play replays the recorded session. +func (tc *TeleportClient) Play(ctx context.Context, sessionID string, speed float64) error { ctx, span := tc.Tracer.Start( ctx, "teleportClient/Play", @@ -2001,140 +2002,142 @@ func (tc *TeleportClient) Play(ctx context.Context, namespace, sessionID string) ) defer span.End() - var sessionEvents []events.EventFields - var stream []byte - if namespace == "" { - return trace.BadParameter(auth.MissingNamespaceError) - } - sid, err := session.ParseID(sessionID) - if err != nil { - return fmt.Errorf("'%v' is not a valid session ID (must be GUID)", sid) - } - // connect to the auth server (site) who made the recording proxyClient, err := tc.ConnectToProxy(ctx) if err != nil { return trace.Wrap(err) } defer proxyClient.Close() - site := proxyClient.CurrentCluster() + return playSession(ctx, sessionID, speed, proxyClient.CurrentCluster()) +} - // request events for that session (to get timing data) - sessionEvents, err = site.GetSessionEvents(namespace, *sid, 0) +const ( + keyCtrlC = 3 + keyCtrlD = 4 + keySpace = 32 + keyLeft = 68 + keyRight = 67 + keyUp = 65 + keyDown = 66 +) + +func playSession(ctx context.Context, sessionID string, speed float64, streamer player.Streamer) error { + sid, err := session.ParseID(sessionID) if err != nil { return trace.Wrap(err) } - // Return an error if it is a desktop session and check to see if this is not a Kube or SSH session - if len(sessionEvents) > 0 { - switch typ := sessionEvents[0].GetType(); typ { - case events.WindowsDesktopSessionStartEvent: - url := getDesktopEventWebURL(tc.localAgent.proxyHost, proxyClient.siteName, sid, sessionEvents) - message := "Desktop sessions cannot be viewed with tsh." + - " Please use the browser to play this session." + - " Click on the URL to view the session in the browser:" - return trace.BadParameter("%s\n%s", message, url) - case events.AppSessionStartEvent, events.DatabaseSessionStartEvent, events.AppSessionChunkEvent: - return trace.BadParameter("Interactive session replay with tsh is supported for SSH and Kubernetes sessions."+ - " To play entries for Application and Database you must use the json or yaml format."+ - " \nEx: tsh play -f json %s", sid) - case events.SessionStartEvent: - // proceed without error - default: - return trace.BadParameter("unknown session type %q", typ) - } + term, err := terminal.New(os.Stdin, os.Stdout, os.Stderr) + if err != nil { + return trace.Wrap(err) } + defer term.Close() - // read the stream into a buffer: - for { - tmp, err := site.GetSessionChunk(namespace, *sid, len(stream), events.MaxChunkBytes) + // configure terminal for direct unbuffered echo-less input + if term.IsAttached() { + err := term.InitRaw(true) if err != nil { return trace.Wrap(err) } - if len(tmp) == 0 { - break + } + + term.Clear() // clear screen between runs: + term.SetCursorPos(1, 1) + + player, err := player.New(&player.Config{ + SessionID: *sid, + Streamer: streamer, + }) + if err != nil { + return trace.Wrap(err) + } + + player.SetSpeed(speed) + if err := player.Play(); err != nil { + return trace.Wrap(err) + } + playing := true + + // playback control goroutine + const skipDuration = 10 * time.Second + go func() { + var key [1]byte + for { + _, err := term.Stdin().Read(key[:]) + if err != nil { + return + } + switch key[0] { + case keyCtrlC, keyCtrlD: + player.Close() + return + case keySpace: + if playing { + player.Pause() + } else { + player.Play() + } + playing = !playing + case keyLeft, keyDown: + current := time.Duration(player.LastPlayed() * int64(time.Millisecond)) + player.SetPos(max(current-skipDuration, 0)) // rewind + term.Clear() + term.SetCursorPos(1, 1) + case keyRight, keyUp: + current := time.Duration(player.LastPlayed() * int64(time.Millisecond)) + player.SetPos(current + skipDuration) // advance forward + } } - stream = append(stream, tmp...) - } + }() - return playSession(sessionEvents, stream) -} - -func (tc *TeleportClient) GetSessionEvents(ctx context.Context, namespace, sessionID string) ([]events.EventFields, error) { - ctx, span := tc.Tracer.Start( - ctx, - "teleportClient/GetSessionEvents", - oteltrace.WithSpanKind(oteltrace.SpanKindClient), - oteltrace.WithAttributes( - attribute.String("session", sessionID), - ), - ) - defer span.End() - - if namespace == "" { - return nil, trace.BadParameter(auth.MissingNamespaceError) - } - sid, err := session.ParseID(sessionID) - if err != nil { - return nil, trace.BadParameter("%q is not a valid session ID (must be GUID)", sid) - } - // connect to the auth server (site) who made the recording - proxyClient, err := tc.ConnectToProxy(ctx) - if err != nil { - return nil, trace.Wrap(err) - } - defer proxyClient.Close() - - site := proxyClient.CurrentCluster() - - events, err := site.GetSessionEvents(namespace, *sid, 0) - if err != nil { - return nil, trace.Wrap(err) - } - return events, nil -} - -// PlayFile plays the recorded session from a tar file -func PlayFile(ctx context.Context, tarFile io.Reader, sid string) error { - var sessionEvents []events.EventFields - var stream []byte - protoReader := events.NewProtoReader(tarFile) - playbackDir, err := os.MkdirTemp("", "playback") - if err != nil { - return trace.Wrap(err) - } - defer os.RemoveAll(playbackDir) - w, err := events.WriteForSSHPlayback(ctx, session.ID(sid), protoReader, playbackDir) - if err != nil { - return trace.Wrap(err) - } - sessionEvents, err = w.SessionEvents() - if err != nil { - return trace.Wrap(err) - } - // Return errors if this is desktop, app, db or unknown session. - if len(sessionEvents) > 0 { - switch typ := sessionEvents[0].GetType(); typ { - case events.WindowsDesktopSessionStartEvent: - message := "Desktop sessions cannot be viewed with tsh." + - " Please use the browser to play this session or use tsh recordings export to get a video download." - return trace.BadParameter("%s", message) - case events.AppSessionStartEvent, events.DatabaseSessionStartEvent, events.AppSessionChunkEvent: - return trace.BadParameter("Interactive session replay with tsh is supported for SSH and Kubernetes sessions."+ - " To play entries for Application and Database you must use the json or yaml format."+ - "\nEx: tsh play -f json %s", sid) - case events.SessionStartEvent: - // proceed without error + var lastTime time.Time + for evt := range player.C() { + switch evt := evt.(type) { + case *apievents.WindowsDesktopSessionStart: + // TODO(zmb3): restore the playback URL + message := "Desktop sessions cannot be played with tsh play." + + " Export the recording to video with tsh recordings export" + + " or view the recording in your web browser." + return trace.BadParameter(message) + case *apievents.AppSessionStart, *apievents.DatabaseSessionStart, *apievents.AppSessionChunk: + return trace.BadParameter("Interactive session replay is only supported for SSH and Kubernetes sessions." + + " To play app or database sessions, specify --format=json or --format=yaml.") + case *apievents.Resize: + if err := setTermSize(term.Stdout(), evt.TerminalSize); err != nil { + continue + } + case *apievents.SessionStart: + if err := setTermSize(term.Stdout(), evt.TerminalSize); err != nil { + continue + } + case *apievents.SessionPrint: + term.Stdout().Write(evt.Data) + if evt.Time != lastTime { + term.SetWindowTitle(evt.Time.Format(time.Stamp)) + } + lastTime = evt.Time default: - return trace.BadParameter("unknown session type %q", typ) + continue } } - stream, err = w.SessionChunks() - if err != nil { - return trace.Wrap(err) - } - return playSession(sessionEvents, stream) + return nil +} + +func setTermSize(w io.Writer, size string) error { + width, height, ok := strings.Cut(size, ":") + if !ok { + return trace.Errorf("invalid terminal size %q", size) + } + // resize terminal window by sending control sequence: + _, err := fmt.Fprintf(w, "\x1b[8;%s;%st", height, width) + return err +} + +// PlayFile plays the recorded session from a file. +func PlayFile(ctx context.Context, filename, sid string, speed float64) error { + streamer := &playFromFileStreamer{filename: filename} + return playSession(ctx, sid, speed, streamer) } // SFTP securely copies files between Nodes or SSH servers using SFTP @@ -5013,73 +5016,6 @@ func isFIPS() bool { return modules.GetModules().IsBoringBinary() } -// playSession plays session in the terminal -func playSession(sessionEvents []events.EventFields, stream []byte) error { - term, err := terminal.New(nil, nil, nil) - if err != nil { - return trace.Wrap(err) - } - - defer term.Close() - - // configure terminal for direct unbuffered echo-less input: - if term.IsAttached() { - err := term.InitRaw(true) - if err != nil { - return trace.Wrap(err) - } - } - - player := newSessionPlayer(sessionEvents, stream, term) - errorCh := make(chan error) - // keys: - const ( - keyCtrlC = 3 - keyCtrlD = 4 - keySpace = 32 - keyLeft = 68 - keyRight = 67 - keyUp = 65 - keyDown = 66 - ) - // playback control goroutine - go func() { - defer player.EndPlayback() - var key [1]byte - for { - _, err := term.Stdin().Read(key[:]) - if err != nil { - errorCh <- err - return - } - switch key[0] { - // Ctrl+C or Ctrl+D - case keyCtrlC, keyCtrlD: - return - // Space key - case keySpace: - player.TogglePause() - // <- arrow - case keyLeft, keyDown: - player.Rewind() - // -> arrow - case keyRight, keyUp: - player.Forward() - } - } - }() - // player starts playing in its own goroutine - player.Play() - // wait for keypresses loop to end - select { - case <-player.stopC: - fmt.Println("\n\nend of session playback") - return nil - case err := <-errorCh: - return trace.Wrap(err) - } -} - func findActiveDatabases(key *Key) ([]tlsca.RouteToDatabase, error) { dbCerts, err := key.DBTLSCertificates() if err != nil { diff --git a/lib/client/player.go b/lib/client/player.go index bc4985440d3..69c68bcdc9e 100644 --- a/lib/client/player.go +++ b/lib/client/player.go @@ -19,287 +19,51 @@ package client import ( - "fmt" + "context" "os" - "strings" - "sync" - "time" - "github.com/jonboulle/clockwork" - "github.com/sirupsen/logrus" + "github.com/gravitational/trace" - "github.com/gravitational/teleport/lib/client/terminal" + apievents "github.com/gravitational/teleport/api/types/events" "github.com/gravitational/teleport/lib/events" + "github.com/gravitational/teleport/lib/session" ) -type tshPlayerState int - -const ( - // The player has stopped, either for an action to take place, - // because the playback has reached the end of the recording, - // or because a hard stop was requested. - stateStopped tshPlayerState = iota - // A stop has been requested so that an action (forward, rewind, etc) can take place. - stateStopping - // An end to the playback has been requested. - stateEnding - // The player is playing. - statePlaying -) - -// sessionPlayer implements replaying terminal sessions. It runs a playback goroutine -// and allows to control it -type sessionPlayer struct { - sync.Mutex - cond *sync.Cond - - state tshPlayerState - position int // position is the index of the last event successfully played back - - clock clockwork.Clock - stream []byte - sessionEvents []events.EventFields - term *terminal.Terminal - - // stopC is closed when playback ends (either because the end of the stream has - // been reached, or a hard stop was requested via EndPlayback(). - stopC chan struct{} - stopOnce sync.Once - - log *logrus.Logger +// playFromFileStreamer implements [player.Streamer] for +// streaming from a local file. +type playFromFileStreamer struct { + filename string } -func newSessionPlayer(sessionEvents []events.EventFields, stream []byte, term *terminal.Terminal) *sessionPlayer { - p := &sessionPlayer{ - clock: clockwork.NewRealClock(), - position: -1, // position is the last successfully written event - stream: stream, - sessionEvents: sessionEvents, - term: term, - stopC: make(chan struct{}), - log: logrus.New(), - } - p.cond = sync.NewCond(p) - return p -} +func (p *playFromFileStreamer) StreamSessionEvents( + ctx context.Context, + sessionID session.ID, + startIndex int64, +) (chan apievents.AuditEvent, chan error) { + evts := make(chan apievents.AuditEvent) + errs := make(chan error, 1) -func (p *sessionPlayer) Play() { - p.playRange(0, 0) -} - -func (p *sessionPlayer) Stopped() bool { - p.Lock() - defer p.Unlock() - return p.state == stateStopped -} - -func (p *sessionPlayer) Rewind() { - p.Lock() - defer p.Unlock() - if p.state != stateStopped { - p.setState(stateStopping) - p.waitUntil(stateStopped) - } - if p.position > 0 { - p.playRange(p.position-1, p.position) - } -} - -func (p *sessionPlayer) stopOrEndRequested() bool { - p.Lock() - defer p.Unlock() - return p.state == stateStopping || p.state == stateEnding -} - -func (p *sessionPlayer) Forward() { - p.Lock() - defer p.Unlock() - if p.state != stateStopped { - p.setState(stateStopping) - p.waitUntil(stateStopped) - } - if p.position < len(p.sessionEvents) { - p.playRange(p.position+2, p.position+2) - } -} - -func (p *sessionPlayer) TogglePause() { - p.Lock() - defer p.Unlock() - if p.state == statePlaying { - p.setState(stateStopping) - p.waitUntil(stateStopped) - } else { - p.playRange(p.position+1, 0) - p.waitUntil(statePlaying) - } -} - -// EndPlayback makes an asynchronous request for the player to end the playback. -// Playback might not stop before this method returns. -func (p *sessionPlayer) EndPlayback() { - p.Lock() - defer p.Unlock() - - switch p.state { - case stateEnding: - // We're already ending, no need to do anything. - case stateStopped: - // The playRange goroutine has already returned, so we can - // signal the end of playback by closing the stopC channel right here. - p.close() - case stateStopping, statePlaying: - // The playRange goroutine is still running, and may be sleeping - // while waiting for the right time to print the next characters. - // setState to stateEnding so that the playRange goroutine - // knows to return on the next loop. The stopC channel will - // be closed by the playback routine upon completion. - p.setState(stateEnding) - default: - // Cases should be exhaustive, this should never happen. - p.log.Error("unexpected playback error") - } -} - -// waitUntil waits for the specified state to be reached. -// Callers must hold the lock on p.Mutex before calling. -func (p *sessionPlayer) waitUntil(state tshPlayerState) { - for state != p.state { - p.cond.Wait() - } -} - -// setState sets the current player state and notifies any -// goroutines waiting in waitUntil(). Callers must hold the -// lock on p.Mutex before calling. -func (p *sessionPlayer) setState(state tshPlayerState) { - p.state = state - p.cond.Broadcast() -} - -// timestampFrame prints 'event timestamp' in the top right corner of the -// terminal after playing every 'print' event -func timestampFrame(term *terminal.Terminal, message string) { - const ( - saveCursor = "7" - restoreCursor = "8" - ) - width, _, err := term.Size() - if err != nil { - return - } - esc := func(s string) { - os.Stdout.Write([]byte("\x1b" + s)) - } - esc(saveCursor) - defer esc(restoreCursor) - - // move cursor to -10:0 - // TODO(timothyb89): message length does not account for unicode characters - // or ANSI sequences. - esc(fmt.Sprintf("[%d;%df", 0, int(width)-len(message))) - os.Stdout.WriteString(message) -} - -func (p *sessionPlayer) close() { - p.stopOnce.Do(func() { close(p.stopC) }) -} - -// playRange plays events from a given from:to range. In order for the replay -// to render correctly, playRange always plays from the beginning, but starts -// applying timing info (delays) only after 'from' event, creating an impression -// that playback starts from there. -func (p *sessionPlayer) playRange(from, to int) { - if to > len(p.sessionEvents) || from < 0 { - p.Lock() - p.setState(stateStopped) - p.Unlock() - return - } - if to == 0 { - to = len(p.sessionEvents) - } - // clear screen between runs: - os.Stdout.Write([]byte("\x1bc")) - - // playback goroutine: go func() { - var i int + f, err := os.Open(p.filename) + if err != nil { + errs <- trace.ConvertSystemError(err) + return + } + defer f.Close() - defer func() { - - p.Lock() - endRequested := p.state == stateEnding - p.setState(stateStopped) - p.Unlock() - - // An end was manually requested, or we played the last event? - if endRequested || i == len(p.sessionEvents) { - p.close() - } - }() - - p.Lock() - p.setState(statePlaying) - p.Unlock() - - prev := time.Duration(0) - offset, bytes := 0, 0 - for i = 0; i < to; i++ { - if p.stopOrEndRequested() { + pr := events.NewProtoReader(f) + for i := int64(0); ; i++ { + evt, err := pr.Read(ctx) + if err != nil { + errs <- trace.Wrap(err) return } - e := p.sessionEvents[i] - - switch e.GetString(events.EventType) { - // 'print' event (output) - case events.SessionPrintEvent: - // delay is only necessary once we've caught up to the "from" event - if i >= from { - prev = p.applyDelay(prev, e) - } - offset = e.GetInt("offset") - bytes = e.GetInt("bytes") - os.Stdout.Write(p.stream[offset : offset+bytes]) - // resize terminal event (also on session start) - case events.ResizeEvent, events.SessionStartEvent: - parts := strings.Split(e.GetString("size"), ":") - if len(parts) != 2 { - continue - } - width, height := parts[0], parts[1] - // resize terminal window by sending control sequence: - os.Stdout.Write([]byte(fmt.Sprintf("\x1b[8;%s;%st", height, width))) - default: - continue + if i >= startIndex { + evts <- evt } - p.Lock() - p.position = i - p.Unlock() } }() -} -// applyDelay waits until it is time to play back the current event. -// It returns the duration from the start of the session up until the current event. -func (p *sessionPlayer) applyDelay(previousTimestamp time.Duration, e events.EventFields) time.Duration { - eventTime := time.Duration(e.GetInt("ms") * int(time.Millisecond)) - delay := eventTime - previousTimestamp - - // make playback smoother: - switch { - case delay < 10*time.Millisecond: - delay = 0 - case delay > 250*time.Millisecond && delay < 500*time.Millisecond: - delay = 250 * time.Millisecond - case delay > 500*time.Millisecond && delay < 1*time.Second: - delay = 500 * time.Millisecond - case delay > time.Second: - delay = time.Second - } - - timestampFrame(p.term, e.GetString("time")) - p.clock.Sleep(delay) - return eventTime + return evts, errs } diff --git a/lib/client/player_test.go b/lib/client/player_test.go deleted file mode 100644 index e4cee0565ff..00000000000 --- a/lib/client/player_test.go +++ /dev/null @@ -1,247 +0,0 @@ -/* - * Teleport - * Copyright (C) 2023 Gravitational, Inc. - * - * This program is free software: you can redistribute it and/or modify - * it under the terms of the GNU Affero General Public License as published by - * the Free Software Foundation, either version 3 of the License, or - * (at your option) any later version. - * - * This program is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU Affero General Public License for more details. - * - * You should have received a copy of the GNU Affero General Public License - * along with this program. If not, see . - */ - -package client - -import ( - "bytes" - "testing" - "time" - - "github.com/jonboulle/clockwork" - "github.com/stretchr/testify/require" - - "github.com/gravitational/teleport/lib/client/terminal" - "github.com/gravitational/teleport/lib/events" -) - -// TestEmptyPlay verifies that a playback of 0 events -// immediately transitions to a stopped state. -func TestEmptyPlay(t *testing.T) { - c := clockwork.NewFakeClock() - p := newSessionPlayer(nil, nil, testTerm(t)) - p.clock = c - - p.Play() - - select { - case <-time.After(2 * time.Second): - t.Fatal("timed out waiting for player to complete") - case <-p.stopC: - } - - require.True(t, p.Stopped(), "p.Stopped() returned an unexpected value") -} - -// TestStop verifies that we can stop playback. -func TestStop(t *testing.T) { - c := clockwork.NewFakeClock() - events := printEvents(100, 200) - p := newSessionPlayer(events, nil, testTerm(t)) - p.clock = c - - p.Play() - - // wait for player to see the first event and apply the delay - c.BlockUntil(1) - - p.EndPlayback() - - // advance the clock: - // at this point, the player will write the first event and then - // see that we requested a stop - c.Advance(100 * time.Millisecond) - - require.Eventually(t, p.Stopped, 2*time.Second, 200*time.Millisecond) -} - -// TestPlayPause verifies the play/pause functionality. -func TestPlayPause(t *testing.T) { - c := clockwork.NewFakeClock() - - // in this test, we let the player play 2 of the 3 events, - // then pause it and verify the pause state before resuming - // playback for the final event. - events := printEvents(100, 200, 300) - var stream []byte // intentionally empty, we dont care about stream contents here - p := newSessionPlayer(events, stream, testTerm(t)) - p.clock = c - - p.Play() - - // wait for player to see the first event and apply the delay - c.BlockUntil(1) - - // advance the clock: - // at this point, the player will write the first event - c.Advance(100 * time.Millisecond) - - // wait for the player to sleep on the 2nd event - c.BlockUntil(1) - - // pause playback - // note: we don't use p.TogglePause here, as it waits for the state transition, - // and the state won't transition until we advance the clock - p.Lock() - p.setState(stateStopping) - p.Unlock() - - // advance the clock again: - // the player will write the second event and - // then realize that it's been asked to pause - c.Advance(100 * time.Millisecond) - - p.Lock() - p.waitUntil(stateStopped) - p.Unlock() - - ch := make(chan struct{}) - go func() { - // resume playback - p.TogglePause() - ch <- struct{}{} - }() - - // playback should resume for the 3rd and final event: - // in this case, the first two events are written immediately without delay, - // and we block here until the player is sleeping prior to the 3rd event - c.BlockUntil(1) - - // make sure that we've resumed - <-ch - require.False(t, p.Stopped(), "p.Stopped() returned true when it should have returned false") - - // advance the clock a final time, forcing the player to write the last event - // note: on the resume, we play the successful events immediately, and then sleep - // up to the resume point, which is why we advance by 300ms here - c.Advance(300 * time.Millisecond) - - select { - case <-time.After(2 * time.Second): - t.Fatal("timed out waiting for player to complete") - case <-p.stopC: - } - require.True(t, p.Stopped(), "p.Stopped() returned an unexpected value") -} - -func TestEndPlaybackWhilePlaying(t *testing.T) { - c := clockwork.NewFakeClock() - - // in this test, we let the player play 1 of the 2 events, - // then end the playback and confirm - // that the stopC channel was written to. - events := printEvents(100, 200) - var stream []byte // intentionally empty, we dont care about stream contents here - p := newSessionPlayer(events, stream, testTerm(t)) - p.clock = c - - p.Play() - - // wait for player to see the first event and apply the delay - c.BlockUntil(1) - - // end playback - p.EndPlayback() - - // advance the clock: - // the player will write the first event and - // then realize that it's been asked to end playback - c.Advance(100 * time.Millisecond) - - // check that stopC was written to - select { - case <-time.After(2 * time.Second): - t.Fatal("timed out waiting for player to complete") - case <-p.stopC: - require.True(t, p.Stopped(), "p.Stopped() returned an unexpected value") - } -} - -// TestEndPlaybackWhilePaused tests that playback can be ended -// by calling EndPlayback while playback is paused. -func TestEndPlaybackWhilePaused(t *testing.T) { - c := clockwork.NewFakeClock() - - // in this test, we let the player play 1 of the 2 events, - // then pause it and verify the pause state before ending playback. - events := printEvents(100, 200) - var stream []byte // intentionally empty, we dont care about stream contents here - p := newSessionPlayer(events, stream, testTerm(t)) - p.clock = c - - p.Play() - - // wait for player to see the first event and apply the delay - c.BlockUntil(1) - - // advance the clock: - // at this point, the player will write the first event - c.Advance(100 * time.Millisecond) - - // wait for the player to sleep on the 2nd event - c.BlockUntil(1) - - // pause playback - // note: we don't use p.TogglePause here, as it waits for the state transition, - // and the state won't transition until we advance the clock - p.Lock() - p.setState(stateStopping) - p.Unlock() - - // advance the clock again: - // the player will write the second event and - // then realize that it's been asked to pause - c.Advance(100 * time.Millisecond) - - // wait until the pause is in effect - p.Lock() - p.waitUntil(stateStopped) - p.Unlock() - - // end playback - p.EndPlayback() - - // check that stopC was written to - select { - case <-time.After(2 * time.Second): - t.Fatal("timed out waiting for player to complete") - case <-p.stopC: - require.True(t, p.Stopped(), "p.Stopped() returned an unexpected value") - } -} - -func testTerm(t *testing.T) *terminal.Terminal { - t.Helper() - term, err := terminal.New(bytes.NewReader(nil), &bytes.Buffer{}, &bytes.Buffer{}) - require.NoError(t, err) - t.Cleanup(func() { - require.NoError(t, term.Close()) - }) - return term -} - -func printEvents(delays ...int) []events.EventFields { - result := make([]events.EventFields, len(delays)) - for i := range result { - result[i] = events.EventFields{ - events.EventType: events.SessionPrintEvent, - "ms": delays[i], - } - } - return result -} diff --git a/lib/client/terminal/terminal_common.go b/lib/client/terminal/terminal_common.go index 4ffdb9201c6..a2e829b19ae 100644 --- a/lib/client/terminal/terminal_common.go +++ b/lib/client/terminal/terminal_common.go @@ -19,6 +19,7 @@ package terminal import ( + "fmt" "sync" "github.com/gravitational/trace" @@ -72,6 +73,19 @@ func (e *signalEmitter) clearSubscribers() { e.subscribers = e.subscribers[:0] } +// SetCursorPos sets the cursor position to the given x, y coordinates. +// Coordinates are 1-indexed. (1, 1) represents the top left corner. +func (t *Terminal) SetCursorPos(x, y int) error { + _, err := fmt.Fprintf(t.stdout, "\x1b[%d;%dH", y, x) + return trace.Wrap(err) +} + +// SetWindowTitle sets the terminal window's title. +func (t *Terminal) SetWindowTitle(s string) error { + _, err := fmt.Fprintf(t.stdout, "\x1b]0;%s\a", s) + return trace.Wrap(err) +} + // Clear clears the terminal, including scrollback. func (t *Terminal) Clear() error { // \x1b[3J - clears scrollback (it is needed at least for the Mac terminal) - diff --git a/lib/client/terminal/terminal_unix.go b/lib/client/terminal/terminal_unix.go index 58e13c989a3..479ddfcfc49 100644 --- a/lib/client/terminal/terminal_unix.go +++ b/lib/client/terminal/terminal_unix.go @@ -141,7 +141,6 @@ func (t *Terminal) InitRaw(input bool) error { }() // NOTE: Unix does not require any special input handling. - return nil } @@ -164,22 +163,30 @@ func (t *Terminal) IsAttached() bool { // Resize makes a best-effort attempt to resize the terminal window. Support // varies between platforms and terminal emulators. func (t *Terminal) Resize(width, height int16) error { - _, err := os.Stdout.Write([]byte(fmt.Sprintf("\x1b[8;%d;%dt", height, width))) - + _, err := fmt.Fprintf(t.stdout, "\x1b[8;%d;%dt", height, width) return trace.Wrap(err) } -func (t *Terminal) Stdin() io.Reader { - return t.stdin +const ( + saveCursor = "7" + restoreCursor = "8" +) + +// SaveCursor saves the current cursor position. +func (t *Terminal) SaveCursor() error { + _, err := t.stdout.Write([]byte("\x1b" + saveCursor)) + return trace.Wrap(err) } -func (t *Terminal) Stdout() io.Writer { - return t.stdout +// RestoreCursor restores the last saved cursor position. +func (t *Terminal) RestoreCursor() error { + _, err := t.stdout.Write([]byte("\x1b" + restoreCursor)) + return trace.Wrap(err) } -func (t *Terminal) Stderr() io.Writer { - return t.stderr -} +func (t *Terminal) Stdin() io.Reader { return t.stdin } +func (t *Terminal) Stdout() io.Writer { return t.stdout } +func (t *Terminal) Stderr() io.Writer { return t.stderr } // Close closes the Terminal, restoring the console to its original state. func (t *Terminal) Close() error { diff --git a/lib/events/auditlog.go b/lib/events/auditlog.go index 3f7ba71f928..d64708b97c7 100644 --- a/lib/events/auditlog.go +++ b/lib/events/auditlog.go @@ -812,7 +812,6 @@ func (l *AuditLog) getSessionChunk(namespace string, sid session.ID, offsetBytes // (oldest first). // // Can be filtered by 'after' (cursor value to return events newer than) - func (l *AuditLog) GetSessionEvents(namespace string, sid session.ID, afterN int) ([]EventFields, error) { l.log.WithFields(log.Fields{"sid": string(sid), "afterN": afterN}).Debugf("GetSessionEvents.") if namespace == "" { diff --git a/lib/session/session.go b/lib/session/session.go index 64b5073fae9..76c02d11100 100644 --- a/lib/session/session.go +++ b/lib/session/session.go @@ -56,7 +56,7 @@ func (s *ID) Check() error { func ParseID(id string) (*ID, error) { _, err := uuid.Parse(id) if err != nil { - return nil, trace.BadParameter("%v not a valid UUID", id) + return nil, trace.BadParameter("%v is not a valid UUID", id) } uid := ID(id) return &uid, nil diff --git a/tool/tsh/common/play.go b/tool/tsh/common/play.go new file mode 100644 index 00000000000..73835f1e918 --- /dev/null +++ b/tool/tsh/common/play.go @@ -0,0 +1,249 @@ +/* +Copyright 2016-2023 Gravitational, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package common + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "os" + "path" + "path/filepath" + "strings" + + "github.com/ghodss/yaml" + "github.com/gravitational/trace" + + "github.com/gravitational/teleport" + apievents "github.com/gravitational/teleport/api/types/events" + "github.com/gravitational/teleport/lib/client" + "github.com/gravitational/teleport/lib/events" + "github.com/gravitational/teleport/lib/session" +) + +// onPlay is used to interact with recorded sessions. +// It has several modes: +// +// 1. If --format is "pty" (the default), then the recorded +// session is played back in the user's terminal. +// 2. Otherwise, `tsh play` is used to export a session from the +// binary protobuf format into YAML or JSON. +// +// Each of these modes has two subcases: +// i) --session-id ends with ".tar" - tsh operates on a local +// file containing a previously downloaded session +// +// b) --session-id is the ID of a session - tsh operates on the +// session recording by connecting to the Teleport cluster +func onPlay(cf *CLIConf) error { + if format := strings.ToLower(cf.Format); format == teleport.PTY { + return playSession(cf) + } + if cf.PlaySpeed != "1x" { + log.Warn("--speed is not applicable for formats other than pty") + } + return exportSession(cf) +} + +var playbackSpeeds = map[string]float64{ + "0.5x": 0.5, + "1x": 1.0, + "2x": 2.0, + "4x": 4.0, + "8x": 8.0, +} + +// playSession implements `tsh play` for the PTY format. +func playSession(cf *CLIConf) error { + speed, ok := playbackSpeeds[cf.PlaySpeed] + if !ok { + speed = 1.0 + } + + isLocalFile := path.Ext(cf.SessionID) == ".tar" + if isLocalFile { + sid := sessionIDFromPath(cf.SessionID) + if err := client.PlayFile(cf.Context, cf.SessionID, sid, speed); err != nil { + return trace.Wrap(err) + } + return nil + } + + tc, err := makeClient(cf) + if err != nil { + return trace.Wrap(err) + } + + if err := tc.Play(cf.Context, cf.SessionID, speed); err != nil { + if trace.IsNotFound(err) { + log.WithError(err).Debug("error playing session") + return trace.NotFound("Recording for session %s not found.", cf.SessionID) + } + return trace.Wrap(err) + } + return nil +} + +func sessionIDFromPath(path string) string { + fileName := filepath.Base(path) + return strings.TrimSuffix(fileName, ".tar") +} + +// exportSession implements `tsh play` for formats other than PTY +func exportSession(cf *CLIConf) error { + format := strings.ToLower(cf.Format) + isLocalFile := path.Ext(cf.SessionID) == ".tar" + if isLocalFile { + return trace.Wrap(exportFile(cf.Context, cf.SessionID, format)) + } + + switch format { + case teleport.JSON, teleport.YAML: + default: + return trace.Errorf("Invalid format %s, only json and yaml are supported", format) + } + + sid, err := session.ParseID(cf.SessionID) + if err != nil { + return trace.BadParameter("'%v' is not a valid session ID (must be GUID)", cf.SessionID) + } + + tc, err := makeClient(cf) + if err != nil { + return trace.Wrap(err) + } + + proxyClient, err := tc.ConnectToProxy(cf.Context) + if err != nil { + return trace.Wrap(err) + } + defer proxyClient.Close() + + site := proxyClient.CurrentCluster() + evts, errs := site.StreamSessionEvents(cf.Context, *sid, 0) + + var exporter sessionExporter + switch format { + case teleport.JSON: + exporter = jsonSessionExporter{} + case teleport.YAML: + exporter = yamlSessionExporter{} + } + + exporter.WriteStart() + defer exporter.WriteEnd() + first := true + + for { + select { + case err := <-errs: + return trace.Wrap(err) + case event, ok := <-evts: + if !ok { + return nil + } + // when playing from a file, id is not included, this + // makes the outputs otherwise identical + event.SetID("") + + if first { + first = false + } else { + exporter.WriteSeparator() + } + + if err := exporter.WriteEvent(event); err != nil { + return trace.Wrap(err) + } + + } + } +} + +type sessionExporter interface { + WriteStart() error + WriteEnd() error + WriteSeparator() error + WriteEvent(evt apievents.AuditEvent) error +} + +type jsonSessionExporter struct{} + +func (jsonSessionExporter) WriteStart() error { + _, err := fmt.Println("[") + return err +} + +func (jsonSessionExporter) WriteEnd() error { + _, err := fmt.Println("]") + return err +} + +func (jsonSessionExporter) WriteSeparator() error { + _, err := fmt.Print(",\n") + return err +} + +func (jsonSessionExporter) WriteEvent(evt apievents.AuditEvent) error { + b, err := json.MarshalIndent(evt, " ", " ") + if err != nil { + return trace.Wrap(err) + } + + // JSON prefix does not apply to the first line, so add it manually + os.Stdout.Write([]byte(" ")) + + _, err = os.Stdout.Write(bytes.TrimSpace(b)) + return trace.Wrap(err) +} + +type yamlSessionExporter struct{} + +func (yamlSessionExporter) WriteStart() error { return nil } + +func (yamlSessionExporter) WriteEnd() error { return nil } + +func (yamlSessionExporter) WriteSeparator() error { + _, err := fmt.Println("---") + return err +} + +func (yamlSessionExporter) WriteEvent(evt apievents.AuditEvent) error { + b, err := yaml.Marshal(evt) + if err != nil { + return err + } + _, err = os.Stdout.Write(b) + return err +} + +// exportFile converts the binary protobuf events from the file +// identified by path to text (JSON/YAML) and writes the converted +// events to standard out. +func exportFile(ctx context.Context, path string, format string) error { + f, err := os.Open(path) + if err != nil { + return trace.ConvertSystemError(err) + } + defer f.Close() + err = events.Export(ctx, f, os.Stdout, format) + if err != nil { + return trace.Wrap(err) + } + return nil +} diff --git a/tool/tsh/common/tsh.go b/tool/tsh/common/tsh.go index 69f57fe39ba..868ae4f5077 100644 --- a/tool/tsh/common/tsh.go +++ b/tool/tsh/common/tsh.go @@ -32,7 +32,6 @@ import ( "os/exec" "os/signal" "path" - "path/filepath" "regexp" "runtime" "runtime/pprof" @@ -76,7 +75,6 @@ import ( dbprofile "github.com/gravitational/teleport/lib/client/db" "github.com/gravitational/teleport/lib/client/identityfile" "github.com/gravitational/teleport/lib/defaults" - "github.com/gravitational/teleport/lib/events" "github.com/gravitational/teleport/lib/kube/kubeconfig" "github.com/gravitational/teleport/lib/modules" "github.com/gravitational/teleport/lib/observability/tracing" @@ -278,6 +276,9 @@ type CLIConf struct { Format string OutFile string + // PlaySpeed controls the playback speed for tsh play. + PlaySpeed string + // SearchKeywords is a list of search keywords to match against resource field values. SearchKeywords string @@ -899,6 +900,7 @@ func Run(ctx context.Context, args []string, opts ...CliOption) error { // play play := app.Command("play", "Replay the recorded session (SSH, Kubernetes, App, DB).") play.Flag("cluster", clusterHelp).Short('c').StringVar(&cf.SiteName) + play.Flag("speed", "Playback speed, applicable when streaming SSH or Kubernetes sessions.").Default("1x").EnumVar(&cf.PlaySpeed, "0.5x", "1x", "2x", "4x", "8x") play.Flag("format", defaults.FormatFlagDescription( teleport.PTY, teleport.JSON, teleport.YAML, )).Short('f').Default(teleport.PTY).EnumVar(&cf.Format, teleport.PTY, teleport.JSON, teleport.YAML) @@ -1691,115 +1693,6 @@ func serializeVersion(format string, proxyVersion string, proxyPublicAddress str return string(out), trace.Wrap(err) } -// onPlay is used to interact with recorded sessions. -// It has several modes: -// -// 1. If --format is "pty" (the default), then the recorded -// session is played back in the user's terminal. -// 2. Otherwise, `tsh play` is used to export a session from the -// binary protobuf format into YAML or JSON. -// -// Each of these modes has two subcases: -// a) --session-id ends with ".tar" - tsh operates on a local file -// -// containing a previously downloaded session -// -// b) --session-id is the ID of a session - tsh operates on the session -// -// recording by connecting to the Teleport cluster -func onPlay(cf *CLIConf) error { - if format := strings.ToLower(cf.Format); format == teleport.PTY { - return playSession(cf) - } - return exportSession(cf) -} - -func exportSession(cf *CLIConf) error { - format := strings.ToLower(cf.Format) - isLocalFile := path.Ext(cf.SessionID) == ".tar" - if isLocalFile { - return trace.Wrap(exportFile(cf.Context, cf.SessionID, format)) - } - - tc, err := makeClient(cf) - if err != nil { - return trace.Wrap(err) - } - events, err := tc.GetSessionEvents(cf.Context, cf.Namespace, cf.SessionID) - if err != nil { - return trace.Wrap(err) - } - for _, event := range events { - // when playing from a file, id is not included, this - // makes the outputs otherwise identical - delete(event, "id") - } - - switch format { - case teleport.JSON: - if err := utils.WriteJSONArray(os.Stdout, events); err != nil { - return trace.Wrap(err) - } - case teleport.YAML: - if err := utils.WriteYAML(os.Stdout, events); err != nil { - return trace.Wrap(err) - } - default: - return trace.Errorf("Invalid format %s, only pty, json and yaml are supported", format) - } - - return nil -} - -func playSession(cf *CLIConf) error { - isLocalFile := path.Ext(cf.SessionID) == ".tar" - if isLocalFile { - sid := sessionIDFromPath(cf.SessionID) - tarFile, err := os.Open(cf.SessionID) - if err != nil { - return trace.ConvertSystemError(err) - } - defer tarFile.Close() - if err := client.PlayFile(cf.Context, tarFile, sid); err != nil { - return trace.Wrap(err) - } - return nil - } - tc, err := makeClient(cf) - if err != nil { - return trace.Wrap(err) - } - if err := tc.Play(cf.Context, cf.Namespace, cf.SessionID); err != nil { - if trace.IsNotFound(err) { - log.WithError(err).Debug("error playing session") - return trace.NotFound("Recording for session %s not found.", cf.SessionID) - } - return trace.Wrap(err) - } - return nil -} - -func sessionIDFromPath(path string) string { - fileName := filepath.Base(path) - return strings.TrimSuffix(fileName, ".tar") -} - -// exportFile converts the binary protobuf events from the file -// identified by path to text (JSON/YAML) and writes the converted -// events to standard out. -func exportFile(ctx context.Context, path string, format string) error { - f, err := os.Open(path) - if err != nil { - return trace.ConvertSystemError(err) - } - defer f.Close() - err = events.Export(ctx, f, os.Stdout, format) - if err != nil { - return trace.Wrap(err) - } - return nil -} - // onLogin logs in with remote proxy and gets signed certificates func onLogin(cf *CLIConf) error { autoRequest := true