From fb72fa8805c761e2ef761b95b26aa3099464d185 Mon Sep 17 00:00:00 2001 From: Tiago Silva Date: Mon, 3 Apr 2023 11:14:15 +0100 Subject: [PATCH] Fixes unsused flags from `tsh kube exec` command (#23836) * Fixes unsused flags from `tsh kube exec` command Since its introduction, `tsh kube exec` didn't have any use for the `--reason` and `--invite` flags and completely ignored what users did with it. This PR transports those flags as query parameters to Teleport Kubernetes Service which includes them when creating the session resource in Teleport. * add tests * fix flaky test --- constants.go | 13 +++++ lib/kube/proxy/exec_test.go | 57 ++++++++++++++------ lib/kube/proxy/moderated_sessions_test.go | 66 +++++++++++++++++++---- lib/kube/proxy/sess.go | 11 +++- tool/tsh/kube.go | 10 +++- 5 files changed, 129 insertions(+), 28 deletions(-) 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)