Teleport Connect Kube gateway POC

integration/proxy/proxy_helpers.go

@@ -515,7 +515,7 @@ func mustCreateKubeLocalProxyListener(t *testing.T, teleportCluster string, caCe
 	ca, err := tls.X509KeyPair(caCert, caKey)
 	require.NoError(t, err)
 
-	listener, err := alpnproxy.NewKubeListener(map[string]tls.Certificate{
+	listener, err := alpnproxy.NewKubeListenerWithRandomPort(map[string]tls.Certificate{
 		teleportCluster: ca,
 	})
 	require.NoError(t, err)
@@ -563,7 +563,7 @@ func mustStartALPNLocalProxyWithConfig(t *testing.T, config alpnproxy.LocalProxy
 func mustStartKubeForwardProxy(t *testing.T, lpAddr string) *alpnproxy.ForwardProxy {
 	t.Helper()
 
-	fp, err := alpnproxy.NewKubeForwardProxy(context.Background(), "", lpAddr)
+	fp, err := alpnproxy.NewKubeForwardProxyWithPort(context.Background(), "", lpAddr)
 	require.NoError(t, err)
 	t.Cleanup(func() {
 		fp.Close()

lib/client/profile.go

@@ -510,6 +510,15 @@ func (p *ProfileStatus) KubeConfigPath(name string) string {
 	return keypaths.KubeConfigPath(p.Dir, p.Name, p.Username, p.Cluster, name)
 }
 
+// TODO
+func (p *ProfileStatus) KubeCertPath(name string) string {
+	if path, ok := p.virtualPathFromEnv(VirtualPathKubernetes, VirtualPathKubernetesParams(name)); ok {
+		return path
+	}
+
+	return keypaths.KubeCertPath(p.Dir, p.Name, p.Username, p.Cluster, name)
+}
+
 // DatabaseServices returns a list of database service names for this profile.
 func (p *ProfileStatus) DatabaseServices() (result []string) {
 	for _, db := range p.Databases {

lib/srv/alpnproxy/kube.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"crypto/tls"
 	"crypto/x509"
+	"crypto/x509/pkix"
 	"errors"
 	"net"
 	"net/http"
@@ -37,6 +38,8 @@ import (
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/client-go/kubernetes/scheme"
 
+	"github.com/gravitational/teleport/api/utils/keys"
+	"github.com/gravitational/teleport/lib/defaults"
 	"github.com/gravitational/teleport/lib/srv/alpnproxy/common"
 	"github.com/gravitational/teleport/lib/tlsca"
 	"github.com/gravitational/teleport/lib/utils"
@@ -238,8 +241,20 @@ func (m *KubeMiddleware) reissueCertIfExpired(ctx context.Context, cert tls.Cert
 	}
 }
 
-// NewKubeListener creates a listener for kube local proxy.
-func NewKubeListener(casByTeleportCluster map[string]tls.Certificate) (net.Listener, error) {
+// NewKubeListenerWithRandomPort creates a listener for kube local proxy using a
+// random localhost port.
+func NewKubeListenerWithRandomPort(casByTeleportCluster map[string]tls.Certificate) (net.Listener, error) {
+	inner, err := net.Listen("tcp", "localhost:0")
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	listener, err := NewKubeListener(inner, casByTeleportCluster)
+	return listener, trace.Wrap(err)
+}
+
+// NewKubeListener creates a TLS listener for kube local proxy using provided
+// inner listener.
+func NewKubeListener(inner net.Listener, casByTeleportCluster map[string]tls.Certificate) (net.Listener, error) {
 	configs := make(map[string]*tls.Config)
 	for teleportCluster, ca := range casByTeleportCluster {
 		caLeaf, err := utils.TLSCertLeaf(ca)
@@ -258,20 +273,20 @@ func NewKubeListener(casByTeleportCluster map[string]tls.Certificate) (net.Liste
 			ClientCAs:    clientCAs,
 		}
 	}
-	listener, err := tls.Listen("tcp", "localhost:0", &tls.Config{
+	return tls.NewListener(inner, &tls.Config{
 		GetConfigForClient: func(hello *tls.ClientHelloInfo) (*tls.Config, error) {
 			config, ok := configs[common.TeleportClusterFromKubeLocalProxySNI(hello.ServerName)]
 			if !ok {
 				return nil, trace.BadParameter("unknown Teleport cluster or invalid TLS server name %v", hello.ServerName)
 			}
 			return config, nil
 		},
-	})
-	return listener, trace.Wrap(err)
+	}), nil
 }
 
-// NewKubeForwardProxy creates a forward proxy for kube access.
-func NewKubeForwardProxy(ctx context.Context, listenPort, forwardAddr string) (*ForwardProxy, error) {
+// NewKubeForwardProxyWithPort creates a forward proxy for kube access with
+// provided port.
+func NewKubeForwardProxyWithPort(ctx context.Context, listenPort, forwardAddr string) (*ForwardProxy, error) {
 	listenAddr := "localhost:0"
 	if listenPort != "" {
 		listenAddr = "localhost:" + listenPort
@@ -282,6 +297,13 @@ func NewKubeForwardProxy(ctx context.Context, listenPort, forwardAddr string) (*
 		return nil, trace.Wrap(err)
 	}
 
+	fp, err := NewKubeForwardProxy(ctx, listener, forwardAddr)
+	return fp, trace.Wrap(err)
+}
+
+// NewKubeForwardProxy creates a forward proxy for kube access with provided
+// net.listener.
+func NewKubeForwardProxy(ctx context.Context, listener net.Listener, forwardAddr string) (*ForwardProxy, error) {
 	fp, err := NewForwardProxy(ForwardProxyConfig{
 		Listener:     listener,
 		CloseContext: ctx,
@@ -297,3 +319,38 @@ func NewKubeForwardProxy(ctx context.Context, listenPort, forwardAddr string) (*
 	}
 	return fp, nil
 }
+
+// CreateKubeLocalCAs generate local CAs used for kube local proxy with provided key.
+func CreateKubeLocalCAs(key *keys.PrivateKey, teleportClusters []string) (map[string]tls.Certificate, error) {
+	cas := make(map[string]tls.Certificate)
+	for _, teleportCluster := range teleportClusters {
+		ca, err := createLocalCA(key, time.Now().Add(defaults.CATTL), common.KubeLocalProxyWildcardDomain(teleportCluster))
+		if err != nil {
+			return nil, trace.Wrap(err)
+		}
+		cas[teleportCluster] = ca
+	}
+	return cas, nil
+}
+
+func createLocalCA(key *keys.PrivateKey, validUntil time.Time, dnsNames ...string) (tls.Certificate, error) {
+	cert, err := tlsca.GenerateSelfSignedCAWithConfig(tlsca.GenerateCAConfig{
+		Entity: pkix.Name{
+			CommonName:   "localhost",
+			Organization: []string{"Teleport"},
+		},
+		Signer:      key,
+		DNSNames:    dnsNames,
+		IPAddresses: []net.IP{net.ParseIP(defaults.Localhost)},
+		TTL:         time.Until(validUntil),
+	})
+	if err != nil {
+		return tls.Certificate{}, trace.Wrap(err)
+	}
+
+	tlsCert, err := keys.X509KeyPair(cert, key.PrivateKeyPEM())
+	if err != nil {
+		return tls.Certificate{}, trace.Wrap(err)
+	}
+	return tlsCert, nil
+}

lib/teleterm/api/uri/uri.go

@@ -29,6 +29,8 @@ var pathServers = urlpath.New("/clusters/:cluster/servers/:serverUUID")
 var pathLeafServers = urlpath.New("/clusters/:cluster/leaves/:leaf/servers/:serverUUID")
 var pathDbs = urlpath.New("/clusters/:cluster/dbs/:dbName")
 var pathLeafDbs = urlpath.New("/clusters/:cluster/leaves/:leaf/dbs/:dbName")
+var pathKubes = urlpath.New("/clusters/:cluster/kubes/:kubeName")
+var pathLeafKubes = urlpath.New("/clusters/:cluster/leaves/:leaf/kubes/:kubeName")
 
 // New creates an instance of ResourceURI
 func New(path string) ResourceURI {
@@ -111,6 +113,21 @@ func (r ResourceURI) GetDbName() string {
 	return ""
 }
 
+// GetKubeName extracts the kube name from r. Returns an empty string if path is not a kube URI.
+func (r ResourceURI) GetKubeName() string {
+	result, ok := pathKubes.Match(r.path)
+	if ok {
+		return result.Params["kubeName"]
+	}
+
+	result, ok = pathLeafKubes.Match(r.path)
+	if ok {
+		return result.Params["kubeName"]
+	}
+
+	return ""
+}
+
 // GetServerUUID extracts the server UUID from r. Returns an empty string if path is not a server URI.
 func (r ResourceURI) GetServerUUID() string {
 	result, ok := pathServers.Match(r.path)
@@ -177,3 +194,13 @@ func (r ResourceURI) AppendAccessRequest(id string) ResourceURI {
 func (r ResourceURI) String() string {
 	return r.path
 }
+
+// TODO
+func IsDB(resourceURI string) bool {
+	return New(resourceURI).GetDbName() != ""
+}
+
+// TODO
+func IsKube(resourceURI string) bool {
+	return New(resourceURI).GetKubeName() != ""
+}

lib/teleterm/api/uri/uri_test.go

@@ -130,6 +130,52 @@ func TestGetDbName(t *testing.T) {
 	}
 }
 
+func TestGetKubeName(t *testing.T) {
+	tests := []struct {
+		name string
+		in   uri.ResourceURI
+		out  string
+	}{
+		{
+			name: "returns root cluster kube name",
+			in:   uri.NewClusterURI("foo").AppendKube("k8s"),
+			out:  "k8s",
+		},
+		{
+			name: "returns leaf cluster kube name",
+			in:   uri.NewClusterURI("foo").AppendLeafCluster("bar").AppendKube("k8s"),
+			out:  "k8s",
+		},
+		{
+			name: "returns empty string when given root cluster URI",
+			in:   uri.NewClusterURI("foo"),
+			out:  "",
+		},
+		{
+			name: "returns empty string when given leaf cluster URI",
+			in:   uri.NewClusterURI("foo").AppendLeafCluster("bar"),
+			out:  "",
+		},
+		{
+			name: "returns empty string when given root cluster non-kube resource URI",
+			in:   uri.NewClusterURI("foo").AppendDB("postgres"),
+			out:  "",
+		},
+		{
+			name: "returns empty string when given leaf cluster non-kube resource URI",
+			in:   uri.NewClusterURI("foo").AppendLeafCluster("bar").AppendDB("postgres"),
+			out:  "",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			out := tt.in.GetKubeName()
+			require.Equal(t, tt.out, out)
+		})
+	}
+}
+
 func TestGetServerUUID(t *testing.T) {
 	tests := []struct {
 		name string

lib/teleterm/clusters/cluster_gateways.go

@@ -21,6 +21,7 @@ import (
 
 	"github.com/gravitational/trace"
 
+	"github.com/gravitational/teleport/lib/teleterm/api/uri"
 	"github.com/gravitational/teleport/lib/teleterm/gateway"
 	"github.com/gravitational/teleport/lib/tlsca"
 )
@@ -42,6 +43,21 @@ type CreateGatewayParams struct {
 
 // CreateGateway creates a gateway
 func (c *Cluster) CreateGateway(ctx context.Context, params CreateGatewayParams) (*gateway.Gateway, error) {
+	switch {
+	case uri.IsDB(params.TargetURI):
+		gw, err := c.createDatabaseGateway(ctx, params)
+		return gw, trace.Wrap(err)
+
+	case uri.IsKube(params.TargetURI):
+		gw, err := c.createKubeGateway(ctx, params)
+		return gw, trace.Wrap(err)
+
+	default:
+		return nil, trace.NotImplemented("gateway not supported for %v", params.TargetURI)
+	}
+}
+
+func (c *Cluster) createDatabaseGateway(ctx context.Context, params CreateGatewayParams) (*gateway.Gateway, error) {
 	db, err := c.GetDatabase(ctx, params.TargetURI)
 	if err != nil {
 		return nil, trace.Wrap(err)
@@ -67,6 +83,7 @@ func (c *Cluster) CreateGateway(ctx context.Context, params CreateGatewayParams)
 		KeyPath:                       c.status.KeyPath(),
 		CertPath:                      c.status.DatabaseCertPathForCluster(c.clusterClient.SiteName, db.GetName()),
 		Insecure:                      c.clusterClient.InsecureSkipVerify,
+		ClusterName:                   c.Name,
 		WebProxyAddr:                  c.clusterClient.WebProxyAddr,
 		Log:                           c.Log,
 		CLICommandProvider:            params.CLICommandProvider,
@@ -82,3 +99,34 @@ func (c *Cluster) CreateGateway(ctx context.Context, params CreateGatewayParams)
 
 	return gw, nil
 }
+
+func (c *Cluster) createKubeGateway(ctx context.Context, params CreateGatewayParams) (*gateway.Gateway, error) {
+	kubeCluster := uri.New(params.TargetURI).GetKubeName()
+	if kubeCluster == "" {
+		return nil, trace.BadParameter("invalid TargetURI %v for Kube gateway", params.TargetURI)
+	}
+
+	if err := c.ReissueKubeCerts(ctx, kubeCluster); err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	// TODO support TargetUser (--as), TargetGroups (--as-groups), TargetSubresourceName (--kube-namespace)
+	gw, err := gateway.New(gateway.Config{
+		LocalPort:                     params.LocalPort,
+		TargetURI:                     params.TargetURI,
+		TargetName:                    kubeCluster,
+		KeyPath:                       c.status.KeyPath(),
+		CertPath:                      c.status.KubeCertPath(kubeCluster),
+		Insecure:                      c.clusterClient.InsecureSkipVerify,
+		ClusterName:                   c.Name,
+		WebProxyAddr:                  c.clusterClient.WebProxyAddr,
+		Log:                           c.Log,
+		CLICommandProvider:            params.CLICommandProvider,
+		TCPPortAllocator:              params.TCPPortAllocator,
+		OnExpiredCert:                 params.OnExpiredCert,
+		Clock:                         c.clock,
+		TLSRoutingConnUpgradeRequired: c.clusterClient.TLSRoutingConnUpgradeRequired,
+		RootClusterCACertPoolFunc:     c.clusterClient.RootClusterCACertPool,
+	})
+	return gw, trace.Wrap(err)
+}

lib/teleterm/clusters/cluster_kubes.go

@@ -98,6 +98,32 @@ func (c *Cluster) GetKubes(ctx context.Context, r *api.GetKubesRequest) (*GetKub
 	}, nil
 }
 
+// TODO
+func (c *Cluster) ReissueKubeCerts(ctx context.Context, kubeCluster string) error {
+	return trace.Wrap(addMetadataToRetryableError(ctx, func() error {
+		// Refresh the certs to account for clusterClient.SiteName pointing at a leaf cluster.
+		err := c.clusterClient.ReissueUserCerts(ctx, client.CertCacheKeep, client.ReissueParams{
+			RouteToCluster: c.clusterClient.SiteName,
+			AccessRequests: c.status.ActiveRequests.AccessRequests,
+		})
+		if err != nil {
+			return trace.Wrap(err)
+		}
+
+		// Fetch the certs for the database.
+		err = c.clusterClient.ReissueUserCerts(ctx, client.CertCacheKeep, client.ReissueParams{
+			RouteToCluster:    c.clusterClient.SiteName,
+			KubernetesCluster: kubeCluster,
+			AccessRequests:    c.status.ActiveRequests.AccessRequests,
+		})
+		if err != nil {
+			return trace.Wrap(err)
+		}
+
+		return nil
+	}))
+}
+
 type GetKubesResponse struct {
 	Kubes []Kube
 	// StartKey is the next key to use as a starting point.

lib/teleterm/clusters/kube_cli_command_provider.go

@@ -0,0 +1,34 @@
+// 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 clusters
+
+import (
+	"fmt"
+	"os/exec"
+
+	"github.com/gravitational/teleport"
+	"github.com/gravitational/teleport/lib/teleterm/gateway"
+)
+
+// TODO
+type KubeCLICommandProvider struct {
+}
+
+func (p KubeCLICommandProvider) GetCommand(gateway *gateway.Gateway) (*exec.Cmd, error) {
+	// Use kubectl version as placeholders. Only env should be used.
+	cmd := exec.Command("kubectl", "version")
+	cmd.Env = []string{fmt.Sprintf("%v=%v", teleport.EnvKubeConfig, gateway.KubeconfigPath())}
+	return cmd, nil
+}