gke:Autopilot: Transform system:masters usage error to be user-friendly (#29633)

GKE Autopilot clusters deny impersonating `system:masters` kubernetes
group due to security reasons ( https://cloud.google.com/kubernetes-engine/docs/concepts/autopilot-security#built-in-security ).

Since it is the only product that enforces this and most administrators impersonate `system:masters`, this limitation has created too many difficulties in being understood.
First because the error is not clear enough and second because it doesn't explain what's the real problem and what part of the Teleport fails.

This PR tries to solve the problem by changing the original error message to a more user-friendly, one that better explains the problem and its origin.

Example of response body that is rewritten:

> Error from server (Forbidden): groups "system:masters" is forbidden:
> User "<user>" cannot impersonate resource "groups" in API group "" at the cluster
> scope: GKE Warden authz [denied by user-impersonation-limitation]: impersonating
> system identities are not allowed

The rewritten response body will look like:

> Error from server (Forbidden): "GKE Autopilot denied the request because it impersonates the "system:masters" group.
> Your Teleport Roles [role1,role2] have given access to the "system:masters" group for the cluster "<cluster>".
> For additional information and resolution, please visit https://goteleport.com/docs/kubernetes-access/troubleshooting/#unable-to-connect-to-gke-autopilot-clusters

Fixes #28506

lib/httplib/reverseproxy/reverse_proxy.go

@@ -167,6 +167,13 @@ func WithPassHostHeader() Option {
 	}
 }
 
+// WithResponseModifier sets the response modifier for the forwarder.
+func WithResponseModifier(m func(*http.Response) error) Option {
+	return func(rp *Forwarder) {
+		rp.ModifyResponse = m
+	}
+}
+
 // Modify the request to handle the target URL.
 func modifyRequest(outReq *http.Request) {
 	u := getURLFromRequest(outReq)

lib/kube/proxy/forwarder.go

@@ -2316,12 +2316,26 @@ func (f *Forwarder) makeSessionForwarder(sess *clusterSession) (*reverseproxy.Fo
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
-
-	forwarder, err := reverseproxy.New(
-		reverseproxy.WithFlushInterval(100*time.Millisecond),
+	opts := []reverseproxy.Option{
+		reverseproxy.WithFlushInterval(100 * time.Millisecond),
 		reverseproxy.WithRoundTripper(transport),
 		reverseproxy.WithLogger(f.log),
 		reverseproxy.WithErrorHandler(f.formatForwardResponseError),
+	}
+	if f.isLocalKubeCluster(sess.teleportCluster.isRemote, sess.kubeClusterName) {
+		// If the target cluster is local, i.e. the cluster that is served by this
+		// teleport service, then we set up the forwarder to allow re-writing
+		// the response to the client to include user friendly error messages.
+		// This is done by adding a response modifier to the forwarder.
+		// Right now, the only error that is re-written is the 403 Forbidden error
+		// that is returned when the user tries to access a GKE Autopilot cluster
+		// with system:masters group impersonation.
+		//nolint:bodyclose // the caller closes the response body in httputils.ReverseProxy
+		opts = append(opts, reverseproxy.WithResponseModifier(f.rewriteResponseForbidden(sess)))
+	}
+
+	forwarder, err := reverseproxy.New(
+		opts...,
 	)
 
 	return forwarder, trace.Wrap(err)

lib/kube/proxy/response_rewriter.go

@@ -0,0 +1,173 @@
+/*
+Copyright 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 proxy
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"net/http"
+	"strconv"
+
+	"github.com/gravitational/trace"
+	"golang.org/x/exp/slices"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"github.com/gravitational/teleport/api/types"
+	"github.com/gravitational/teleport/lib/httplib/reverseproxy"
+	"github.com/gravitational/teleport/lib/kube/proxy/responsewriters"
+	"github.com/gravitational/teleport/lib/services"
+)
+
+// rewriteResponseForbidden rewrites the response body when the response includes
+// a GKE Autopilot forbidden error caused by impersonating system:masters group.
+// The response body is rewritten to include a more user friendly error message.
+// All other responses are returned as is.
+// Example of response body that is rewritten:
+//
+// Error from server (Forbidden): groups "system:masters" is forbidden:
+// User "<user>" cannot impersonate resource "groups" in API group "" at the cluster
+// scope: GKE Warden authz [denied by user-impersonation-limitation]: impersonating
+// system identities are not allowed
+//
+// The rewritten response body will look like:
+//
+// Error from server (Forbidden): "GKE Autopilot denied the request because it impersonates the "system:masters" group.
+// Your Teleport Roles [role1,role2] have given access to the "system:masters" group for the cluster "<cluster>".
+// For additional information and resolution, please visit https://goteleport.com/docs/kubernetes-access/troubleshooting/#unable-to-connect-to-gke-autopilot-clusters
+func (f *Forwarder) rewriteResponseForbidden(s *clusterSession) func(r *http.Response) error {
+	return func(r *http.Response) error {
+		const (
+			// The string that is returned by the GKE Autopilot cluster when
+			// users try to impersonate system:masters group.
+			autopilotForbidden = "impersonating system identities are not allowed"
+		)
+		// If the response is not forbidden, we don't need to do anything.
+		// The response will be returned as is and written to the client.
+		if r.StatusCode != http.StatusForbidden || r.Body == nil {
+			return nil
+		}
+		// create a new buffer to read the response body into.
+		b := bytes.NewBuffer(make([]byte, 0, 4096))
+
+		// Read the response body into the buffer.
+		if _, err := io.Copy(b, r.Body); err != nil {
+			return trace.Wrap(err)
+		}
+		// Close the response body.
+		if err := r.Body.Close(); err != nil {
+			return trace.Wrap(err)
+		}
+
+		// Replace the response body with the new buffer.
+		r.Body = io.NopCloser(b)
+
+		switch {
+		case bytes.Contains(b.Bytes(), []byte(autopilotForbidden)):
+			// If the response body contains the forbidden string, we rewrite the
+			// response body to include a more user friendly error message.
+			encoder, _, err := newEncoderAndDecoderForContentType(
+				r.Header.Get(responsewriters.ContentTypeHeader),
+				newClientNegotiator(&globalKubeCodecs),
+			)
+			if err != nil {
+				f.log.WithError(err).Error("Failed to create encoder")
+				return nil
+			}
+
+			status := &metav1.Status{
+				Status: metav1.StatusFailure,
+				Code:   int32(http.StatusForbidden),
+				Reason: metav1.StatusReasonForbidden,
+				Message: "GKE Autopilot denied the request because it impersonates the \"system:masters\" group.\n" +
+					fmt.Sprintf(
+						"Your Teleport Roles %v have given access to the \"system:masters\" group "+
+							"for the cluster %q.\n", collectSystemMastersTeleportRoles(s), s.kubeClusterName) +
+					"For additional information and resolution, " +
+					"please visit https://goteleport.com/docs/kubernetes-access/troubleshooting/#unable-to-connect-to-gke-autopilot-clusters\n",
+			}
+			// Reset the buffer to write the new response.
+			b.Reset()
+
+			// Encode the new response.
+			if err = encoder.Encode(status, b); err != nil {
+				f.log.WithError(err).Error("Failed to encode response")
+				return trace.Wrap(err)
+			}
+
+			// This function rewrote the response body, so we need update delete the
+			// Content-Length header to avoid mismatch between the actual body
+			// length and the original Content-Length header value.
+			r.Header.Set(reverseproxy.ContentLength, strconv.Itoa(b.Len()))
+			return nil
+		}
+
+		return nil
+	}
+}
+
+// collectSystemMastersTeleportRoles returns a list of teleport roles that grant
+// system:masters to the target cluster.
+func collectSystemMastersTeleportRoles(s *clusterSession) []string {
+	const (
+		systemMastersGroup = "system:masters"
+	)
+	accessChecker := s.authContext.Checker
+	matchers := make([]services.RoleMatcher, 0, 3)
+	// Creates a matcher that matches the cluster labels against `kubernetes_labels`
+	// defined for each user's role.
+	matchers = append(
+		matchers,
+		services.NewKubernetesClusterLabelMatcher(s.kubeClusterLabels, accessChecker.Traits()),
+	)
+
+	// If the kubeResource is available, append an extra matcher that validates
+	// if the kubernetes resource is allowed by the user roles that satisfy the
+	// target cluster labels.
+	// Each role defines `kubernetes_resources` and when kubeResource is available,
+	// KubernetesResourceMatcher will match roles that statisfy the resources at the
+	// same time that ClusterLabelMatcher matches the role's "kubernetes_labels".
+	// The call to roles.CheckKubeGroupsAndUsers when both matchers are provided
+	// results in the intersection of roles that match the "kubernetes_labels" and
+	// roles that allow access to the desired "kubernetes_resource".
+	// If from the intersection results an empty set, the request is denied.
+	if s.kubeResource != nil {
+		matchers = append(
+			matchers,
+			services.NewKubernetesResourceMatcher(*s.kubeResource),
+		)
+	}
+	var rolesWithSystemMasters []string
+	matchers = append(matchers,
+		// Creates a matcher that checks if the role grants system:masters group.
+		// The matcher will be called for each role that matches the cluster labels
+		// and the kubernetes resource (if available).
+		// It's important to note that this matcher must be the last one in the list
+		// otherwise the returned roles may not match the cluster labels and the
+		// kubernetes resource.
+		services.RoleMatcherFunc(func(r types.Role, cond types.RoleConditionType) (bool, error) {
+			groups := r.GetKubeGroups(cond)
+			if slices.Contains(groups, systemMastersGroup) {
+				rolesWithSystemMasters = append(rolesWithSystemMasters, r.GetName())
+			}
+			return true, nil
+		}),
+	)
+
+	_, _, _ = accessChecker.CheckKubeGroupsAndUsers(s.sessionTTL, false /* overrideTTL */, matchers...)
+	return rolesWithSystemMasters
+}

lib/kube/proxy/response_rewriter_test.go

@@ -0,0 +1,167 @@
+/*
+Copyright 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 proxy
+
+import (
+	"context"
+	"net/http"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	testingkubemock "github.com/gravitational/teleport/lib/kube/proxy/testing/kube_server"
+)
+
+func TestErrorRewriter(t *testing.T) {
+	t.Parallel()
+	const (
+		gkeAutopilotCluster = "gke-autopilot"
+		otherCluster        = "any-cluster"
+		username            = "user"
+	)
+	// kubeMock is a Kubernetes API mock for the session tests.
+	// Once a new session is created, this mock will write to
+	// stdout and stdin (if available) the pod name, followed
+	// by copying the contents of stdin into both streams.
+	gkeKubeMock, err := testingkubemock.NewKubeAPIMock(
+		testingkubemock.WithGetPodError(
+			metav1.Status{
+				Status: metav1.StatusFailure,
+				Message: "groups \"system:masters\" is forbidden: User \"<user>\" cannot " +
+					"impersonate resource \"groups\" in API group \"\" at the cluster scope: GKE " +
+					"Warden authz [denied by user-impersonation-limitation]: impersonating system " +
+					"identities are not allowed",
+				Reason: metav1.StatusReasonForbidden,
+				Code:   http.StatusForbidden,
+			},
+		),
+	)
+	require.NoError(t, err)
+	t.Cleanup(func() { gkeKubeMock.Close() })
+
+	otherKubeMock, err := testingkubemock.NewKubeAPIMock(
+		testingkubemock.WithGetPodError(
+			metav1.Status{
+				Status:  metav1.StatusFailure,
+				Message: "request denied",
+				Reason:  metav1.StatusReasonForbidden,
+				Code:    http.StatusForbidden,
+			},
+		),
+	)
+	require.NoError(t, err)
+	t.Cleanup(func() { otherKubeMock.Close() })
+
+	// creates a Kubernetes service with a configured cluster pointing to mock api server
+	testCtx := SetupTestContext(
+		context.Background(),
+		t,
+		TestConfig{
+			Clusters: []KubeClusterConfig{
+				{Name: gkeAutopilotCluster, APIEndpoint: gkeKubeMock.URL},
+				{Name: otherCluster, APIEndpoint: otherKubeMock.URL},
+			},
+		},
+	)
+	// close tests
+	t.Cleanup(func() { require.NoError(t, testCtx.Close()) })
+
+	// create a user with full access to kubernetes Pods.
+	// (kubernetes_user and kubernetes_groups specified)
+	user, _ := testCtx.CreateUserAndRole(
+		testCtx.Context,
+		t,
+		username,
+		RoleSpec{
+			Name:       username,
+			KubeUsers:  roleKubeUsers,
+			KubeGroups: []string{"system:masters"},
+		},
+	)
+
+	type args struct {
+		kubeCluster string
+	}
+	type want struct {
+		getTestPodResult error
+	}
+	tests := []struct {
+		name string
+		args args
+		want want
+	}{
+		{
+			name: "rewrite gke autopilot error",
+			args: args{
+				kubeCluster: gkeAutopilotCluster,
+			},
+			want: want{
+				getTestPodResult: &errors.StatusError{
+					ErrStatus: metav1.Status{
+						Status: metav1.StatusFailure,
+						Message: "GKE Autopilot denied the request because it impersonates the " +
+							"\"system:masters\" group.\nYour Teleport Roles [user:user] have " +
+							"given access to the \"system:masters\" group for the cluster " +
+							"\"gke-autopilot\".\nFor additional information and resolution, " +
+							"please visit https://goteleport.com/docs/kubernetes-access/troubleshooting/#unable-to-connect-to-gke-autopilot-clusters\n",
+						Reason: metav1.StatusReasonForbidden,
+						Code:   http.StatusForbidden,
+					},
+				},
+			},
+		},
+		{
+			name: "don't rewrite other errors",
+			args: args{
+				kubeCluster: otherCluster,
+			},
+			want: want{
+				getTestPodResult: &errors.StatusError{
+					ErrStatus: metav1.Status{
+						Status:  metav1.StatusFailure,
+						Message: "request denied",
+						Reason:  metav1.StatusReasonForbidden,
+						Code:    http.StatusForbidden,
+					},
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			// generate a kube client with user certs for auth
+			client, _ := testCtx.GenTestKubeClientTLSCert(
+				t,
+				user.GetName(),
+				tt.args.kubeCluster,
+			)
+
+			_, err := client.CoreV1().Pods(metav1.NamespaceDefault).Get(
+				testCtx.Context,
+				"test-pod",
+				metav1.GetOptions{},
+			)
+			require.Error(t, err)
+			require.Equal(t, tt.want.getTestPodResult, err)
+		})
+	}
+}

lib/kube/proxy/testing/kube_server/kube_mock.go

@@ -89,6 +89,16 @@ const (
 	PortForwardPayload = "Portforward handler message"
 )
 
+// Option is a functional option for KubeMockServer
+type Option func(*KubeMockServer)
+
+// WithGetPodError sets the error to be returned by the GetPod call
+func WithGetPodError(status metav1.Status) Option {
+	return func(s *KubeMockServer) {
+		s.getPodError = &status
+	}
+}
+
 type deletedResource struct {
 	requestID string
 	kind      string
@@ -102,6 +112,7 @@ type KubeMockServer struct {
 	Address          string
 	CA               []byte
 	deletedResources map[deletedResource][]string
+	getPodError      *metav1.Status
 	mu               sync.Mutex
 }
 
@@ -112,12 +123,17 @@ type KubeMockServer struct {
 // The output returns the container followed by a dump of the data received from stdin.
 // More endpoints can be configured
 // TODO(tigrato): add support for other endpoints
-func NewKubeAPIMock() (*KubeMockServer, error) {
+func NewKubeAPIMock(opts ...Option) (*KubeMockServer, error) {
 	s := &KubeMockServer{
 		router:           httprouter.New(),
 		log:              log.NewEntry(log.New()),
 		deletedResources: make(map[deletedResource][]string),
 	}
+
+	for _, o := range opts {
+		o(s)
+	}
+
 	s.setup()
 	if err := http2.ConfigureServer(s.server.Config, &http2.Server{}); err != nil {
 		return nil, err
@@ -183,6 +199,10 @@ func (s *KubeMockServer) formatResponseError(rw http.ResponseWriter, respErr err
 		Message: respErr.Error(),
 		Code:    int32(trace.ErrorToCode(respErr)),
 	}
+	s.writeResponseError(rw, respErr, status)
+}
+
+func (s *KubeMockServer) writeResponseError(rw http.ResponseWriter, respErr error, status *metav1.Status) {
 	data, err := runtime.Encode(kubeCodecs.LegacyCodec(), status)
 	if err != nil {
 		s.log.Warningf("Failed encoding error into kube Status object: %v", err)
@@ -193,7 +213,7 @@ func (s *KubeMockServer) formatResponseError(rw http.ResponseWriter, respErr err
 	// Always write InternalServerError, that's the only code that kubectl will
 	// parse the Status object for. The Status object has the real status code
 	// embedded.
-	rw.WriteHeader(http.StatusInternalServerError)
+	rw.WriteHeader(int(status.Code))
 	if _, err := rw.Write(data); err != nil {
 		s.log.Warningf("Failed writing kube error response body: %v", err)
 	}

lib/kube/proxy/testing/kube_server/pods.go

@@ -87,6 +87,10 @@ func (s *KubeMockServer) listPods(w http.ResponseWriter, req *http.Request, p ht
 }
 
 func (s *KubeMockServer) getPod(w http.ResponseWriter, req *http.Request, p httprouter.Params) (any, error) {
+	if s.getPodError != nil {
+		s.writeResponseError(w, nil, s.getPodError)
+		return nil, nil
+	}
 	namespace := p.ByName("namespace")
 	name := p.ByName("name")
 	filter := func(pod corev1.Pod) bool {

lib/services/role.go

@@ -1976,6 +1976,13 @@ func matchDenyRoleImpersonateCondition(cond types.ImpersonateConditions, imperso
 	return false, nil
 }
 
+// RoleMatcherFunc is a convenience type for creating a role matcher from a function.
+type RoleMatcherFunc func(types.Role, types.RoleConditionType) (bool, error)
+
+func (f RoleMatcherFunc) Match(role types.Role, condition types.RoleConditionType) (bool, error) {
+	return f(role, condition)
+}
+
 // RoleMatcher defines an interface for a generic role matcher.
 type RoleMatcher interface {
 	Match(types.Role, types.RoleConditionType) (bool, error)