Add cluster_name to proxy kubernetes config

Cluster name from this field plug all clusters from kubeconfig are
stored on the auth server via heartbeats.
This info will later be used to route k8s requests back to proxies.

Updates #3952

lib/config/configuration.go

@@ -537,6 +537,9 @@ func applyProxyConfig(fc *FileConfig, cfg *service.Config) error {
 	if fc.Proxy.Kube.KubeconfigFile != "" {
 		cfg.Proxy.Kube.KubeconfigPath = fc.Proxy.Kube.KubeconfigFile
 	}
+	if fc.Proxy.Kube.ClusterName != "" {
+		cfg.Proxy.Kube.ClusterName = fc.Proxy.Kube.ClusterName
+	}
 	if fc.Proxy.Kube.ListenAddress != "" {
 		addr, err := utils.ParseHostPortAddr(fc.Proxy.Kube.ListenAddress, int(defaults.KubeProxyListenPort))
 		if err != nil {

lib/config/fileconf.go

@@ -829,6 +829,9 @@ type Kube struct {
 	// if specified, teleport will use API server address and
 	// trusted certificate authority information from it
 	KubeconfigFile string `yaml:"kubeconfig_file,omitempty"`
+	// ClusterName is the name of a kubernetes cluster this proxy is running
+	// in. If set, this proxy will handle kubernetes requests for the cluster.
+	ClusterName string `yaml:"cluster_name,omitempty"`
 }
 
 // ReverseTunnel is a SSH reverse tunnel maintained by one cluster's

lib/service/cfg.go

@@ -21,17 +21,20 @@ import (
 	"io"
 	"os"
 	"path/filepath"
+	"sort"
 	"time"
 
 	"golang.org/x/crypto/ssh"
 
+	"github.com/gravitational/teleport"
 	"github.com/gravitational/teleport/lib/auth"
 	"github.com/gravitational/teleport/lib/backend"
 	"github.com/gravitational/teleport/lib/backend/lite"
 	"github.com/gravitational/teleport/lib/backend/memory"
 	"github.com/gravitational/teleport/lib/bpf"
 	"github.com/gravitational/teleport/lib/defaults"
 	"github.com/gravitational/teleport/lib/events"
+	"github.com/gravitational/teleport/lib/kube/kubeconfig"
 	"github.com/gravitational/teleport/lib/limiter"
 	"github.com/gravitational/teleport/lib/pam"
 	"github.com/gravitational/teleport/lib/services"
@@ -368,6 +371,71 @@ type KubeProxyConfig struct {
 
 	// KubeconfigPath is a path to kubeconfig
 	KubeconfigPath string
+
+	// ClusterName is the name of a kubernetes cluster this proxy is running
+	// in. If set, this proxy will handle kubernetes requests for the cluster.
+	ClusterName string
+
+	// runningInPod reports whether the current process is running inside of a
+	// Kubernetes Pod. If nil, checks for the presence of on-disk service
+	// account credentials in standard paths.
+	//
+	// Used for test mocking.
+	runningInPod func() bool
+}
+
+// ClusterNames returns the complete list of kubernetes clusters from
+// ClusterName and kubeconfig.
+func (c KubeProxyConfig) ClusterNames(teleportClusterName string) ([]string, error) {
+	if !c.Enabled {
+		return nil, nil
+	}
+
+	clusters := make(map[string]struct{})
+	if c.ClusterName != "" {
+		clusters[c.ClusterName] = struct{}{}
+	} else {
+		// If a service account CA bundle exists at the standard path, we must
+		// be running inside of a Kubernetes pod. Since we don't have a
+		// ClusterName specified, use teleport cluster name as a fallback.
+		if c.runningInPod == nil {
+			c.runningInPod = func() bool {
+				_, err := os.Stat(teleport.KubeCAPath)
+				return err == nil
+			}
+		}
+		if c.runningInPod() {
+			clusters[teleportClusterName] = struct{}{}
+		}
+	}
+	if c.KubeconfigPath != "" {
+		cfg, err := kubeconfig.Load(c.KubeconfigPath)
+		if err != nil {
+			return nil, trace.Wrap(err, "failed parsing kubeconfig file %q", c.KubeconfigPath)
+		}
+		for n := range cfg.Contexts {
+			if n != "" {
+				clusters[n] = struct{}{}
+			}
+		}
+		if cfg.CurrentContext != "" {
+			// DELETE IN 5.2
+			// Use teleport cluster name as an alias for the k8s cluster in
+			// current-context.
+			//
+			// This enables backwards-compatibility: when an older client
+			// requests a k8s cert without specifying a k8s cluster, we want to
+			// preserve pre-5.0 behavior. That is, when no specific cluster is
+			// requested, use current-context.
+			clusters[teleportClusterName] = struct{}{}
+		}
+	}
+	var uniqClusters []string
+	for n := range clusters {
+		uniqClusters = append(uniqClusters, n)
+	}
+	sort.Strings(uniqClusters)
+	return uniqClusters, nil
 }
 
 // AuthConfig is a configuration of the auth server

lib/service/cfg_test.go

@@ -17,12 +17,14 @@ limitations under the License.
 package service
 
 import (
+	"io/ioutil"
 	"path/filepath"
 	"testing"
 
 	"github.com/gravitational/teleport/lib/backend/lite"
 	"github.com/gravitational/teleport/lib/defaults"
 	"github.com/gravitational/teleport/lib/utils"
+	"github.com/stretchr/testify/assert"
 
 	"gopkg.in/check.v1"
 )
@@ -98,3 +100,107 @@ func (s *ConfigSuite) TestDefaultConfig(c *check.C) {
 	c.Assert(proxy.Limiter.MaxConnections, check.Equals, int64(defaults.LimiterMaxConnections))
 	c.Assert(proxy.Limiter.MaxNumberOfUsers, check.Equals, defaults.LimiterMaxConcurrentUsers)
 }
+
+func TestKubeClusterNames(t *testing.T) {
+	t.Parallel()
+
+	kubeconfigFile, err := ioutil.TempFile("", "teleport")
+	assert.NoError(t, err)
+	kubeconfigPath := kubeconfigFile.Name()
+	_, err = kubeconfigFile.Write([]byte(`
+apiVersion: v1
+kind: Config
+preferences: {}
+clusters:
+- cluster:
+    server: https://localhost:1
+  name: kubeconfig-cluster-1
+- cluster:
+    server: https://localhost:2
+  name: kubeconfig-cluster-2
+contexts:
+- context:
+    cluster: kubeconfig-cluster-1
+    user: user
+  name: kubeconfig-cluster-1
+- context:
+    cluster: kubeconfig-cluster-2
+    user: user
+  name: kubeconfig-cluster-2
+current-context: "kubeconfig-cluster-1"
+users:
+- name: user
+  user:
+`))
+	assert.NoError(t, err)
+	assert.NoError(t, kubeconfigFile.Close())
+
+	tests := []struct {
+		desc string
+		cfg  KubeProxyConfig
+		want []string
+	}{
+		{
+			desc: "no ClusterName, Kubeconfig, not running in a pod",
+			cfg: KubeProxyConfig{
+				Enabled:      true,
+				runningInPod: func() bool { return false },
+			},
+			want: nil,
+		},
+		{
+			desc: "only ClusterName set",
+			cfg: KubeProxyConfig{
+				Enabled:      true,
+				ClusterName:  "foo",
+				runningInPod: func() bool { return false },
+			},
+			want: []string{"foo"},
+		},
+		{
+			desc: "only Kubeconfig set",
+			cfg: KubeProxyConfig{
+				Enabled:        true,
+				KubeconfigPath: kubeconfigPath,
+				runningInPod:   func() bool { return false },
+			},
+			want: []string{"kubeconfig-cluster-1", "kubeconfig-cluster-2", "teleport-cluster-name"},
+		},
+		{
+			desc: "no ClusterName and Kubeconfig, running in a pod",
+			cfg: KubeProxyConfig{
+				Enabled:      true,
+				runningInPod: func() bool { return true },
+			},
+			want: []string{"teleport-cluster-name"},
+		},
+		{
+			desc: "ClusterName, Kubeconfig set and running in a pod",
+			cfg: KubeProxyConfig{
+				Enabled:        true,
+				ClusterName:    "foo",
+				KubeconfigPath: kubeconfigPath,
+				runningInPod:   func() bool { return true },
+			},
+			want: []string{"foo", "kubeconfig-cluster-1", "kubeconfig-cluster-2", "teleport-cluster-name"},
+		},
+		{
+			desc: "Kubernetes support not enabled",
+			cfg: KubeProxyConfig{
+				Enabled:        false,
+				ClusterName:    "foo",
+				KubeconfigPath: kubeconfigPath,
+				runningInPod:   func() bool { return true },
+			},
+			want: nil,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.desc, func(t *testing.T) {
+			got, err := tt.cfg.ClusterNames("teleport-cluster-name")
+			assert.NoError(t, err)
+			assert.EqualValues(t, tt.want, got)
+		})
+	}
+}

lib/service/service.go

@@ -2339,6 +2339,10 @@ func (process *TeleportProcess) initProxyEndpoint(conn *Connector) error {
 	if err != nil {
 		return trace.Wrap(err)
 	}
+	kubeClusterNames, err := cfg.Proxy.Kube.ClusterNames(conn.ServerIdentity.Cert.Extensions[utils.CertExtensionAuthority])
+	if err != nil {
+		return trace.Wrap(err)
+	}
 	sshProxy, err := regular.New(cfg.Proxy.SSHAddr,
 		cfg.Hostname,
 		[]ssh.Signer{conn.ServerIdentity.KeySigner},
@@ -2363,6 +2367,7 @@ func (process *TeleportProcess) initProxyEndpoint(conn *Connector) error {
 			}
 		}),
 		regular.SetEmitter(&events.StreamerAndEmitter{Emitter: emitter, Streamer: streamer}),
+		regular.SetKubernetesClusters(kubeClusterNames),
 	)
 	if err != nil {
 		return trace.Wrap(err)

lib/services/server.go

@@ -57,6 +57,9 @@ type Server interface {
 	SetRotation(Rotation)
 	// GetUseTunnel gets if a reverse tunnel should be used to connect to this node.
 	GetUseTunnel() bool
+	// GetKubernetesClusters gets kubernetes clusters accessible through this
+	// proxy.
+	GetKubernetesClusters() []string
 	// SetUseTunnel sets if a reverse tunnel should be used to connect to this node.
 	SetUseTunnel(bool)
 	// String returns string representation of the server
@@ -67,6 +70,9 @@ type Server interface {
 	SetPublicAddr(string)
 	// SetNamespace sets server namespace
 	SetNamespace(namespace string)
+	// SetKubernetesClusters sets kubernetes clusters accessible through this
+	// proxy.
+	SetKubernetesClusters([]string)
 	// V1 returns V1 version for backwards compatibility
 	V1() *ServerV1
 	// MatchAgainst takes a map of labels and returns True if this server
@@ -271,6 +277,18 @@ func (s *ServerV2) GetAllLabels() map[string]string {
 	return lmap
 }
 
+// GetKubernetesClusters gets kubernetes clusters accessible through this
+// proxy.
+func (s *ServerV2) GetKubernetesClusters() []string {
+	return s.Spec.KubernetesClusters
+}
+
+// SetKubernetesClusters sets kubernetes clusters accessible through this
+// proxy.
+func (s *ServerV2) SetKubernetesClusters(clusters []string) {
+	s.Spec.KubernetesClusters = clusters
+}
+
 // MatchAgainst takes a map of labels and returns True if this server
 // has ALL of them
 //
@@ -313,6 +331,10 @@ func (s *ServerV2) CheckAndSetDefaults() error {
 		}
 	}
 
+	if len(s.Spec.KubernetesClusters) > 0 && s.Kind != KindProxy {
+		return trace.BadParameter("KubernetesClusters are only allowed on Proxy servers; got clusters %q set on a %q server %q", s.Spec.KubernetesClusters, s.Kind, s.Metadata.Name)
+	}
+
 	return nil
 }
 
@@ -362,6 +384,9 @@ func CompareServers(a, b Server) int {
 	if a.GetTeleportVersion() != b.GetTeleportVersion() {
 		return Different
 	}
+	if !utils.StringSlicesEqual(a.GetKubernetesClusters(), b.GetKubernetesClusters()) {
+		return Different
+	}
 	return Equal
 }
 
@@ -416,7 +441,8 @@ const ServerSpecV2Schema = `{
         }
       }
     },
-    "rotation": %v
+    "rotation": %v,
+    "kubernetes_clusters": {"type": "array", "items": {"type": "string"}}
   }
 }`
 

lib/services/servers_test.go

@@ -50,9 +50,10 @@ func (s *ServerSuite) TestServersCompare(c *check.C) {
 			Labels:    map[string]string{"a": "b"},
 		},
 		Spec: ServerSpecV2{
-			Addr:      "localhost:3022",
-			CmdLabels: map[string]CommandLabelV2{"a": CommandLabelV2{Period: Duration(time.Minute), Command: []string{"ls", "-l"}}},
-			Version:   "4.0.0",
+			Addr:               "localhost:3022",
+			CmdLabels:          map[string]CommandLabelV2{"a": CommandLabelV2{Period: Duration(time.Minute), Command: []string{"ls", "-l"}}},
+			Version:            "4.0.0",
+			KubernetesClusters: []string{"kube-a"},
 		},
 	}
 	node.SetExpiry(time.Date(2018, 1, 2, 3, 4, 5, 6, time.UTC))
@@ -110,6 +111,11 @@ func (s *ServerSuite) TestServersCompare(c *check.C) {
 		},
 	}
 	c.Assert(CompareServers(node, &node2), check.Equals, Different)
+
+	// KubernetesServers have changed
+	node2 = *node
+	node2.Spec.KubernetesClusters = []string{"kube-a", "kube-b"}
+	c.Assert(CompareServers(node, &node2), check.Equals, Different)
 }
 
 // TestGuessProxyHostAndVersion checks that the GuessProxyHostAndVersion