Add session_type and format to session.recording.access audit e…

…vent event (#47309)

* feat: add `session_type` and `format` to session recording access event

* chore(metadata): change conditional

api/metadata/metadata.go

@@ -28,7 +28,8 @@ import (
 )
 
 const (
-	VersionKey = "version"
+	VersionKey                       = "version"
+	SessionRecordingFormatContextKey = "session-recording-format"
 )
 
 // defaultMetadata returns the default metadata which will be added to all outgoing calls.
@@ -133,3 +134,20 @@ func UserAgentFromContext(ctx context.Context) string {
 	}
 	return strings.Join(values, " ")
 }
+
+// WithSessionRecordingFormatContext returns a context.Context containing the
+// format of the accessed session recording.
+func WithSessionRecordingFormatContext(ctx context.Context, format string) context.Context {
+	return metadata.AppendToOutgoingContext(ctx, SessionRecordingFormatContextKey, format)
+}
+
+// SessionRecordingFormatFromContext returns the format of the accessed session
+// recording (if present).
+func SessionRecordingFormatFromContext(ctx context.Context) string {
+	values := metadata.ValueFromIncomingContext(ctx, SessionRecordingFormatContextKey)
+	if len(values) == 0 {
+		return ""
+	}
+
+	return values[0]
+}

api/proto/teleport/legacy/types/events/events.proto

@@ -5666,6 +5666,10 @@ message SessionRecordingAccess {
     (gogoproto.embed) = true,
     (gogoproto.jsontag) = ""
   ];
+  // SessionType is type of the session.
+  string SessionType = 4 [(gogoproto.jsontag) = "session_type,omitempty"];
+  // Format is the format the session recording was accessed.
+  string Format = 5 [(gogoproto.jsontag) = "format,omitempty"];
 }
 
 // KubeClusterMetadata contains common kubernetes cluster information.

lib/auth/auth_with_roles.go

@@ -42,6 +42,7 @@ import (
 	auditlogpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/auditlog/v1"
 	mfav1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/mfa/v1"
 	"github.com/gravitational/teleport/api/internalutils/stream"
+	"github.com/gravitational/teleport/api/metadata"
 	"github.com/gravitational/teleport/api/types"
 	apievents "github.com/gravitational/teleport/api/types/events"
 	"github.com/gravitational/teleport/api/types/wrappers"
@@ -192,15 +193,32 @@ func (a *ServerWithRoles) actionWithExtendedContext(namespace, kind, verb string
 }
 
 // actionForKindSession is a special checker that grants access to session
-// recordings.  It can allow access to a specific recording based on the
+// recordings. It can allow access to a specific recording based on the
 // `where` section of the user's access rule for kind `session`.
-func (a *ServerWithRoles) actionForKindSession(namespace string, sid session.ID) error {
+func (a *ServerWithRoles) actionForKindSession(namespace string, sid session.ID) (types.SessionKind, error) {
+	sessionEnd, err := a.findSessionEndEvent(namespace, sid)
+
 	extendContext := func(ctx *services.Context) error {
-		sessionEnd, err := a.findSessionEndEvent(namespace, sid)
 		ctx.Session = sessionEnd
 		return trace.Wrap(err)
 	}
-	return trace.Wrap(a.actionWithExtendedContext(namespace, types.KindSession, types.VerbRead, extendContext))
+
+	var sessionKind types.SessionKind
+	switch e := sessionEnd.(type) {
+	case *apievents.SessionEnd:
+		sessionKind = types.SSHSessionKind
+		if e.KubernetesCluster != "" {
+			sessionKind = types.KubernetesSessionKind
+		}
+	case *apievents.DatabaseSessionEnd:
+		sessionKind = types.DatabaseSessionKind
+	case *apievents.AppSessionEnd:
+		sessionKind = types.AppSessionKind
+	case *apievents.WindowsDesktopSessionEnd:
+		sessionKind = types.WindowsDesktopSessionKind
+	}
+
+	return sessionKind, trace.Wrap(a.actionWithExtendedContext(namespace, types.KindSession, types.VerbRead, extendContext))
 }
 
 // localServerAction returns an access denied error if the role is not one of the builtin server roles.
@@ -5933,8 +5951,11 @@ func (a *ServerWithRoles) StreamSessionEvents(ctx context.Context, sessionID ses
 	err := a.localServerAction()
 	isTeleportServer := err == nil
 
+	var sessionType types.SessionKind
 	if !isTeleportServer {
-		if err := a.actionForKindSession(apidefaults.Namespace, sessionID); err != nil {
+		var err error
+		sessionType, err = a.actionForKindSession(apidefaults.Namespace, sessionID)
+		if err != nil {
 			c, e := make(chan apievents.AuditEvent), make(chan error, 1)
 			e <- trace.Wrap(err)
 			return c, e
@@ -5951,6 +5972,8 @@ func (a *ServerWithRoles) StreamSessionEvents(ctx context.Context, sessionID ses
 			},
 			SessionID:    sessionID.String(),
 			UserMetadata: a.context.Identity.GetIdentity().GetUserMetadata(),
+			SessionType:  string(sessionType),
+			Format:       metadata.SessionRecordingFormatFromContext(ctx),
 		}); err != nil {
 			return createErrorChannel(err)
 		}

lib/auth/auth_with_roles_test.go

@@ -52,6 +52,7 @@ import (
 	mfav1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/mfa/v1"
 	trustpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/trust/v1"
 	userpreferencesv1 "github.com/gravitational/teleport/api/gen/proto/go/userpreferences/v1"
+	"github.com/gravitational/teleport/api/metadata"
 	"github.com/gravitational/teleport/api/mfa"
 	"github.com/gravitational/teleport/api/types"
 	apievents "github.com/gravitational/teleport/api/types/events"
@@ -2256,6 +2257,60 @@ func TestStreamSessionEvents(t *testing.T) {
 	require.Equal(t, username, event.User)
 }
 
+// TestStreamSessionEvents ensures that when a user streams a session's events
+// a "session recording access" event is emitted with the correct session type.
+func TestStreamSessionEvents_SessionType(t *testing.T) {
+	t.Parallel()
+
+	srv := newTestTLSServer(t)
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+
+	username := "user"
+	user, _, err := CreateUserAndRole(srv.Auth(), username, []string{}, nil)
+	require.NoError(t, err)
+
+	identity := TestUser(user.GetName())
+	clt, err := srv.NewClient(identity)
+	require.NoError(t, err)
+	sessionID := "44c6cea8-362f-11ea-83aa-125400432324"
+
+	// Emitting a session end event will cause the listing to correctly locate
+	// the recording (even if there might not be a recording file to stream).
+	require.NoError(t, srv.Auth().EmitAuditEvent(ctx, &apievents.DatabaseSessionEnd{
+		Metadata: apievents.Metadata{
+			Type: events.DatabaseSessionEndEvent,
+			Code: events.DatabaseSessionEndCode,
+		},
+		SessionMetadata: apievents.SessionMetadata{
+			SessionID: sessionID,
+		},
+	}))
+
+	accessedFormat := teleport.PTY
+	clt.StreamSessionEvents(metadata.WithSessionRecordingFormatContext(ctx, accessedFormat), session.ID(sessionID), 0)
+
+	// Perform the listing an eventually loop to ensure the event is emitted.
+	var searchEvents []apievents.AuditEvent
+	require.EventuallyWithT(t, func(t *assert.CollectT) {
+		var err error
+		searchEvents, _, err = srv.AuthServer.AuditLog.SearchEvents(ctx, events.SearchEventsRequest{
+			From:       srv.Clock().Now().Add(-time.Hour),
+			To:         srv.Clock().Now().Add(time.Hour),
+			EventTypes: []string{events.SessionRecordingAccessEvent},
+			Limit:      1,
+			Order:      types.EventOrderDescending,
+		})
+		assert.NoError(t, err)
+		assert.Len(t, searchEvents, 1, "expected one event but got %d", len(searchEvents))
+	}, 5*time.Second, 200*time.Millisecond)
+
+	event := searchEvents[0].(*apievents.SessionRecordingAccess)
+	require.Equal(t, username, event.User)
+	require.Equal(t, string(types.DatabaseSessionKind), event.SessionType)
+	require.Equal(t, accessedFormat, event.Format)
+}
+
 // TestAPILockedOut tests Auth API when there are locks involved.
 func TestAPILockedOut(t *testing.T) {
 	t.Parallel()

lib/player/player.go

@@ -33,6 +33,7 @@ import (
 	"golang.org/x/exp/maps"
 
 	"github.com/gravitational/teleport"
+	"github.com/gravitational/teleport/api/metadata"
 	"github.com/gravitational/teleport/api/types/events"
 	"github.com/gravitational/teleport/lib/defaults"
 	"github.com/gravitational/teleport/lib/player/db"
@@ -189,7 +190,7 @@ func (p *Player) stream() {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	eventsC, errC := p.streamer.StreamSessionEvents(ctx, p.sessionID, 0)
+	eventsC, errC := p.streamer.StreamSessionEvents(metadata.WithSessionRecordingFormatContext(ctx, teleport.PTY), p.sessionID, 0)
 	var lastDelay time.Duration
 	for {
 		select {

tool/tsh/common/play.go

@@ -32,6 +32,7 @@ import (
 	"github.com/gravitational/trace"
 
 	"github.com/gravitational/teleport"
+	"github.com/gravitational/teleport/api/metadata"
 	apievents "github.com/gravitational/teleport/api/types/events"
 	"github.com/gravitational/teleport/lib/client"
 	"github.com/gravitational/teleport/lib/events"
@@ -137,7 +138,7 @@ func exportSession(cf *CLIConf) error {
 	}
 	defer clusterClient.Close()
 
-	eventC, errC := clusterClient.AuthClient.StreamSessionEvents(cf.Context, *sid, 0)
+	eventC, errC := clusterClient.AuthClient.StreamSessionEvents(metadata.WithSessionRecordingFormatContext(cf.Context, format), *sid, 0)
 
 	var exporter sessionExporter
 	switch format {