[v16] Reduce resource consumption when generating Kubernetes certificates

e2e/aws/eks_test.go

@@ -91,6 +91,7 @@ func awsEKSDiscoveryMatchedCluster(t *testing.T) {
 	)
 	// Get the auth server.
 	authC := teleport.Process.GetAuthServer()
+	expectedClusterName := os.Getenv(eksClusterNameEnv)
 	// Wait for the discovery service to discover the cluster and create a
 	// KubernetesCluster resource.
 	// Discovery service will scan the AWS account each minutes.
@@ -105,7 +106,7 @@ func awsEKSDiscoveryMatchedCluster(t *testing.T) {
 		// Fail fast if the discovery service creates more than one cluster.
 		assert.Len(t, clusters, 1)
 		// Fail fast if the discovery service creates a cluster with a different name.
-		assert.Equal(t, os.Getenv(eksClusterNameEnv), clusters[0].GetName())
+		assert.Equal(t, expectedClusterName, clusters[0].GetName())
 		return true
 	}, 3*time.Minute, 10*time.Second, "wait for the discovery service to create a cluster")
 
@@ -116,12 +117,19 @@ func awsEKSDiscoveryMatchedCluster(t *testing.T) {
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
 
-		kubeServers, err := authC.GetKubernetesServers(ctx)
-		return err == nil && len(kubeServers) == 1
-	}, 2*time.Minute, time.Second, "wait for the kubernetes service to create a KubernetesServer")
+		kubeServers, err := authC.UnifiedResourceCache.GetKubernetesServers(ctx)
+		if err != nil {
+			return false
+		}
 
-	clusters, err := authC.GetKubernetesClusters(context.Background())
-	require.NoError(t, err)
+		for _, ks := range kubeServers {
+			if ks.GetCluster().GetName() == expectedClusterName {
+				return true
+			}
+		}
+
+		return false
+	}, 2*time.Minute, time.Second, "wait for the kubernetes service to create a KubernetesServer")
 
 	// kubeClient is a Kubernetes client for the user created above
 	// that will be used to verify that the user can access the cluster and
@@ -131,7 +139,7 @@ func awsEKSDiscoveryMatchedCluster(t *testing.T) {
 		Username:    hostUser,
 		KubeUsers:   kubeUsers,
 		KubeGroups:  kubeGroups,
-		KubeCluster: clusters[0].GetName(),
+		KubeCluster: expectedClusterName,
 	})
 	require.NoError(t, err)
 

lib/auth/auth.go

@@ -99,7 +99,6 @@ import (
 	"github.com/gravitational/teleport/lib/gitlab"
 	"github.com/gravitational/teleport/lib/inventory"
 	kubetoken "github.com/gravitational/teleport/lib/kube/token"
-	kubeutils "github.com/gravitational/teleport/lib/kube/utils"
 	"github.com/gravitational/teleport/lib/limiter"
 	"github.com/gravitational/teleport/lib/loginrule"
 	"github.com/gravitational/teleport/lib/modules"
@@ -3159,9 +3158,29 @@ func generateCert(ctx context.Context, a *Server, req certRequest, caType types.
 	// If the certificate is targeting a trusted Teleport cluster, it is the
 	// responsibility of the cluster to ensure its existence.
 	if req.routeToCluster == clusterName && req.kubernetesCluster != "" {
-		if err := kubeutils.CheckKubeCluster(a.closeCtx, a, req.kubernetesCluster); err != nil {
+		found, _, err := a.UnifiedResourceCache.IterateUnifiedResources(a.closeCtx, func(rwl types.ResourceWithLabels) (bool, error) {
+			if rwl.GetKind() != types.KindKubeServer {
+				return false, nil
+			}
+
+			ks, ok := rwl.(types.KubeServer)
+			if !ok {
+				return false, nil
+			}
+
+			return ks.GetCluster().GetName() == req.kubernetesCluster, nil
+		}, &proto.ListUnifiedResourcesRequest{
+			Kinds:  []string{types.KindKubeServer},
+			SortBy: types.SortBy{Field: services.SortByName},
+			Limit:  1,
+		})
+		if err != nil {
 			return nil, trace.Wrap(err)
 		}
+
+		if len(found) == 0 {
+			return nil, trace.BadParameter("Kubernetes cluster %q is not registered in this Teleport cluster; you can list registered Kubernetes clusters using 'tsh kube ls'", req.kubernetesCluster)
+		}
 	}
 
 	// See which database names and users this user is allowed to use.

lib/auth/auth_test.go

@@ -140,12 +140,25 @@ func newTestPack(
 	}
 	p.a.SetLockWatcher(lockWatcher)
 
-	// set cluster name
-	err = p.a.SetClusterName(p.clusterName)
+	urc, err := services.NewUnifiedResourceCache(ctx, services.UnifiedResourceCacheConfig{
+		Clock: p.a.clock,
+		ResourceWatcherConfig: services.ResourceWatcherConfig{
+			Component: teleport.ComponentAuth,
+			Client:    p.a,
+		},
+		ResourceGetter: p.a,
+	})
 	if err != nil {
 		return p, trace.Wrap(err)
 	}
 
+	p.a.SetUnifiedResourcesCache(urc)
+
+	// set cluster name
+	if err := p.a.SetClusterName(p.clusterName); err != nil {
+		return p, trace.Wrap(err)
+	}
+
 	// set static tokens
 	staticTokens, err := types.NewStaticTokens(types.StaticTokensSpecV2{
 		StaticTokens: []types.ProvisionTokenV1{},
@@ -655,6 +668,22 @@ func TestAuthenticateSSHUser(t *testing.T) {
 	_, err = s.a.UpsertKubernetesServer(ctx, kubeServer)
 	require.NoError(t, err)
 
+	// Wait for cache propagation of the kubernetes resources before proceeding with the tests.
+	require.Eventually(t, func() bool {
+		kubeServers, err := s.a.UnifiedResourceCache.GetKubernetesServers(ctx)
+		if err != nil {
+			return false
+		}
+
+		for _, ks := range kubeServers {
+			if ks.GetCluster().GetName() == kubeCluster.GetName() {
+				return true
+			}
+		}
+
+		return false
+	}, 10*time.Second, 100*time.Millisecond)
+
 	// Login specifying a valid kube cluster. It should appear in the TLS cert.
 	resp, err = s.a.AuthenticateSSHUser(ctx, authclient.AuthenticateSSHRequest{
 		AuthenticateUserRequest: authclient.AuthenticateUserRequest{
@@ -2993,6 +3022,98 @@ func TestGenerateUserCertWithHardwareKeySupport(t *testing.T) {
 	}
 }
 
+func TestGenerateKubernetesUserCert(t *testing.T) {
+	ctx := context.Background()
+	p, err := newTestPack(ctx, t.TempDir())
+	require.NoError(t, err)
+
+	user, _, err := CreateUserAndRole(p.a, "test-user", []string{}, nil)
+	require.NoError(t, err)
+
+	rc, err := types.NewRemoteCluster("leaf")
+	require.NoError(t, err)
+	_, err = p.a.CreateRemoteCluster(ctx, rc)
+	require.NoError(t, err)
+
+	kubeCluster, err := types.NewKubernetesClusterV3(types.Metadata{Name: "kube-cluster"}, types.KubernetesClusterSpecV3{})
+	require.NoError(t, err)
+	kubeServer, err := types.NewKubernetesServerV3FromCluster(kubeCluster, "foo", "1")
+	require.NoError(t, err)
+	_, err = p.a.UpsertKubernetesServer(ctx, kubeServer)
+	require.NoError(t, err)
+
+	// Wait for cache propagation of the kubernetes resources before proceeding with the tests.
+	require.EventuallyWithT(t, func(t *assert.CollectT) {
+		found, _, err := p.a.UnifiedResourceCache.IterateUnifiedResources(ctx, func(rwl types.ResourceWithLabels) (bool, error) {
+			if rwl.GetKind() != types.KindKubeServer {
+				return false, nil
+			}
+
+			ks, ok := rwl.(types.KubeServer)
+			if !ok {
+				return false, nil
+			}
+
+			return ks.GetCluster().GetName() == kubeCluster.GetName(), nil
+		}, &proto.ListUnifiedResourcesRequest{
+			Kinds:  []string{types.KindKubeServer},
+			SortBy: types.SortBy{Field: services.SortByName},
+			Limit:  1,
+		})
+
+		assert.NoError(t, err)
+		assert.Len(t, found, 1)
+	}, 10*time.Second, 100*time.Millisecond)
+
+	accessInfo := services.AccessInfoFromUserState(user)
+	accessChecker, err := services.NewAccessChecker(accessInfo, p.clusterName.GetClusterName(), p.a)
+	require.NoError(t, err)
+
+	priv, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+	require.NoError(t, err)
+	key, err := keys.NewPrivateKey(priv, nil)
+	require.NoError(t, err)
+
+	for _, tt := range []struct {
+		name              string
+		teleportCluster   string
+		kubernetesCluster string
+		assertErr         require.ErrorAssertionFunc
+	}{
+		{
+			name:              "leaf clusters not validated",
+			teleportCluster:   "leaf",
+			kubernetesCluster: "foo",
+			assertErr:         require.NoError,
+		},
+		{
+			name:              "kubernetes cluster not registered",
+			teleportCluster:   p.clusterName.GetClusterName(),
+			kubernetesCluster: "foo",
+			assertErr:         require.Error,
+		},
+		{
+			name:              "kubernetes cluster registered",
+			teleportCluster:   p.clusterName.GetClusterName(),
+			kubernetesCluster: kubeCluster.GetName(),
+			assertErr:         require.NoError,
+		},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			certReq := certRequest{
+				user:              user,
+				checker:           accessChecker,
+				publicKey:         key.MarshalSSHPublicKey(),
+				routeToCluster:    tt.teleportCluster,
+				kubernetesCluster: tt.kubernetesCluster,
+			}
+
+			_, err = p.a.generateUserCert(ctx, certReq)
+			tt.assertErr(t, err)
+		})
+	}
+}
+
 func TestNewWebSession(t *testing.T) {
 	t.Parallel()
 	ctx := context.Background()

lib/kube/utils/utils.go

@@ -22,7 +22,6 @@ import (
 	"context"
 	"encoding/hex"
 	"errors"
-	"slices"
 	"strings"
 
 	"github.com/gravitational/trace"
@@ -147,48 +146,6 @@ func EncodeClusterName(clusterName string) string {
 	return "k" + hex.EncodeToString([]byte(clusterName))
 }
 
-// KubeServicesPresence fetches a list of registered kubernetes servers.
-// It's a subset of services.Presence.
-type KubeServicesPresence interface {
-	// GetKubernetesServers returns a list of registered kubernetes servers.
-	GetKubernetesServers(context.Context) ([]types.KubeServer, error)
-}
-
-// KubeClusterNames returns a sorted list of unique kubernetes cluster
-// names registered in p.
-//
-// DELETE IN 11.0.0, replaced by ListKubeClustersWithFilters
-func KubeClusterNames(ctx context.Context, p KubeServicesPresence) ([]string, error) {
-	kss, err := p.GetKubernetesServers(ctx)
-	if err != nil {
-		return nil, trace.Wrap(err)
-	}
-	return extractAndSortKubeClusterNames(kss), nil
-}
-
-func extractAndSortKubeClusterNames(kubeServers []types.KubeServer) []string {
-	kubeClusters := extractAndSortKubeClusters(kubeServers)
-	kubeClusterNames := make([]string, len(kubeClusters))
-	for i := range kubeClusters {
-		kubeClusterNames[i] = kubeClusters[i].GetName()
-	}
-
-	return kubeClusterNames
-}
-
-// KubeClusters returns a sorted list of unique kubernetes clusters
-// registered in p.
-//
-// DELETE IN 11.0.0, replaced by ListKubeClustersWithFilters
-func KubeClusters(ctx context.Context, p KubeServicesPresence) ([]types.KubeCluster, error) {
-	kubeServers, err := p.GetKubernetesServers(ctx)
-	if err != nil {
-		return nil, trace.Wrap(err)
-	}
-
-	return extractAndSortKubeClusters(kubeServers), nil
-}
-
 // ListKubeClustersWithFilters returns a sorted list of unique kubernetes clusters
 // registered in p.
 func ListKubeClustersWithFilters(ctx context.Context, p client.GetResourcesClient, req proto.ListResourcesRequest) ([]types.KubeCluster, error) {
@@ -244,19 +201,3 @@ func GetKubeAgentVersion(ctx context.Context, pinger Pinger, clusterFeatures pro
 
 	return strings.TrimPrefix(agentVersion, "v"), nil
 }
-
-// CheckKubeCluster validates kubeClusterName is registered with this Teleport cluster.
-func CheckKubeCluster(ctx context.Context, p KubeServicesPresence, kubeClusterName string) error {
-	if kubeClusterName == "" {
-		return trace.BadParameter("kube cluster name should not be empty.")
-	}
-	kubeClusterNames, err := KubeClusterNames(ctx, p)
-	if err != nil {
-		return trace.Wrap(err, "failed to get list of available Kubernetes clusters.")
-	}
-	if !slices.Contains(kubeClusterNames, kubeClusterName) {
-		return trace.BadParameter("Kubernetes cluster %q is not registered in this Teleport cluster; you can list registered Kubernetes clusters using 'tsh kube ls'", kubeClusterName)
-	}
-
-	return nil
-}