Convert tsh play to a streaming API (#34547)

* Convert tsh play to use the new streaming API

Instead of downloading the entire session before starting playback,
tsh now streams the recording directly to the terminal. This results
in a significant reduction in "time to first byte" when playing back
large session recordings.

This has the side effect of making playback of SSH consistent with
that of desktop sessions (we no longer "skip" dead time for SSH
sessions and instead play them back at true speed). Additionally,
a new CLI flag was added to support configurable playback speeds.

Fixes #10579
Closes #11385
Updates gravitational/teleport-private#1024

* Start to phase out the GetSessionEvents API

- Mark the old HTTP client implementations as not implemented.
  Nothing is calling these, but they remain for now until we can
  fully remove them from the ClientI interface.
- Move `tsh play --format=json` over to a streaming approach.
  This ensures that we don't load the entire session into memory
  before converting it to JSON/YAML format.

* Clean up terminal escape codes

- Move more utilities into lib/client.Terminal
- Stop timestamping the frame by moving the cursor and writing
  to the terminal - this is racy. Write the timestamp to the
  terminal title instead.

* Update changelog

* Move terminal functions from Unix to common build

This is valid because we initialize terminals on Windows with
ENABLE_VIRTUAL_TERMINAL_PROCESSING, so the same escape sequences
that work on Unix will also work on Windows.

This matches Microsoft's recommendation to use virtual terminal
sequences over the Windows Console API [1].

For more info, see: https://learn.microsoft.com/en-us/windows/console/console-virtual-terminal-sequences

[1] https://learn.microsoft.com/en-us/windows/console/classic-vs-vt
This commit is contained in:
Zac Bergquist
2024-01-02 19:57:12 +00:00
committed by GitHub
parent d34368483f
commit 25ee7b1343
13 changed files with 458 additions and 830 deletions
+11
View File
@@ -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
+3 -5
View File
@@ -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 <Var name="teleport.example.com" />
with your Proxy Service address (e.g. `mytenant.teleport.sh` for Teleport
Enterprise Cloud):
```code
$ curl https://<Var name="teleport.example.com" />/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
+7 -4
View File
@@ -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),
+6 -5
View File
@@ -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)
}
+117 -181
View File
@@ -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 {
+29 -265
View File
@@ -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
}
-247
View File
@@ -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 <http://www.gnu.org/licenses/>.
*/
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
}
+14
View File
@@ -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) -
+17 -10
View File
@@ -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 {
-1
View File
@@ -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 == "" {
+1 -1
View File
@@ -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
+249
View File
@@ -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
}
+4 -111
View File
@@ -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