diff --git a/constants.go b/constants.go index a4b8659c5d7..4e79e5636a0 100644 --- a/constants.go +++ b/constants.go @@ -814,3 +814,16 @@ const ( // the upgraded connection should be handled by the ALPN handler. WebAPIConnUpgradeTypeALPN = "alpn" ) + +const ( + // KubeSessionDisplayParticipantRequirementsQueryParam is the query parameter used to + // indicate that the client wants to display the participant requirements + // for the given session. + KubeSessionDisplayParticipantRequirementsQueryParam = "displayParticipantRequirements" + // KubeSessionReasonQueryParam is the query parameter used to indicate the reason + // for the session request. + KubeSessionReasonQueryParam = "reason" + // KubeSessionInvitedQueryParam is the query parameter used to indicate the users + // to invite to the session. + KubeSessionInvitedQueryParam = "invite" +) diff --git a/lib/kube/proxy/exec_test.go b/lib/kube/proxy/exec_test.go index 5c34494ace3..844a2d9746f 100644 --- a/lib/kube/proxy/exec_test.go +++ b/lib/kube/proxy/exec_test.go @@ -23,6 +23,7 @@ import ( "io" "net/http" "net/url" + "strings" "testing" "github.com/stretchr/testify/require" @@ -32,6 +33,7 @@ import ( "k8s.io/client-go/tools/remotecommand" "k8s.io/kubectl/pkg/scheme" + "github.com/gravitational/teleport" testingkubemock "github.com/gravitational/teleport/lib/kube/proxy/testing/kube_server" ) @@ -182,12 +184,14 @@ func TestExecKubeService(t *testing.T) { } req, err := generateExecRequest( - testCtx.KubeServiceAddress(), - podName, - podNamespace, - podContainerName, - containerCommmandExecute, // placeholder for commands to execute in the dummy pod - streamOpts, + generateExecRequestConfig{ + addr: testCtx.KubeServiceAddress(), + podName: podName, + podNamespace: podNamespace, + containerName: podContainerName, + cmd: containerCommmandExecute, // placeholder for commands to execute in the dummy pod + options: streamOpts, + }, ) require.NoError(t, err) // configure the client to impersonate the user. @@ -208,12 +212,31 @@ func TestExecKubeService(t *testing.T) { } } +type generateExecRequestConfig struct { + // addr is the address of the Kube API server. + addr string + // podName is the name of the pod to execute the command in. + podName string + // podNamespace is the namespace of the pod to execute the command in. + podNamespace string + // containerName is the name of the container to execute the command in. + containerName string + // cmd is the command to execute in the container. + cmd []string + // options are the options for the command execution. + options remotecommand.StreamOptions + // reason is the reason for the command execution. + reason string + // invite is the list of users to invite. + invite []string +} + // generateExecRequest generates a Kube API url for executing commands in pods. // The url format is the following: -// "/api/v1/namespaces/{podNamespace}/pods/{podName}/exec?stderr={stdout}&stdout={stdout}&tty={tty}. -func generateExecRequest(addr, podName, podNamespace, containerName string, cmd []string, options remotecommand.StreamOptions) (*rest.Request, error) { +// "/api/v1/namespaces/{podNamespace}/pods/{podName}/exec?stderr={stdout}&stdout={stdout}&tty={tty}&reason={reason}&container={containerName}&command={command}" +func generateExecRequest(cfg generateExecRequestConfig) (*rest.Request, error) { restClient, err := rest.RESTClientFor(&rest.Config{ - Host: addr, + Host: cfg.addr, APIPath: "/api", ContentConfig: rest.ContentConfig{ GroupVersion: &corev1.SchemeGroupVersion, @@ -231,13 +254,15 @@ func generateExecRequest(addr, podName, podNamespace, containerName string, cmd Namespace(podNamespace). SubResource("exec"). VersionedParams(&corev1.PodExecOptions{ - Container: containerName, - Command: cmd, - Stdin: options.Stdin != nil, - Stdout: options.Stdout != nil, - Stderr: options.Stderr != nil, - TTY: options.Tty, - }, scheme.ParameterCodec) + Container: cfg.containerName, + Command: cfg.cmd, + Stdin: cfg.options.Stdin != nil, + Stdout: cfg.options.Stdout != nil, + Stderr: cfg.options.Stderr != nil, + TTY: cfg.options.Tty, + }, scheme.ParameterCodec). + Param(teleport.KubeSessionInvitedQueryParam, strings.Join(cfg.invite, ",")). + Param(teleport.KubeSessionReasonQueryParam, cfg.reason) return req, nil } diff --git a/lib/kube/proxy/moderated_sessions_test.go b/lib/kube/proxy/moderated_sessions_test.go index 534f4f23650..49603000d7e 100644 --- a/lib/kube/proxy/moderated_sessions_test.go +++ b/lib/kube/proxy/moderated_sessions_test.go @@ -22,6 +22,7 @@ import ( "fmt" "io" "net/http" + "reflect" "regexp" "strings" "sync" @@ -158,6 +159,8 @@ func TestModeratedSessions(t *testing.T) { moderator types.User closeSession bool moderatorForcedClose bool + reason string + invite []string } type want struct { sessionEndEvent bool @@ -170,7 +173,9 @@ func TestModeratedSessions(t *testing.T) { { name: "create session for user without moderation", args: args{ - user: user, + user: user, + reason: "reason 1", + invite: []string{"user1", "user2"}, }, want: want{ sessionEndEvent: true, @@ -181,6 +186,8 @@ func TestModeratedSessions(t *testing.T) { args: args{ user: userRequiringModerator, moderator: moderator, + reason: "reason 2", + invite: []string{"user1", "user2"}, }, want: want{ sessionEndEvent: true, @@ -191,6 +198,8 @@ func TestModeratedSessions(t *testing.T) { args: args{ user: user, closeSession: true, + reason: "reason 3", + invite: []string{"user1", "user2"}, }, want: want{ sessionEndEvent: true, @@ -201,6 +210,8 @@ func TestModeratedSessions(t *testing.T) { args: args{ user: userRequiringModerator, closeSession: true, + reason: "reason 4", + invite: []string{"user1", "user2"}, }, want: want{ // until moderator joins the session is not started. If the connection @@ -214,6 +225,8 @@ func TestModeratedSessions(t *testing.T) { user: userRequiringModerator, moderator: moderator, moderatorForcedClose: true, + reason: "reason 5", + invite: []string{"user1", "user2"}, }, want: want{ sessionEndEvent: true, @@ -249,12 +262,16 @@ func TestModeratedSessions(t *testing.T) { Tty: true, } req, err := generateExecRequest( - testCtx.KubeServiceAddress(), - podName, - podNamespace, - podContainerName, - containerCommmandExecute, // placeholder for commands to execute in the dummy pod - streamOpts, + generateExecRequestConfig{ + addr: testCtx.KubeServiceAddress(), + podName: podName, + podNamespace: podNamespace, + containerName: podContainerName, + cmd: containerCommmandExecute, // placeholder for commands to execute in the dummy pod + options: streamOpts, + reason: tt.args.reason, + invite: tt.args.invite, + }, ) require.NoError(t, err) @@ -279,6 +296,10 @@ func TestModeratedSessions(t *testing.T) { group.Go(func() error { // waits for user to send the sessionID of his exec request. sessionID := <-sessionIDC + // validate that the sessionID is valid and the reason is the one we expect. + if err := validateSessionTracker(testCtx, sessionID, tt.args.reason, tt.args.invite); err != nil { + return trace.Wrap(err) + } t.Logf("moderator is joining sessionID %q", sessionID) // join the session. stream, err := testCtx.NewJoiningSession(config, sessionID, types.SessionModeratorMode) @@ -295,10 +316,13 @@ func TestModeratedSessions(t *testing.T) { // moderator waits for the user informed that he joined the session. <-moderatorJoined + dataFound := false for { p := make([]byte, 1024) n, err := stream.Read(p) - if err != nil { + if errors.Is(err, io.EOF) { + break + } else if err != nil { return trace.Wrap(err) } stringData := string(p[:n]) @@ -311,16 +335,22 @@ func TestModeratedSessions(t *testing.T) { // stdinPayload is sent by the user after the session started. if strings.Contains(stringData, stdinPayload) { - break + dataFound = true } // podContainerName is returned by the kubemock server and it's used // to control that the session has effectively started. // return to force the defer to run. if strings.Contains(stringData, podContainerName) && tt.args.moderatorForcedClose { - return nil + if err := stream.ForceTerminate(); err != nil { + return trace.Wrap(err) + } + continue } } + if !dataFound && !tt.args.moderatorForcedClose { + return trace.Wrap(errors.New("stdinPayload was not received")) + } return nil }) } @@ -429,3 +459,19 @@ func TestModeratedSessions(t *testing.T) { }) } } + +// validateSessionTracker validates that the session tracker has the expected +// reason and invited users. +func validateSessionTracker(testCtx *TestContext, sessionID string, reason string, invited []string) error { + sessionTracker, err := testCtx.AuthClient.GetSessionTracker(testCtx.Context, sessionID) + if err != nil { + return trace.Wrap(err) + } + if sessionTracker.GetReason() != reason { + return trace.BadParameter("expected reason %q, got %q", reason, sessionTracker.GetReason()) + } + if !reflect.DeepEqual(sessionTracker.GetInvited(), invited) { + return trace.BadParameter("expected invited %q, got %q", invited, sessionTracker.GetInvited()) + } + return nil +} diff --git a/lib/kube/proxy/sess.go b/lib/kube/proxy/sess.go index 5c62525b3d8..273747a50e3 100644 --- a/lib/kube/proxy/sess.go +++ b/lib/kube/proxy/sess.go @@ -352,6 +352,11 @@ type session struct { // Set if we should broadcast information about participant requirements to the session. displayParticipantRequirements bool + // invitedUsers is a list of users that were invited to the session. + invitedUsers []string + // reason is the reason for the session. + reason string + // eventsWaiter is used to wait for events to be emitted and goroutines closed // when a session is closed. eventsWaiter sync.WaitGroup @@ -405,7 +410,9 @@ func newSession(ctx authContext, forwarder *Forwarder, req *http.Request, params initiator: initiator.ID, expires: time.Now().UTC().Add(sessionMaxLifetime), PresenceEnabled: ctx.Identity.GetIdentity().MFAVerified != "", - displayParticipantRequirements: utils.AsBool(q.Get("displayParticipantRequirements")), + displayParticipantRequirements: utils.AsBool(q.Get(teleport.KubeSessionDisplayParticipantRequirementsQueryParam)), + invitedUsers: strings.Split(q.Get(teleport.KubeSessionInvitedQueryParam), ","), + reason: q.Get(teleport.KubeSessionReasonQueryParam), streamContext: streamContext, streamContextCancel: streamContextCancel, partiesWg: sync.WaitGroup{}, @@ -1190,6 +1197,8 @@ func (s *session) trackSession(p *party, policySet []*types.SessionTrackerPolicy HostPolicies: policySet, Login: "root", Created: time.Now(), + Reason: s.reason, + Invited: s.invitedUsers, } s.log.Debug("Creating session tracker") diff --git a/tool/tsh/kube.go b/tool/tsh/kube.go index be061011478..f21ef5e548c 100644 --- a/tool/tsh/kube.go +++ b/tool/tsh/kube.go @@ -313,6 +313,10 @@ type ExecOptions struct { GetPodTimeout time.Duration Config *restclient.Config displayParticipantRequirements bool + // invited is a list of users that are invited to the session + invited []string + // reason is the reason for the session + reason string } // Run executes a validated remote execution against a pod. @@ -382,7 +386,9 @@ func (p *ExecOptions) Run(ctx context.Context) error { Name(pod.Name). Namespace(pod.Namespace). SubResource("exec"). - Param("displayParticipantRequirements", strconv.FormatBool(p.displayParticipantRequirements)) + Param(teleport.KubeSessionDisplayParticipantRequirementsQueryParam, strconv.FormatBool(p.displayParticipantRequirements)). + Param(teleport.KubeSessionInvitedQueryParam, strings.Join(p.invited, ",")). + Param(teleport.KubeSessionReasonQueryParam, p.reason) req.VersionedParams(&corev1.PodExecOptions{ Container: containerName, Command: p.Command, @@ -454,6 +460,8 @@ func (c *kubeExecCommand) run(cf *CLIConf) error { p.restClientGetter = f p.Executor = &DefaultRemoteExecutor{} p.displayParticipantRequirements = c.displayParticipantRequirements + p.invited = strings.Split(c.invited, ",") + p.reason = c.reason p.Namespace, p.EnforceNamespace, err = f.ToRawKubeConfigLoader().Namespace() if err != nil { return trace.Wrap(err)